fix(agents): dispatch subagent spawn in process (#90612)

lanzhi-lee · steipete · web-flow · commit 58bab0c2761e · 2026-06-07T03:01:51.000-07:00
* fix(agents): dispatch subagent spawn in process

* docs: update subagent gateway dispatch note

* fix(gateway): keep in-process dispatch timeout budget

* test(gateway): avoid promise executor timer returns

---------

Co-authored-by: Peter Steinberger &lt;steipete@gmail.com&gt;
diff --git a/docs/tools/subagents.md b/docs/tools/subagents.md
@@ -621,10 +621,12 @@ tombstoned sessions.
 <Note>
 If a sub-agent spawn fails with Gateway `PAIRING_REQUIRED` /
 `scope-upgrade`, check the RPC caller before editing pairing state.
-Internal `sessions_spawn` coordination should connect as
-`client.id: "gateway-client"` with `client.mode: "backend"` over direct
-loopback shared-token/password auth; that path does not depend on the
-CLI's paired-device scope baseline. Remote callers, explicit
+Internal `sessions_spawn` coordination dispatches in process when the
+caller is already running inside the gateway request context, so it does
+not open a loopback WebSocket or depend on the CLI's paired-device scope
+baseline. Callers outside the gateway process still use the WebSocket
+fallback as `client.id: "gateway-client"` with `client.mode: "backend"`
+over direct loopback shared-token/password auth. Remote callers, explicit
 `deviceIdentity`, explicit device-token paths, and browser/node clients
 still need normal device approval for scope upgrades.
 </Note>
diff --git a/src/agents/subagent-spawn.runtime.ts b/src/agents/subagent-spawn.runtime.ts
@@ -17,6 +17,10 @@ export {
 export { ensureContextEnginesInitialized } from "../context-engine/init.js";
 export { resolveContextEngine } from "../context-engine/registry.js";
 export { callGateway } from "../gateway/call.js";
+export {
+  dispatchGatewayMethodInProcess,
+  hasInProcessGatewayContext,
+} from "../gateway/server-plugins.js";
 export { ADMIN_SCOPE, isAdminOnlyMethod } from "../gateway/method-scopes.js";
 export { getSessionBindingService } from "../infra/outbound/session-binding-service.js";
 export {
diff --git a/src/agents/subagent-spawn.test-helpers.ts b/src/agents/subagent-spawn.test-helpers.ts
@@ -128,6 +128,8 @@ export function expectPersistedRuntimeModel(params: {
 /** Load subagent-spawn with runtime dependencies replaced by test doubles. */
 export async function loadSubagentSpawnModuleForTest(params: {
   callGatewayMock: MockFn;
+  dispatchGatewayMethodInProcessMock?: MockFn;
+  hasInProcessGatewayContextMock?: MockFn;
   getRuntimeConfig?: () => Record<string, unknown>;
   loadSessionStoreMock?: MockFn;
   ensureContextEnginesInitializedMock?: MockFn;
@@ -209,6 +211,9 @@ export async function loadSubagentSpawnModuleForTest(params: {
 
   vi.doMock("./subagent-spawn.runtime.js", () => ({
     callGateway: (opts: unknown) => params.callGatewayMock(opts),
+    dispatchGatewayMethodInProcess: (...args: unknown[]) =>
+      params.dispatchGatewayMethodInProcessMock?.(...args),
+    hasInProcessGatewayContext: () => Boolean(params.hasInProcessGatewayContextMock?.()),
     buildSubagentSystemPrompt: () => "system-prompt",
     forkSessionFromParent:
       params.forkSessionFromParentMock ??
diff --git a/src/agents/subagent-spawn.test.ts b/src/agents/subagent-spawn.test.ts
@@ -17,6 +17,8 @@ const hoisted = vi.hoisted(() => ({
   pruneLegacyStoreKeysMock: vi.fn(),
   registerSubagentRunMock: vi.fn(),
   emitSessionLifecycleEventMock: vi.fn(),
+  dispatchGatewayMethodInProcessMock: vi.fn(),
+  hasInProcessGatewayContextMock: vi.fn(),
   resolveAgentConfigMock: vi.fn(),
   configOverride: {} as Record<string, unknown>,
 }));
@@ -67,6 +69,8 @@ describe("spawnSubagentDirect seam flow", () => {
   beforeAll(async () => {
     ({ resetSubagentRegistryForTests, spawnSubagentDirect } = await loadSubagentSpawnModuleForTest({
       callGatewayMock: hoisted.callGatewayMock,
+      dispatchGatewayMethodInProcessMock: hoisted.dispatchGatewayMethodInProcessMock,
+      hasInProcessGatewayContextMock: hoisted.hasInProcessGatewayContextMock,
       getRuntimeConfig: () => hoisted.configOverride,
       loadSessionStoreMock: hoisted.loadSessionStoreMock,
       updateSessionStoreMock: hoisted.updateSessionStoreMock,
@@ -88,6 +92,8 @@ describe("spawnSubagentDirect seam flow", () => {
     hoisted.pruneLegacyStoreKeysMock.mockReset();
     hoisted.registerSubagentRunMock.mockReset();
     hoisted.emitSessionLifecycleEventMock.mockReset();
+    hoisted.dispatchGatewayMethodInProcessMock.mockReset();
+    hoisted.hasInProcessGatewayContextMock.mockReset().mockReturnValue(false);
     hoisted.resolveAgentConfigMock.mockReset();
     hoisted.resolveAgentConfigMock.mockImplementation(
       (cfg: { agents?: { list?: Array<{ id?: string }> } }, agentId: string) =>
@@ -266,6 +272,76 @@ describe("spawnSubagentDirect seam flow", () => {
     expect(agentParams.cleanupBundleMcpOnRunEnd).toBe(true);
   });
 
+  it("dispatches spawned agent runs in process when a gateway context is available", async () => {
+    hoisted.hasInProcessGatewayContextMock.mockReturnValue(true);
+    hoisted.callGatewayMock.mockRejectedValue(new Error("unexpected websocket gateway call"));
+    hoisted.dispatchGatewayMethodInProcessMock.mockImplementation(async (method: string) => {
+      if (method === "agent") {
+        return { runId: "run-in-process" };
+      }
+      return { ok: true };
+    });
+
+    const result = await spawnSubagentDirect(
+      {
+        task: "spawn without websocket self-connection",
+      },
+      {
+        agentSessionKey: "agent:main:main",
+      },
+    );
+
+    expect(result.status).toBe("accepted");
+    expect(result.runId).toBe("run-in-process");
+    expect(hoisted.callGatewayMock).not.toHaveBeenCalled();
+    expect(hoisted.dispatchGatewayMethodInProcessMock).toHaveBeenCalledWith(
+      "agent",
+      expect.objectContaining({
+        message: expect.stringContaining("spawn without websocket self-connection"),
+        sessionKey: result.childSessionKey,
+      }),
+      expect.objectContaining({
+        timeoutMs: expect.any(Number),
+      }),
+    );
+  });
+
+  it("keeps admin-scoped cleanup on in-process spawn failure", async () => {
+    hoisted.hasInProcessGatewayContextMock.mockReturnValue(true);
+    hoisted.callGatewayMock.mockRejectedValue(new Error("unexpected websocket gateway call"));
+    hoisted.dispatchGatewayMethodInProcessMock.mockImplementation(async (method: string) => {
+      if (method === "agent") {
+        throw new Error("spawn failed");
+      }
+      return { ok: true };
+    });
+
+    const result = await spawnSubagentDirect(
+      {
+        task: "spawn failure cleanup",
+      },
+      {
+        agentSessionKey: "agent:main:main",
+      },
+    );
+
+    expect(result.status).toBe("error");
+    expect(result.error).toContain("spawn failed");
+    expect(hoisted.callGatewayMock).not.toHaveBeenCalled();
+    expect(hoisted.dispatchGatewayMethodInProcessMock).toHaveBeenCalledWith(
+      "sessions.delete",
+      expect.objectContaining({
+        key: result.childSessionKey,
+        deleteTranscript: true,
+      }),
+      expect.objectContaining({
+        forceSyntheticClient: true,
+        syntheticScopes: ["operator.admin"],
+        timeoutMs: 60_000,
+      }),
+    );
+  });
+
   it("inherits requester thinking level when no spawn or subagent default is configured", async () => {
     let persistedStore: Record<string, Record<string, unknown>> | undefined;
     hoisted.loadSessionStoreMock.mockReturnValue({
diff --git a/src/agents/subagent-spawn.ts b/src/agents/subagent-spawn.ts
@@ -87,11 +87,13 @@ import {
   DEFAULT_SUBAGENT_MAX_SPAWN_DEPTH,
   buildSubagentSystemPrompt,
   callGateway,
+  dispatchGatewayMethodInProcess,
   emitSessionLifecycleEvent,
   forkSessionFromParent,
   getGlobalHookRunner,
   getSessionBindingService,
   getRuntimeConfig,
+  hasInProcessGatewayContext,
   loadSessionStore,
   mergeSessionEntry,
   mergeDeliveryContext,
@@ -133,9 +135,11 @@ function resolveConfiguredAgentIds(cfg: OpenClawConfig): string[] {
 
 type SubagentSpawnDeps = {
   callGateway: typeof callGateway;
+  dispatchGatewayMethodInProcess: typeof dispatchGatewayMethodInProcess;
   forkSessionFromParent: typeof forkSessionFromParent;
   getGlobalHookRunner: () => SubagentLifecycleHookRunner | null;
   getRuntimeConfig: typeof getRuntimeConfig;
+  hasInProcessGatewayContext: typeof hasInProcessGatewayContext;
   ensureContextEnginesInitialized: typeof ensureContextEnginesInitialized;
   resolveContextEngine: typeof resolveContextEngine;
   resolveParentForkDecision: typeof resolveParentForkDecision;
@@ -144,9 +148,11 @@ type SubagentSpawnDeps = {
 
 const defaultSubagentSpawnDeps: SubagentSpawnDeps = {
   callGateway,
+  dispatchGatewayMethodInProcess,
   forkSessionFromParent,
   getGlobalHookRunner,
   getRuntimeConfig,
+  hasInProcessGatewayContext,
   ensureContextEnginesInitialized,
   resolveContextEngine,
   resolveParentForkDecision,
@@ -245,10 +251,30 @@ async function callSubagentGateway(
   // Only admin-only methods are pinned to ADMIN_SCOPE; other methods (e.g.
   // "agent" -> write) keep their least-privilege scope.
   const scopes = params.scopes ?? (isAdminOnlyMethod(params.method) ? [ADMIN_SCOPE] : undefined);
-  return await subagentSpawnDeps.callGateway({
+  const request = {
     ...params,
     ...(scopes != null ? { scopes } : {}),
-  });
+  };
+  if (
+    subagentSpawnDeps.hasInProcessGatewayContext() &&
+    request.params != null &&
+    typeof request.params === "object" &&
+    !Array.isArray(request.params)
+  ) {
+    // Spawn is already running in the gateway process for channel/tool calls.
+    // Direct dispatch avoids self-connecting over WS while the same event loop is busy.
+    return await subagentSpawnDeps.dispatchGatewayMethodInProcess(
+      request.method,
+      request.params as Record<string, unknown>,
+      {
+        expectFinal: request.expectFinal,
+        ...(scopes != null ? { forceSyntheticClient: true } : {}),
+        timeoutMs: request.timeoutMs,
+        ...(scopes != null ? { syntheticScopes: scopes } : {}),
+      },
+    );
+  }
+  return await subagentSpawnDeps.callGateway(request);
 }
 
 function readGatewayRunId(response: Awaited<ReturnType<typeof callGateway>>): string | undefined {
diff --git a/src/gateway/server-plugins.test.ts b/src/gateway/server-plugins.test.ts
@@ -773,6 +773,97 @@ describe("loadGatewayPlugins", () => {
     });
   });
 
+  test("times out while waiting for the first in-process gateway response", async () => {
+    serverPluginsModule.setFallbackGatewayContext(createTestContext("initial-response-timeout"));
+    handleGatewayRequest.mockImplementationOnce(async () => {
+      await new Promise(() => {});
+    });
+
+    await expect(
+      serverPluginsModule.dispatchGatewayMethodInProcess(
+        "sessions.delete",
+        { key: "stuck-session" },
+        { timeoutMs: 5 },
+      ),
+    ).rejects.toThrow("gateway request timeout for sessions.delete");
+  });
+
+  test("returns an accepted in-process response without waiting for handler completion", async () => {
+    serverPluginsModule.setFallbackGatewayContext(createTestContext("accepted-before-complete"));
+    handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => {
+      opts.respond(true, { status: "accepted", runId: "run-accepted" });
+      await new Promise(() => {});
+    });
+
+    await expect(
+      serverPluginsModule.dispatchGatewayMethodInProcess(
+        "agent",
+        { sessionKey: "s-accepted" },
+        { timeoutMs: 5 },
+      ),
+    ).resolves.toEqual({ status: "accepted", runId: "run-accepted" });
+  });
+
+  test("uses one timeout budget across accepted and final in-process responses", async () => {
+    vi.useFakeTimers();
+    try {
+      serverPluginsModule.setFallbackGatewayContext(createTestContext("single-final-deadline"));
+      handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => {
+        setTimeout(() => {
+          opts.respond(true, { status: "accepted", runId: "run-deadline" });
+        }, 7);
+        setTimeout(() => {
+          opts.respond(true, { status: "ok", runId: "run-deadline" });
+        }, 13);
+        await new Promise((resolve) => {
+          setTimeout(resolve, 13);
+        });
+      });
+
+      const result = expect(
+        serverPluginsModule.dispatchGatewayMethodInProcess(
+          "agent",
+          { sessionKey: "s-deadline" },
+          { expectFinal: true, timeoutMs: 10 },
+        ),
+      ).rejects.toThrow("gateway request timeout for agent");
+
+      await vi.advanceTimersByTimeAsync(10);
+      await result;
+      await vi.advanceTimersByTimeAsync(10);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
+  test("clears final-response timeout when handler rejects after accepted response", async () => {
+    vi.useFakeTimers();
+    try {
+      serverPluginsModule.setFallbackGatewayContext(createTestContext("accepted-then-error"));
+      handleGatewayRequest.mockImplementationOnce(async (opts: HandleGatewayRequestOptions) => {
+        opts.respond(true, { status: "accepted", runId: "run-error-after-accepted" });
+        await new Promise((resolve) => {
+          setTimeout(resolve, 5);
+        });
+        throw new Error("handler failed after accepted");
+      });
+
+      const result = expect(
+        serverPluginsModule.dispatchGatewayMethodInProcess(
+          "agent",
+          { sessionKey: "s-error-after-accepted" },
+          { expectFinal: true, timeoutMs: 1_000 },
+        ),
+      ).rejects.toThrow("handler failed after accepted");
+
+      await vi.advanceTimersByTimeAsync(5);
+      await result;
+      expect(vi.getTimerCount()).toBe(0);
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   test("filters connected plugin nodes locally without sending unsupported node.list params", async () => {
     loadOpenClawPlugins.mockReturnValue(createRegistry([]));
     loadGatewayStartupPluginsForTest();
@@ -1388,11 +1479,14 @@ describe("loadGatewayPlugins", () => {
     const runtime = await createSubagentRuntime(serverPlugins);
     let currentContext = createTestContext("before-resolver-update");
 
+    expect(serverPlugins.hasInProcessGatewayContext()).toBe(false);
     serverPlugins.setFallbackGatewayContextResolver(() => currentContext);
+    expect(serverPlugins.hasInProcessGatewayContext()).toBe(true);
     await runtime.run({ sessionKey: "s-4", message: "before resolver update" });
     expect(getLastDispatchedContext()).toBe(currentContext);
 
     currentContext = createTestContext("after-resolver-update");
+    expect(serverPlugins.hasInProcessGatewayContext()).toBe(true);
     await runtime.run({ sessionKey: "s-4", message: "after resolver update" });
     expect(getLastDispatchedContext()).toBe(currentContext);
   });
@@ -1434,6 +1528,7 @@ describe("loadGatewayPlugins", () => {
 
     serverPlugins.clearFallbackGatewayContext();
 
+    expect(serverPlugins.hasInProcessGatewayContext()).toBe(false);
     await expect(runtime.run({ sessionKey: "s-7", message: "after clear" })).rejects.toThrow(
       "No scope set and no fallback context available",
     );
diff --git a/src/gateway/server-plugins.ts b/src/gateway/server-plugins.ts
