fix(acp): drain runtime handles on gateway shutdown (#121359)

* fix(acp): drain runtime handles on gateway shutdown

* refactor(acp): isolate manager shutdown lifecycle

* style(acp): brace lifecycle guard

* fix(acp): cancel active turns before shutdown drain

* fix(acpx): reap process tree before wrapper close

* fix(test): diagnose surviving Vitest process groups

* fix(test): type process completion signals portably

* fix(test): reap orphaned live Docker workers

* fix(test): preserve child signal typing

* fix(gateway): retain ACPX through manager drain

* test(gateway): type deferred ACP disposal exactly

scripts/lib/live-docker-auth.sh

@@ -462,10 +462,12 @@ openclaw_live_init_docker_run_args() {
     return 127
   fi
   quoted_timeout="$(printf '%q' "$timeout_value")"
+  # Provider CLIs can leave orphaned tooling grandchildren; Docker init reaps
+  # them so sequential Vitest process groups can join after teardown.
   if openclaw_live_timeout_supports_kill_after "$timeout_bin"; then
-    eval "${target_array}=(${timeout_bin} --kill-after=30s ${quoted_timeout} docker run)"
+    eval "${target_array}=(${timeout_bin} --kill-after=30s ${quoted_timeout} docker run --init)"
   else
-    eval "${target_array}=(${timeout_bin} ${quoted_timeout} docker run)"
+    eval "${target_array}=(${timeout_bin} ${quoted_timeout} docker run --init)"
   fi
   openclaw_live_docker_run_resource_args resource_args || return $?
   openclaw_live_append_array "$target_array" resource_args

scripts/vitest-process-group.mts

@@ -1,5 +1,5 @@
 // Shared Vitest child process-group signal forwarding helpers.
-import type { ChildProcess } from "node:child_process";
+import { execFileSync, type ChildProcess } from "node:child_process";
 
 type VitestProcessSignal = "SIGINT" | "SIGKILL" | "SIGTERM";
 type KillProcess = (pid: number, signal?: VitestProcessSignal | 0) => boolean;
@@ -79,6 +79,7 @@ export function forceKillVitestProcessGroup(
 }
 
 const PROCESS_GROUP_JOIN_TIMEOUT_MS = 1_000;
+const PROCESS_GROUP_INSPECT_TIMEOUT_MS = 1_000;
 
 function errorCode(error: unknown) {
   return error && typeof error === "object" && "code" in error ? error.code : undefined;
@@ -100,6 +101,39 @@ function isVitestProcessGroupAlive(target: number, kill: KillProcess) {
   }
 }
 
+export function parseVitestProcessGroupMembers(output: string, processGroupId: number): string {
+  const members: string[] = [];
+  for (const line of output.split(/\r?\n/)) {
+    const match = /^\s*(\d+)\s+(\d+)\s+(\d+)\s+(\S+)\s+(.+?)\s*$/.exec(line);
+    if (!match || Number(match[3]) !== processGroupId) {
+      continue;
+    }
+    members.push(
+      `pid=${match[1]} ppid=${match[2]} state=${match[4]} comm=${match[5]?.slice(0, 80)}`,
+    );
+    if (members.length >= 20) {
+      break;
+    }
+  }
+  return members.length > 0 ? members.join("; ") : "none";
+}
+
+function inspectVitestProcessGroup(processGroupId: number, platform: NodeJS.Platform): string {
+  if (platform === "win32") {
+    return "unavailable";
+  }
+  try {
+    const output = execFileSync("ps", ["-axo", "pid=,ppid=,pgid=,stat=,comm="], {
+      encoding: "utf8",
+      maxBuffer: 1024 * 1024,
+      timeout: PROCESS_GROUP_INSPECT_TIMEOUT_MS,
+    });
+    return parseVitestProcessGroupMembers(output, processGroupId);
+  } catch {
+    return "unavailable";
+  }
+}
+
 async function joinVitestProcessGroup(
   child: VitestChild,
   platform: NodeJS.Platform,
@@ -114,8 +148,9 @@ async function joinVitestProcessGroup(
   while (isVitestProcessGroupAlive(target, kill)) {
     const remainingMs = deadlineAt - Date.now();
     if (remainingMs <= 0) {
+      const members = inspectVitestProcessGroup(child.pid!, platform);
       throw new Error(
-        `[vitest] process group ${child.pid ?? "unknown"} remained alive ${PROCESS_GROUP_JOIN_TIMEOUT_MS}ms after SIGKILL; inspect its descendants before starting another shard.`,
+        `[vitest] process group ${child.pid ?? "unknown"} remained alive ${PROCESS_GROUP_JOIN_TIMEOUT_MS}ms after SIGKILL; members: ${members}.`,
       );
     }
     await new Promise((resolve) => {
@@ -125,10 +160,12 @@ async function joinVitestProcessGroup(
 }
 
 function waitForChildCompletionEvent(child: ChildProcess, event: "exit" | "close") {
-  return new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve, reject) => {
-    child.once(event, (code, signal) => resolve({ code, signal }));
-    child.once("error", reject);
-  });
+  return new Promise<{ code: number | null; signal: ChildProcess["signalCode"] }>(
+    (resolve, reject) => {
+      child.once(event, (code, signal) => resolve({ code, signal }));
+      child.once("error", reject);
+    },
+  );
 }
 
 /**

src/acp/control-plane/manager.cancel-session.ts

@@ -29,7 +29,7 @@ export async function runManagerCancelSession(params: {
   const actorKey = normalizeActorKey(params.sessionKey);
   const activeTurn = params.activeTurnBySession.get(actorKey);
   if (activeTurn) {
-    await cancelActiveTurn({
+    await cancelManagerActiveTurn({
       activeTurn,
       reason: params.reason,
     });
@@ -72,7 +72,8 @@ export async function runManagerCancelSession(params: {
   });
 }
 
-async function cancelActiveTurn(params: {
+/** Aborts and deduplicates runtime cancellation for one active manager turn. */
+export async function cancelManagerActiveTurn(params: {
   activeTurn: ActiveTurnState;
   reason?: string;
 }): Promise<void> {

src/acp/control-plane/manager.core.ts

@@ -10,10 +10,11 @@ import { logVerbose } from "../../globals.js";
 import { toErrorObject } from "../../infra/errors.js";
 import { isAcpSessionKey } from "../../sessions/session-key-utils.js";
 import { AcpRuntimeError } from "../runtime/errors.js";
-import { runManagerCancelSession } from "./manager.cancel-session.js";
+import { cancelManagerActiveTurn, runManagerCancelSession } from "./manager.cancel-session.js";
 import { runManagerCloseSession } from "./manager.close-session.js";
 import { reconcileManagerRuntimeSessionIdentifiers } from "./manager.identity-reconcile.js";
 import { runManagerInitializeSession } from "./manager.initialize-session.js";
+import { registerAcpSessionManagerDisposer } from "./manager.lifecycle.js";
 import {
   applyManagerRuntimeControls,
   resolveManagerRuntimeCapabilities,
@@ -79,6 +80,21 @@ export class AcpSessionManager {
 
   constructor(deps: AcpSessionManagerDeps = DEFAULT_DEPS) {
     this.deps = deps;
+    registerAcpSessionManagerDisposer(this, async (reason) => {
+      await Promise.all(
+        [...this.activeTurnBySession.values()].map(async (activeTurn) => {
+          try {
+            await cancelManagerActiveTurn({ activeTurn, reason });
+          } catch (error) {
+            logVerbose(
+              `acp-manager: active runtime cancel failed for ${activeTurn.handle.sessionKey}: ${String(error)}`,
+            );
+          }
+        }),
+      );
+      await this.runtimeHandles.closeAll({ actorQueue: this.actorQueue, reason });
+      this.activeTurnBySession.clear();
+    });
   }
 
   resolveSession(params: { cfg: OpenClawConfig; sessionKey: string }): AcpSessionResolution {

src/acp/control-plane/manager.lifecycle.ts

@@ -0,0 +1,23 @@
+/** Internal process-lifecycle registry for ACP session manager instances. */
+type AcpSessionManagerDisposer = (reason: string) => Promise<void>;
+
+const ACP_SESSION_MANAGER_DISPOSERS = new WeakMap<object, AcpSessionManagerDisposer>();
+
+export function registerAcpSessionManagerDisposer(
+  manager: object,
+  dispose: AcpSessionManagerDisposer,
+): void {
+  ACP_SESSION_MANAGER_DISPOSERS.set(manager, dispose);
+}
+
+/** Stops active turns and closes process-local handles without widening the public manager API. */
+export async function disposeAcpSessionManagerInstance(
+  manager: object,
+  reason: string,
+): Promise<void> {
+  const dispose = ACP_SESSION_MANAGER_DISPOSERS.get(manager);
+  if (!dispose) {
+    throw new Error("ACP session manager disposer unavailable");
+  }
+  await dispose(reason);
+}

src/acp/control-plane/manager.runtime-handle-cache.ts

@@ -72,6 +72,17 @@ export class ManagerRuntimeHandleCache {
     }
   }
 
+  /** Drains every cached handle behind its session actor before process shutdown. */
+  async closeAll(params: { actorQueue: SessionActorQueue; reason: string }): Promise<void> {
+    await Promise.all(
+      this.runtimeCache.snapshot().map(({ actorKey }) =>
+        params.actorQueue.run(actorKey, async () => {
+          await this.close({ sessionKey: actorKey, reason: params.reason });
+        }),
+      ),
+    );
+  }
+
   /** Clears a cached handle only when the caller still owns the same runtime identifiers. */
   clearIfHandleMatches(params: { sessionKey: string; handle: AcpRuntimeHandle }): void {
     const cached = this.get(params.sessionKey);

src/acp/control-plane/manager.runtime-handles.test.ts

@@ -1,10 +1,11 @@
 /** Tests ACP runtime handle caching, reuse, re-ensure, and eviction behavior. */
-import { describe, expect, it } from "vitest";
+import { describe, expect, it, vi } from "vitest";
 import {
   AcpRuntimeError,
   AcpSessionManager,
   baseCfg,
   createRuntime,
+  disposeAcpSessionManagerInstance,
   expectRecordFields,
   hoisted,
   installAcpSessionManagerTestLifecycle,
@@ -51,6 +52,97 @@ describe("AcpSessionManager runtime handles", () => {
     expect(runtimeState.runTurn).toHaveBeenCalledTimes(2);
   });
 
+  it("disposes every retained runtime handle", async () => {
+    const runtimeState = createRuntime();
+    hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
+      id: "acpx",
+      runtime: runtimeState.runtime,
+    });
+    hoisted.readAcpSessionEntryMock.mockImplementation((input: unknown) => {
+      const sessionKey = (input as { sessionKey: string }).sessionKey;
+      return {
+        sessionKey,
+        storeSessionKey: sessionKey,
+        acp: readySessionMeta(),
+      };
+    });
+    const manager = new AcpSessionManager();
+
+    for (const [index, sessionKey] of [
+      "agent:claude:acp:session-1",
+      "agent:codex:acp:session-2",
+    ].entries()) {
+      await manager.runTurn({
+        provenance: "system",
+        cfg: baseCfg,
+        sessionKey,
+        text: `turn ${index + 1}`,
+        mode: "prompt",
+        requestId: `r${index + 1}`,
+      });
+    }
+
+    await disposeAcpSessionManagerInstance(manager, "gateway-shutdown");
+
+    expect(runtimeState.close).toHaveBeenCalledTimes(2);
+    expect(
+      new Set(
+        runtimeState.close.mock.calls.map(
+          ([input]) => (input as { handle: { sessionKey: string } }).handle.sessionKey,
+        ),
+      ),
+    ).toEqual(new Set(["agent:claude:acp:session-1", "agent:codex:acp:session-2"]));
+    expect(manager.getObservabilitySnapshot().runtimeCache.activeSessions).toBe(0);
+  });
+
+  it("cancels an active turn before closing its retained runtime handle", async () => {
+    const runtimeState = createRuntime();
+    let releaseTurn!: () => void;
+    const turnReleased = new Promise<void>((resolve) => {
+      releaseTurn = resolve;
+    });
+    const lifecycle: string[] = [];
+    runtimeState.runTurn.mockImplementation(async function* () {
+      await turnReleased;
+      yield { type: "done" as const };
+    });
+    runtimeState.cancel.mockImplementation(async () => {
+      lifecycle.push("cancel");
+      releaseTurn();
+    });
+    runtimeState.close.mockImplementation(async () => {
+      lifecycle.push("close");
+    });
+    hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
+      id: "acpx",
+      runtime: runtimeState.runtime,
+    });
+    const sessionKey = "agent:claude:acp:active-session";
+    hoisted.readAcpSessionEntryMock.mockReturnValue({
+      sessionKey,
+      storeSessionKey: sessionKey,
+      acp: readySessionMeta(),
+    });
+    const manager = new AcpSessionManager();
+    const turnPromise = manager
+      .runTurn({
+        provenance: "system",
+        cfg: baseCfg,
+        sessionKey,
+        text: "active turn",
+        mode: "prompt",
+        requestId: "r-active",
+      })
+      .catch(() => undefined);
+    await vi.waitFor(() => expect(runtimeState.runTurn).toHaveBeenCalledOnce());
+
+    await disposeAcpSessionManagerInstance(manager, "gateway-shutdown");
+    await turnPromise;
+
+    expect(lifecycle).toEqual(["cancel", "close"]);
+    expect(manager.getObservabilitySnapshot().runtimeCache.activeSessions).toBe(0);
+  });
+
   it("re-ensures cached runtime handles when the runtime config changes", async () => {
     const runtimeState = createRuntime();
     hoisted.requireAcpRuntimeBackendMock.mockReturnValue({

src/acp/control-plane/manager.test-helpers.ts

@@ -43,6 +43,9 @@ const managerModule = await import("./manager.js");
 export const AcpSessionManager = managerModule.AcpSessionManager;
 export const resetAcpSessionManagerForTests = () =>
   managerModule.testing.resetAcpSessionManagerForTests();
+const managerLifecycleModule = await import("./manager.lifecycle.js");
+export const disposeAcpSessionManagerInstance =
+  managerLifecycleModule.disposeAcpSessionManagerInstance;
 export const { AcpRuntimeError } = await import("../runtime/errors.js");
 
 export const baseCfg = {

src/gateway/server-close.test.ts

@@ -27,6 +27,8 @@ const mocks = vi.hoisted(() => ({
   disposeAllBundleLspRuntimes: vi.fn(async () => undefined),
   drainRetainedEmbeddingProviders: vi.fn(async () => undefined),
   clearSessionSuspensionTimers: vi.fn(() => 0),
+  disposeAcpSessionManagerInstance: vi.fn(async () => undefined),
+  getAcpSessionManager: vi.fn(() => ({})),
   closePluginStateDatabase: vi.fn(async () => undefined),
 }));
 const WEBSOCKET_CLOSE_GRACE_MS = 1_000;
@@ -87,6 +89,14 @@ vi.mock("../agents/session-suspension.js", () => ({
   clearSessionSuspensionTimers: mocks.clearSessionSuspensionTimers,
 }));
 
+vi.mock("../acp/control-plane/manager.lifecycle.js", () => ({
+  disposeAcpSessionManagerInstance: mocks.disposeAcpSessionManagerInstance,
+}));
+
+vi.mock("../acp/control-plane/manager.js", () => ({
+  getAcpSessionManager: mocks.getAcpSessionManager,
+}));
+
 vi.mock("../plugin-state/plugin-state-store.js", async () => ({
   ...(await vi.importActual<typeof import("../plugin-state/plugin-state-store.js")>(
     "../plugin-state/plugin-state-store.js",
@@ -195,6 +205,9 @@ describe("createGatewayCloseHandler", () => {
     mocks.drainRetainedEmbeddingProviders.mockResolvedValue(undefined);
     mocks.clearSessionSuspensionTimers.mockReset();
     mocks.clearSessionSuspensionTimers.mockReturnValue(0);
+    mocks.disposeAcpSessionManagerInstance.mockReset();
+    mocks.disposeAcpSessionManagerInstance.mockResolvedValue(undefined);
+    mocks.getAcpSessionManager.mockClear();
     mocks.closePluginStateDatabase.mockReset();
     mocks.closePluginStateDatabase.mockResolvedValue(undefined);
   });
@@ -356,8 +369,11 @@ describe("createGatewayCloseHandler", () => {
     ]);
   });
 
-  it("stops plugin services before channel runtimes", async () => {
+  it("disposes ACP sessions before plugin services and channel runtimes", async () => {
     const events: string[] = [];
+    mocks.disposeAcpSessionManagerInstance.mockImplementation(async () => {
+      events.push("acp-sessions");
+    });
     const pluginServices = {
       stop: vi.fn(async () => {
         events.push("plugin-services");
@@ -376,11 +392,56 @@ describe("createGatewayCloseHandler", () => {
 
     await close({ reason: "test" });
 
-    expect(events).toEqual(["plugin-services", "channel:discord"]);
+    expect(events).toEqual(["acp-sessions", "plugin-services", "channel:discord"]);
+    expect(mocks.disposeAcpSessionManagerInstance).toHaveBeenCalledWith(
+      expect.anything(),
+      "gateway-shutdown",
+    );
     expect(pluginServices.stop).toHaveBeenCalledTimes(1);
     expect(stopChannel).toHaveBeenCalledWith("discord");
   });
 
+  it("continues plugin shutdown when ACP session disposal fails", async () => {
+    mocks.disposeAcpSessionManagerInstance.mockRejectedValue(new Error("ACP close failed"));
+    const pluginServices = { stop: vi.fn(async () => undefined) };
+    const close = createGatewayCloseHandler(
+      createGatewayCloseTestDeps({ pluginServices: pluginServices as never }),
+    );
+
+    const result = await close({ reason: "test" });
+
+    expect(pluginServices.stop).toHaveBeenCalledOnce();
+    expect(result.warnings).toContain("acp-session-manager");
+  });
+
+  it("keeps plugin services alive until a slow ACP session disposal settles", async () => {
+    vi.useFakeTimers();
+    let releaseDisposal!: () => void;
+    mocks.disposeAcpSessionManagerInstance.mockReturnValue(
+      new Promise<undefined>((resolve) => {
+        releaseDisposal = () => resolve(undefined);
+      }),
+    );
+    const pluginServices = { stop: vi.fn(async () => undefined) };
+    const close = createGatewayCloseHandler(
+      createGatewayCloseTestDeps({ pluginServices: pluginServices as never }),
+    );
+
+    try {
+      const closePromise = close({ reason: "test" });
+      await vi.advanceTimersByTimeAsync(5_001);
+
+      expect(mocks.disposeAcpSessionManagerInstance).toHaveBeenCalledOnce();
+      expect(pluginServices.stop).not.toHaveBeenCalled();
+
+      releaseDisposal();
+      await closePromise;
+      expect(pluginServices.stop).toHaveBeenCalledOnce();
+    } finally {
+      vi.useRealTimers();
+    }
+  });
+
   it("clears the secrets runtime snapshot only after channels stop (#112681)", async () => {
     const events: string[] = [];
     const stopChannel = vi.fn(async (channelId: string) => {

src/gateway/server-close.ts

@@ -4,6 +4,8 @@ import type { Server as HttpServer } from "node:http";
 import { cleanupSessionResources } from "@openclaw/ai/internal/runtime";
 import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce";
 import type { WebSocketServer } from "ws";
+import { getAcpSessionManager } from "../acp/control-plane/manager.js";
+import { disposeAcpSessionManagerInstance } from "../acp/control-plane/manager.lifecycle.js";
 import { disposeAllSessionMcpRuntimes } from "../agents/agent-bundle-mcp-tools.js";
 import { disposeRegisteredAgentHarnesses } from "../agents/harness/registry.js";
 import { createAgentRunRestartAbortError } from "../agents/run-termination.js";
@@ -855,6 +857,15 @@ export function createGatewayCloseHandler(
           }
         });
       }
+      // ACPX owns agent-process cleanup, so plugin teardown must not overtake
+      // the manager drain even when cancellation and handle close are slow.
+      await measureCloseStep("acp-session-manager", () =>
+        shutdownStep(
+          "acp-session-manager",
+          () => disposeAcpSessionManagerInstance(getAcpSessionManager(), "gateway-shutdown"),
+          warnings,
+        ),
+      );
       if (params.pluginServices) {
         await measureCloseStep("plugin-services", () =>
           // A stalled plugin must not prevent later runtime and child-process cleanup.

src/gateway/server-startup-post-attach.test.ts

@@ -173,6 +173,10 @@ vi.mock("../acp/control-plane/manager.js", () => ({
   })),
 }));
 
+vi.mock("../acp/control-plane/manager.lifecycle.js", () => ({
+  disposeAcpSessionManagerInstance: vi.fn(async () => undefined),
+}));
+
 vi.mock("../acp/runtime/registry.js", () => ({
   getAcpRuntimeBackend: hoisted.getAcpRuntimeBackend,
 }));

test/scripts/live-docker-auth.test.ts

@@ -219,6 +219,7 @@ describe("scripts/lib/live-docker-auth.sh", () => {
       "42s",
       "docker",
       "run",
+      "--init",
       "--memory",
       "8g",
       "--cpus",
@@ -270,6 +271,7 @@ describe("scripts/lib/live-docker-auth.sh", () => {
       "42s",
       "docker",
       "run",
+      "--init",
       "--memory",
       "8g",
       "--cpus",
@@ -293,6 +295,7 @@ describe("scripts/lib/live-docker-auth.sh", () => {
       "42s",
       "docker",
       "run",
+      "--init",
       "--memory",
       "8g",
       "--cpus",
@@ -322,6 +325,7 @@ describe("scripts/lib/live-docker-auth.sh", () => {
       "42s",
       "docker",
       "run",
+      "--init",
       "--memory",
       "8g",
       "--cpus",
@@ -373,6 +377,7 @@ describe("scripts/lib/live-docker-auth.sh", () => {
       "42s",
       "docker",
       "run",
+      "--init",
     ]);
   });
 

test/scripts/vitest-process-group.test.ts

@@ -5,6 +5,7 @@ import {
   createVitestProcessCompletion,
   forwardSignalToVitestProcessGroup,
   installVitestProcessGroupCleanup,
+  parseVitestProcessGroupMembers,
   resolveVitestProcessGroupSignalTarget,
   shouldUseDetachedVitestProcessGroup,
 } from "../../scripts/vitest-process-group.mts";
@@ -42,6 +43,15 @@ describe("vitest process group helpers", () => {
     );
   });
 
+  it("formats bounded process-group diagnostics without command arguments", () => {
+    expect(
+      parseVitestProcessGroupMembers(
+        [" 116 1 116 Z node", " 117 1 116 Sl claude", " 118 1 999 S unrelated"].join("\n"),
+        116,
+      ),
+    ).toBe("pid=116 ppid=1 state=Z comm=node; pid=117 ppid=1 state=Sl comm=claude");
+  });
+
   it("forwards signals to the computed target and ignores cleanup races", () => {
     const kill = vi.fn();
     expect(