fix(acp): complete turns against agents that notify before session/new resolves

src/runtimes/acp/acp-agent-process.ts

@@ -15,7 +15,7 @@ export type AcpAgentProcess = ChildProcessByStdio<Writable, Readable, Readable>;
 
 const ACP_AGENT_EXIT_TIMEOUT_MS = 1_500;
 const ACP_AGENT_FORCE_KILL_TIMEOUT_MS = 500;
-const agentProcessCloseTasks = new WeakMap<AcpAgentProcess, Promise<void>>();
+const agentProcessExitTasks = new WeakMap<AcpAgentProcess, Promise<void>>();
 
 export async function startAcpAgentProcess(
   context: AgentDriverContext,
@@ -52,15 +52,19 @@ export async function startAcpAgentProcess(
   });
   const onAbort = () => killProcessGroup(agentProcess, "SIGKILL");
   signal.addEventListener("abort", onAbort, { once: true });
-  agentProcessCloseTasks.set(
-    agentProcess,
-    new Promise<void>((resolve) =>
-      agentProcess.once("close", () => {
-        signal.removeEventListener("abort", onAbort);
-        resolve();
-      }),
-    ),
-  );
+  // Stdio "close" is not a reliable exit signal: the ACP transport keeps the
+  // child's stdout locked behind web-stream readers, so "close" can stay
+  // pending after the process itself is gone. Observe "exit" as well.
+  const exited = Promise.withResolvers<void>();
+  const onExit = () => {
+    agentProcess.off("exit", onExit);
+    agentProcess.off("close", onExit);
+    signal.removeEventListener("abort", onAbort);
+    exited.resolve();
+  };
+  agentProcess.once("exit", onExit);
+  agentProcess.once("close", onExit);
+  agentProcessExitTasks.set(agentProcess, exited.promise);
 
   agentProcess.stderr.setEncoding("utf8");
   agentProcess.stderr.on("data", (chunk: string) => {
@@ -144,13 +148,17 @@ async function waitForChildProcessExit(
   timeoutMs: number,
   signal?: AbortSignal,
 ): Promise<boolean> {
-  const closed = agentProcessCloseTasks.get(process);
+  if (process.exitCode !== null || process.signalCode !== null) {
+    return true;
+  }
 
-  if (closed === undefined) {
-    return process.exitCode !== null || process.signalCode !== null;
+  const exited = agentProcessExitTasks.get(process);
+
+  if (exited === undefined) {
+    return false;
   }
 
-  const result = await settlePromiseWithTimeout(closed, {
+  const result = await settlePromiseWithTimeout(exited, {
     label: "ACP agent process exit",
     ...(signal === undefined ? {} : { signal }),
     timeoutMs,
@@ -161,7 +169,7 @@ async function waitForChildProcessExit(
   }
 
   if (result.status === "timed_out") {
-    return false;
+    return process.exitCode !== null || process.signalCode !== null;
   }
 
   throw result.error;

src/runtimes/acp/acp-client-request-handler.ts

@@ -29,6 +29,8 @@ import { AcpSessionUpdateInbox, type AcpSessionUpdateScope } from "./acp-session
 import { AcpTerminalManager } from "./acp-terminal-manager";
 import { isRecord, raceWithAbort, readNonEmptyString, stringifyForDisplay } from "./acp-types";
 
+const MAX_SETUP_DEFERRED_UPDATES = 1_024;
+
 interface AcpClientRequestHandlerOptions {
   readonly allowedRoots: readonly string[];
   readonly cwd: string;
@@ -45,6 +47,10 @@ export class AcpClientRequestHandler {
   readonly #isCancelling: () => boolean;
   readonly #nativeSessionId: () => string | null;
   readonly #push: AcpClientRequestHandlerOptions["push"];
+  #setupDeferredUpdates: {
+    readonly notification: SessionNotification;
+    readonly scope: AcpSessionUpdateScope;
+  }[] = [];
   #stopping = false;
   readonly #updateInbox: AcpSessionUpdateInbox;
   readonly #terminalManager: AcpTerminalManager;
@@ -183,6 +189,27 @@ export class AcpClientRequestHandler {
     await this.#updateInbox.drain();
   }
 
+  async applySetupDeferredUpdates(context: AgentDriverContext): Promise<void> {
+    while (this.#setupDeferredUpdates.length > 0) {
+      const deferred = this.#setupDeferredUpdates;
+      this.#setupDeferredUpdates = [];
+
+      for (const entry of deferred) {
+        const record = isRecord(entry.notification) ? entry.notification : null;
+        const sessionId = readNonEmptyString(record, "sessionId");
+
+        if (sessionId === null || sessionId !== this.#nativeSessionId()) {
+          context.logger.warn("driver.acp.session.update.setup_mismatch_dropped", {
+            expectedSessionPresent: this.#nativeSessionId() !== null,
+          });
+          continue;
+        }
+
+        await this.#applyUpdate(context, entry.notification, entry.scope);
+      }
+    }
+  }
+
   async withSessionReplay<T>(operation: () => Promise<T>): Promise<T> {
     return this.#updateInbox.withReplay(operation);
   }
@@ -237,6 +264,19 @@ export class AcpClientRequestHandler {
     params: SessionNotification,
     scope: AcpSessionUpdateScope,
   ): Promise<void> {
+    // Some ACP agents (the OpenClaw Gateway bridge) notify session/update for
+    // a newly created session before the session/new response arrives. No
+    // session id is registered yet, so hold those updates instead of failing
+    // the transport; setup applies them right after the id registers.
+    if (this.#nativeSessionId() === null) {
+      if (this.#setupDeferredUpdates.length >= MAX_SETUP_DEFERRED_UPDATES) {
+        throw new Error("ACP session update queue limit exceeded.");
+      }
+
+      this.#setupDeferredUpdates.push({ notification: params, scope });
+      return;
+    }
+
     this.#assertSession("session/update", params);
 
     if (scope.suppressed && shouldIgnoreReplay(params)) {

src/runtimes/acp/acp-driver-backend.ts

@@ -237,7 +237,7 @@ export class AcpDriverBackend implements AgentDriverBackend {
       );
       const setup = await withAcpStartupStage(
         "ACP session setup",
-        () => this.#setupSession(signal),
+        () => this.#setupSession(context, signal),
         signal,
       );
 
@@ -572,7 +572,10 @@ export class AcpDriverBackend implements AgentDriverBackend {
     }
   }
 
-  async #setupSession(signal: AbortSignal): Promise<Awaited<ReturnType<typeof setupAcpSession>>> {
+  async #setupSession(
+    context: AgentDriverContext,
+    signal: AbortSignal,
+  ): Promise<Awaited<ReturnType<typeof setupAcpSession>>> {
     const hostSnapshot = this.#requireHostSnapshot();
     signal.throwIfAborted();
     const setup = await raceWithAbort(
@@ -588,6 +591,7 @@ export class AcpDriverBackend implements AgentDriverBackend {
     );
     signal.throwIfAborted();
     this.#nativeSessionId = setup.sessionId;
+    await raceWithAbort(this.#clientRequests.applySetupDeferredUpdates(context), signal);
     await raceWithAbort(this.#clientRequests.drainUpdates(), signal);
 
     return setup;

tests/acp-agent-process.test.ts

@@ -94,7 +94,7 @@ describe("ACP agent process lifecycle", () => {
     }
   });
 
-  test("waits for close when a descendant keeps the exited leader's stdio open", async () => {
+  test("stops promptly when a descendant keeps the exited leader's stdio open", async () => {
     const harness = createHarness();
     const root = await mkdtemp(join(tmpdir(), "driver-acp-process-"));
     const boot = {
@@ -142,13 +142,23 @@ describe("ACP agent process lifecycle", () => {
       }
 
       expect(closed).toBe(false);
-      await stopAcpAgentProcess(
-        harness.context,
-        child,
-        "test.stop",
-        Date.now() + 2_000,
-        new AbortController().signal,
-      );
+      // Process exit is the stop signal; stdio held open by a descendant must
+      // not block or fail the stop. The group SIGKILL still reaps the
+      // descendant, so close arrives shortly after.
+      await expect(
+        stopAcpAgentProcess(
+          harness.context,
+          child,
+          "test.stop",
+          Date.now() + 2_000,
+          new AbortController().signal,
+        ),
+      ).resolves.toBeUndefined();
+
+      if (!closed) {
+        await once(child, "close");
+      }
+
       expect(closed).toBe(true);
     } finally {
       if (child !== undefined && !closed) {