fix(gateway): bind channel completion dispatch (#126113)

* fix(gateway): bind channel completion dispatch

* fix(plugin-sdk): scope harness completion dispatch

* test(agents): cover scoped completion dispatch

extensions/qa-channel/src/gateway.ts

@@ -44,6 +44,7 @@ export async function startQaGatewayAccount(
       config: ctx.cfg as CoreConfig,
       message,
       buildContext,
+      ...(channelRuntime ? { channelRuntime } : {}),
     });
   const captureInboundError = (error: unknown) => {
     inboundError ??= error instanceof Error ? error : new Error(String(error));

extensions/qa-channel/src/inbound.ts

@@ -22,6 +22,7 @@ import {
   type QaBusMessage,
 } from "./bus-client.js";
 import { sendQaChannelMediaBatch } from "./outbound.js";
+import type { PluginRuntime } from "./runtime-api.js";
 import { getQaChannelRuntime } from "./runtime.js";
 import type { CoreConfig, ResolvedQaChannelAccount } from "./types.js";
 
@@ -266,8 +267,9 @@ export async function handleQaInbound(params: {
   config: CoreConfig;
   message: QaBusMessage;
   buildContext?: typeof buildChannelInboundEventContext;
+  channelRuntime?: PluginRuntime["channel"];
 }) {
-  const runtime = getQaChannelRuntime();
+  const channelRuntime = params.channelRuntime ?? getQaChannelRuntime().channel;
   const inbound = params.message;
   const target = buildQaTarget({
     chatType: inbound.conversation.kind,
@@ -297,12 +299,9 @@ export async function handleQaInbound(params: {
   });
   const isGroup = inbound.conversation.kind !== "direct";
   const wasMentioned = isGroup
-    ? runtime.channel.mentions.matchesMentionPatterns(
+    ? channelRuntime.mentions.matchesMentionPatterns(
         inbound.text,
-        runtime.channel.mentions.buildMentionRegexes(
-          params.config as OpenClawConfig,
-          route.agentId,
-        ),
+        channelRuntime.mentions.buildMentionRegexes(params.config as OpenClawConfig, route.agentId),
       )
     : undefined;
   const groupConfig = isGroup
@@ -422,7 +421,7 @@ export async function handleQaInbound(params: {
     },
   });
 
-  await runtime.channel.inbound.dispatch({
+  await channelRuntime.inbound.dispatch({
     cfg: params.config as OpenClawConfig,
     channel: params.channelId,
     accountId: params.account.accountId,

src/agents/embedded-agent-runner/run/run-attempt-dispatch.ts

@@ -1,3 +1,4 @@
+import { getGatewayContextResolver } from "../../../plugins/runtime/gateway-request-scope.js";
 import { createAgentHarnessTaskRuntimeScope } from "../../../tasks/agent-harness-task-runtime-scope.js";
 import type { ToolOutcomeObserver } from "../../agent-tools.before-tool-call.js";
 import type { AuthProfileStore } from "../../auth-profiles.js";
@@ -381,6 +382,7 @@ export async function dispatchEmbeddedRunAttempt(input: {
       ? {
           agentHarnessTaskRuntimeScope: createAgentHarnessTaskRuntimeScope({
             requesterSessionKey: params.sessionKey,
+            gatewayContextResolver: getGatewayContextResolver(params.admittedRunContext),
           }),
         }
       : {}),

src/agents/subagents/announce/subagent-announce-delivery.runtime.test.ts

@@ -5,6 +5,7 @@ import type {
   GatewayRequestContext,
   GatewayRequestHandlers,
 } from "../../../gateway/server-methods/types.js";
+import { withPluginRuntimeGatewayContextResolver } from "../../../plugins/runtime/gateway-request-scope.js";
 import { dispatchSubagentAnnounceAgent } from "./subagent-announce-delivery.runtime.js";
 
 function createContext(handlers: GatewayRequestHandlers): GatewayRequestContext {
@@ -59,4 +60,37 @@ describe("subagent announce Gateway instance dispatch", () => {
       ),
     ).resolves.toEqual({ runId: "announce-run", status: "ok", summary: "delivered" });
   });
+
+  it("delivers through a lifecycle-fenced instance resolver scope", async () => {
+    const context = createContext({
+      agent: ({ respond }) => respond(true, { raw: true }),
+    });
+    const idempotencyKey = "scoped-subagent-announce";
+    context.dedupe.set(`agent:${idempotencyKey}`, {
+      ts: Date.now(),
+      ok: true,
+      payload: { runId: "scoped-announce-run", status: "ok", summary: "delivered" },
+    });
+
+    await expect(
+      withPluginRuntimeGatewayContextResolver(
+        () => context,
+        () =>
+          dispatchSubagentAnnounceAgent(
+            {
+              message: "Process one completed child result.",
+              idempotencyKey,
+            },
+            {
+              expectFinal: true,
+              forceSyntheticClient: true,
+            },
+          ),
+      ),
+    ).resolves.toEqual({
+      runId: "scoped-announce-run",
+      status: "ok",
+      summary: "delivered",
+    });
+  });
 });

src/auto-reply/dispatch.ts

@@ -23,7 +23,10 @@ import {
 import { withReplyDispatcher } from "./dispatch-dispatcher.js";
 import type { CommandSessionMetadataChange } from "./reply/command-session-metadata.js";
 import { dispatchReplyFromConfig } from "./reply/dispatch-from-config.js";
-import type { DispatchFromConfigResult } from "./reply/dispatch-from-config.types.js";
+import type {
+  DispatchFromConfigResult,
+  DispatchReplyFromConfig,
+} from "./reply/dispatch-from-config.types.js";
 import type {
   InternalGetReplyFromConfig,
   InternalGetReplyOptions,
@@ -200,6 +203,7 @@ export async function dispatchInboundMessage(params: {
   toolsAllow?: string[];
   replyOptions?: InternalDispatchReplyOptions;
   replyResolver?: InternalGetReplyFromConfig;
+  dispatchReplyFromConfig?: DispatchReplyFromConfig;
   onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void;
   replyPayloadRunState?: ReplyPayloadRunState;
   /** Observe-only turns run the agent without entering outbound hook stages. */
@@ -240,7 +244,7 @@ export async function dispatchInboundMessage(params: {
       measureDiagnosticsTimelineSpan(
         "auto_reply.dispatch_reply_from_config",
         () =>
-          dispatchReplyFromConfig({
+          (params.dispatchReplyFromConfig ?? dispatchReplyFromConfig)({
             ctx: finalized,
             cfg: params.cfg,
             dispatcher: params.dispatcher,
@@ -269,6 +273,7 @@ type BufferedInboundDispatcherParams = {
   toolsAllow?: string[];
   replyOptions?: InternalDispatchReplyOptions;
   replyResolver?: InternalGetReplyFromConfig;
+  dispatchReplyFromConfig?: DispatchReplyFromConfig;
   onSessionMetadataChanges?: (changes: CommandSessionMetadataChange[]) => void;
 };
 
@@ -348,6 +353,7 @@ async function dispatchInboundMessageWithBufferedDispatcherCore(
       dispatcher,
       toolsAllow: params.toolsAllow,
       replyResolver: params.replyResolver,
+      dispatchReplyFromConfig: params.dispatchReplyFromConfig,
       replyOptions: {
         ...params.replyOptions,
         ...replyOptions,

src/auto-reply/reply/agent-runner-execution.ts

@@ -39,7 +39,10 @@ import { formatErrorMessage } from "../../infra/errors.js";
 import { recordMessageToolRunOutcome } from "../../infra/message-tool-run-outcome-store.js";
 import { logSessionTurnCreated } from "../../logging/diagnostic.js";
 import { createSubsystemLogger } from "../../logging/subsystem.js";
-import { bindGatewayContextResolver } from "../../plugins/runtime/gateway-request-scope.js";
+import {
+  bindGatewayContextResolver,
+  getPluginRuntimeGatewayRequestScope,
+} from "../../plugins/runtime/gateway-request-scope.js";
 import { isInternalMessageChannel } from "../../utils/message-channel.js";
 import type { ReplyPayload } from "../types.js";
 import {
@@ -527,7 +530,9 @@ async function executeAgentTurnInternal(
   };
   const runId = params.opts?.runId ?? crypto.randomUUID();
   const admittedRunContext: { current?: AdmittedRunContext } = {};
-  const gatewayContextResolver = readChannelContextGatewayContextResolver(params.sessionCtx);
+  const gatewayContextResolver =
+    readChannelContextGatewayContextResolver(params.sessionCtx) ??
+    getPluginRuntimeGatewayRequestScope()?.resolveGatewayContext;
   const preparedRunAdmission = prepareChannelRunAdmission({
     cfg: resolveQueuedReplyRuntimeConfig(params.followupRun.run.config),
     runId,

src/channels/turn/lifecycle.ts

@@ -625,6 +625,7 @@ async function dispatchChannelTurnWithDeliveryOwner(
                   },
                   onError: delivery.onError,
                 },
+                dispatchReplyFromConfig: params.dispatchReplyFromConfig,
                 toolsAllow: params.toolsAllow,
                 replyOptions,
                 replyResolver: params.replyResolver,

src/channels/turn/types.ts

@@ -4,7 +4,10 @@ import type {
   TurnAdoptionLifecycle,
 } from "../../auto-reply/get-reply-options.types.js";
 import type { ReplyPayload } from "../../auto-reply/reply-payload.js";
-import type { DispatchFromConfigResult } from "../../auto-reply/reply/dispatch-from-config.types.js";
+import type {
+  DispatchFromConfigResult,
+  DispatchReplyFromConfig,
+} from "../../auto-reply/reply/dispatch-from-config.types.js";
 import type { GetReplyFromConfig } from "../../auto-reply/reply/get-reply.types.js";
 import type { HistoryEntry, HistoryMediaEntry } from "../../auto-reply/reply/history.types.js";
 import type { DispatchReplyWithBufferedBlockDispatcher } from "../../auto-reply/reply/provider-dispatcher.types.js";
@@ -328,6 +331,8 @@ export type AssembledChannelTurn = {
   toolsAllow?: string[];
   replyOptions?: ChannelTurnReplyOptions;
   replyResolver?: GetReplyFromConfig;
+  /** Instance-bound reply dispatcher supplied by the owning plugin runtime. */
+  dispatchReplyFromConfig?: DispatchReplyFromConfig;
   sessionInitRetry?: {
     delaysMs: readonly number[];
     signal?: AbortSignal;

src/gateway/server-plugin-in-process-dispatch.ts

@@ -68,7 +68,8 @@ function resolveInProcessGatewayDispatch(
   options?: DispatchGatewayMethodInProcessOptions,
 ): ResolvedInProcessGatewayDispatch {
   const scope = getPluginRuntimeGatewayRequestScope();
-  const context = scope?.context ?? options?.resolveGatewayContext?.();
+  const context =
+    options?.resolveGatewayContext?.() ?? scope?.resolveGatewayContext?.() ?? scope?.context;
   const isWebchatConnect = scope?.isWebchatConnect ?? (() => false);
   if (!context) {
     throw new Error(
@@ -188,7 +189,8 @@ export async function dispatchGatewayMethodInProcessRaw(
 export function getInProcessGatewayRequestContext(
   resolveGatewayContext?: GatewayContextResolver,
 ): GatewayRequestContext | undefined {
-  return getPluginRuntimeGatewayRequestScope()?.context ?? resolveGatewayContext?.();
+  const scope = getPluginRuntimeGatewayRequestScope();
+  return resolveGatewayContext?.() ?? scope?.resolveGatewayContext?.() ?? scope?.context;
 }
 
 export async function dispatchGatewayMethodInProcess<T>(

src/gateway/server-plugins-node-runtime.ts

@@ -5,7 +5,8 @@ import type { GatewayContextResolver, GatewayRequestContext } from "./server-met
 export function hasInProcessGatewayContext(
   resolveGatewayContext?: GatewayContextResolver,
 ): boolean {
-  return Boolean(getPluginRuntimeGatewayRequestScope()?.context ?? resolveGatewayContext?.());
+  const scope = getPluginRuntimeGatewayRequestScope();
+  return Boolean(resolveGatewayContext?.() ?? scope?.resolveGatewayContext?.() ?? scope?.context);
 }
 
 export function projectGatewayRuntimeNodes(

src/gateway/server-plugins.ts

@@ -20,6 +20,7 @@ import { getActivePluginRegistry } from "../plugins/runtime.js";
 import {
   bindGatewayContextResolver,
   getPluginRuntimeGatewayRequestScope,
+  withPluginRuntimeGatewayContextResolver,
 } from "../plugins/runtime/gateway-request-scope.js";
 import { createPluginRuntimeLoaderLogger } from "../plugins/runtime/load-context.js";
 import { resolvePluginSubagentCompletionRequester } from "../plugins/runtime/subagent-requester-context.js";
@@ -461,10 +462,14 @@ function createGatewayPluginRuntimeBindings(
         const sessionWorkerPlacementContext = getInProcessGatewayRequestContext(
           resolveBoundGatewayContext,
         );
-        return await dispatchReplyFromConfig({
-          ...params,
-          ...(sessionWorkerPlacementContext ? { sessionWorkerPlacementContext } : {}),
-        });
+        return await withPluginRuntimeGatewayContextResolver(
+          resolveBoundGatewayContext,
+          async () =>
+            await dispatchReplyFromConfig({
+              ...params,
+              ...(sessionWorkerPlacementContext ? { sessionWorkerPlacementContext } : {}),
+            }),
+        );
       },
       gateway: {
         isAvailable: async () => hasInProcessGatewayContext(resolveBoundGatewayContext),

src/plugin-sdk/agent-harness-task-runtime.test.ts

@@ -3,6 +3,7 @@
  */
 import { beforeEach, describe, expect, it, vi } from "vitest";
 import { deliverSubagentAnnouncement } from "../agents/subagents/announce/subagent-announce-delivery.js";
+import { getPluginRuntimeGatewayRequestScope } from "../plugins/runtime/gateway-request-scope.js";
 import { createAgentHarnessTaskRuntimeScope } from "../tasks/agent-harness-task-runtime-scope.js";
 import { createRunningTaskRun, finalizeTaskRunByRunId } from "../tasks/detached-task-runtime.js";
 import { listTaskRecords } from "../tasks/runtime-internal.js";
@@ -166,8 +167,18 @@ describe("agent-harness-task-runtime", () => {
   });
 
   it("delivers a generic harness completion through subagent announcement delivery", async () => {
+    const gatewayContextResolver = vi.fn();
+    vi.mocked(deliverSubagentAnnouncement).mockImplementationOnce(async () => {
+      expect(getPluginRuntimeGatewayRequestScope()?.resolveGatewayContext).toBe(
+        gatewayContextResolver,
+      );
+      return { delivered: true, path: "steered" };
+    });
     await deliverAgentHarnessTaskCompletion({
-      scope: createScope("agent:main:main"),
+      scope: createAgentHarnessTaskRuntimeScope({
+        requesterSessionKey: "agent:main:main",
+        gatewayContextResolver,
+      }),
       childSessionKey: "harness-thread:child",
       childSessionId: "child",
       announceId: "harness:parent:child:succeeded",
@@ -188,6 +199,9 @@ describe("agent-harness-task-runtime", () => {
         directIdempotencyKey: "announce:harness:parent:child:succeeded",
       }),
     );
+    expect(vi.mocked(deliverSubagentAnnouncement).mock.calls[0]?.[0]).not.toHaveProperty(
+      "resolveGatewayContext",
+    );
   });
 
   it("checks durable direct delivery phases", () => {

src/plugin-sdk/agent-harness-task-runtime.ts

@@ -20,6 +20,10 @@ import {
   resolveAnnounceOrigin,
   resolveSubagentCompletionOrigin,
 } from "../agents/subagents/announce/subagent-announce-origin.js";
+import {
+  getGatewayContextResolver,
+  withPluginRuntimeGatewayContextResolver,
+} from "../plugins/runtime/gateway-request-scope.js";
 import {
   assertAgentHarnessTaskRuntimeScope,
   type AgentHarnessTaskRuntimeScope,
@@ -233,27 +237,32 @@ export async function deliverAgentHarnessTaskCompletion(params: {
     },
   ];
   const prompt = formatAgentInternalEventsForPrompt(internalEvents);
-  return await deliverSubagentAnnouncement({
-    requesterSessionKey,
-    announceId: params.announceId,
-    triggerMessage: prompt,
-    steerMessage: prompt,
-    internalEvents,
-    summaryLine: taskLabel,
-    requesterSessionOrigin: scope.requesterOrigin,
-    requesterOrigin: completionDirectOrigin ?? directOrigin,
-    completionDirectOrigin: completionDirectOrigin ?? directOrigin,
-    directOrigin,
-    sourceSessionKey: childSessionKey,
-    sourceChannel: INTERNAL_MESSAGE_CHANNEL,
-    sourceTool: AGENT_HARNESS_COMPLETION_SOURCE_TOOL,
-    targetRequesterSessionKey: requesterSessionKey,
-    requesterIsSubagent,
-    expectsCompletionMessage: true,
-    bestEffortDeliver: true,
-    directIdempotencyKey: buildAnnounceIdempotencyKey(params.announceId),
-    signal: params.signal,
-  });
+  const deliver = () =>
+    deliverSubagentAnnouncement({
+      requesterSessionKey,
+      announceId: params.announceId,
+      triggerMessage: prompt,
+      steerMessage: prompt,
+      internalEvents,
+      summaryLine: taskLabel,
+      requesterSessionOrigin: scope.requesterOrigin,
+      requesterOrigin: completionDirectOrigin ?? directOrigin,
+      completionDirectOrigin: completionDirectOrigin ?? directOrigin,
+      directOrigin,
+      sourceSessionKey: childSessionKey,
+      sourceChannel: INTERNAL_MESSAGE_CHANNEL,
+      sourceTool: AGENT_HARNESS_COMPLETION_SOURCE_TOOL,
+      targetRequesterSessionKey: requesterSessionKey,
+      requesterIsSubagent,
+      expectsCompletionMessage: true,
+      bestEffortDeliver: true,
+      directIdempotencyKey: buildAnnounceIdempotencyKey(params.announceId),
+      signal: params.signal,
+    });
+  const resolveGatewayContext = getGatewayContextResolver(scope);
+  return resolveGatewayContext
+    ? await withPluginRuntimeGatewayContextResolver(resolveGatewayContext, deliver)
+    : await deliver();
 }
 
 function mapHarnessCompletionStatus(

src/plugins/runtime/gateway-request-scope.ts

@@ -11,6 +11,7 @@ import type { PluginRegistry } from "../registry-types.js";
 
 type PluginRuntimeGatewayRequestScope = {
   context?: GatewayRequestContext;
+  resolveGatewayContext?: GatewayContextResolver;
   client?: GatewayRequestOptions["client"];
   isWebchatConnect: GatewayRequestOptions["isWebchatConnect"];
   pluginId?: string;
@@ -63,6 +64,21 @@ export function withPluginRuntimeGatewayRequestScope<T>(
   return pluginRuntimeGatewayRequestScope.run(scope, run);
 }
 
+/** Runs detached plugin work against one lifecycle-fenced Gateway instance. */
+export function withPluginRuntimeGatewayContextResolver<T>(
+  resolveGatewayContext: GatewayContextResolver,
+  run: () => T,
+): T {
+  const current = pluginRuntimeGatewayRequestScope.getStore();
+  const scoped: PluginRuntimeGatewayRequestScope = {
+    ...current,
+    isWebchatConnect: current?.isWebchatConnect ?? (() => false),
+    resolveGatewayContext,
+  };
+  delete scoped.context;
+  return pluginRuntimeGatewayRequestScope.run(scoped, run);
+}
+
 /** Runs work against an owned registry handle while preserving any gateway request facts. */
 export function withPluginRuntimeRegistryScope<T>(
   registry: PluginRegistry | undefined,

src/plugins/runtime/runtime-channel.test.ts

@@ -3,6 +3,13 @@ import { getEventListeners } from "node:events";
 import { describe, expect, it, vi } from "vitest";
 import { createRuntimeChannel } from "./runtime-channel.js";
 
+const dispatchRoutedChannelTurn = vi.hoisted(() => vi.fn(async () => ({ status: "handled" })));
+
+vi.mock("../../channels/turn/lifecycle.js", async (importOriginal) => ({
+  ...(await importOriginal<typeof import("../../channels/turn/lifecycle.js")>()),
+  dispatchRoutedChannelTurn,
+}));
+
 function requireWatcherEvent(mock: ReturnType<typeof vi.fn>, index: number) {
   const event = mock.mock.calls[index]?.[0] as { type?: string } | undefined;
   if (!event) {
@@ -11,6 +18,21 @@ function requireWatcherEvent(mock: ReturnType<typeof vi.fn>, index: number) {
   return event;
 }
 
+describe("inbound dispatch", () => {
+  it("carries the owning runtime reply dispatcher into routed channel turns", async () => {
+    const boundReplyDispatch = vi.fn();
+    const channel = createRuntimeChannel({ dispatchReplyFromConfig: boundReplyDispatch });
+    const turn = { channel: "qa-channel" } as Parameters<typeof channel.inbound.dispatch>[0];
+
+    await channel.inbound.dispatch(turn);
+
+    expect(dispatchRoutedChannelTurn).toHaveBeenCalledWith({
+      ...turn,
+      dispatchReplyFromConfig: boundReplyDispatch,
+    });
+  });
+});
+
 describe("runtimeContexts", () => {
   it("registers, resolves, watches, and unregisters contexts", () => {
     const channel = createRuntimeChannel();

src/plugins/runtime/runtime-channel.ts

@@ -83,6 +83,13 @@ import type { PluginRuntime } from "./types.js";
 export function createRuntimeChannel(options?: {
   dispatchReplyFromConfig?: PluginRuntime["channel"]["reply"]["dispatchReplyFromConfig"];
 }): PluginRuntime["channel"] {
+  const dispatchInbound: typeof dispatchRoutedChannelTurn = (params) =>
+    dispatchRoutedChannelTurn({
+      ...params,
+      ...(options?.dispatchReplyFromConfig
+        ? { dispatchReplyFromConfig: options.dispatchReplyFromConfig }
+        : {}),
+    });
   const sessionRuntime = {
     resolveStorePath: resolveSessionStorePathCore,
     readSessionUpdatedAt: readSessionUpdatedAtCore,
@@ -190,7 +197,7 @@ export function createRuntimeChannel(options?: {
       buildContext: buildChannelInboundEventContext,
       run: runChannelTurn,
       runPreparedReply: runPreparedChannelTurn,
-      dispatch: dispatchRoutedChannelTurn,
+      dispatch: dispatchInbound,
       dispatchReply: dispatchAssembledChannelTurn,
     },
     threadBindings: {

src/tasks/agent-harness-task-runtime-scope.ts

@@ -1,4 +1,6 @@
 // Resolves task runtime scope for agent harness launches.
+import type { GatewayContextResolver } from "../gateway/server-methods/types.js";
+import { bindGatewayContextResolver } from "../plugins/runtime/gateway-request-scope.js";
 import { normalizeDeliveryContext } from "../utils/delivery-context.shared.js";
 import type { DeliveryContext } from "../utils/delivery-context.types.js";
 
@@ -30,6 +32,7 @@ export type AgentHarnessTaskRuntimeScope = {
 export function createAgentHarnessTaskRuntimeScope(params: {
   requesterSessionKey: string;
   requesterOrigin?: DeliveryContext;
+  gatewayContextResolver?: GatewayContextResolver;
 }): AgentHarnessTaskRuntimeScope {
   const requesterSessionKey = params.requesterSessionKey.trim();
   if (!requesterSessionKey) {
@@ -41,6 +44,7 @@ export function createAgentHarnessTaskRuntimeScope(params: {
     ...(requesterOrigin ? { requesterOrigin } : {}),
   };
   getScopeRegistry().hostIssuedScopes.add(scope);
+  bindGatewayContextResolver(scope, params.gatewayContextResolver);
   return scope;
 }