From aa8dd52a3f085d9896e43017eea2dfe80e8c345a Mon Sep 17 00:00:00 2001 From: Chen Xiefan Date: Fri, 28 Aug 2026 15:08:23 +0800 Subject: [PATCH 1/2] fix(acp): defer session updates that arrive before session registration The OpenClaw Gateway ACP bridge (openclaw acp 2026.7.1-2) writes session/update notifications for a newly created session before it writes the session/new response. The client request handler asserted the update against the not-yet-registered native session id, threw "ACP driver backend session is not initialized.", and the update inbox failure closed the whole transport - every fresh-session turn against an OpenClaw gateway failed deterministically during ACP session setup. Hold updates that arrive while no native session id is registered in a bounded queue, and apply them right after session setup registers the id. Pre-registration updates for a different session are dropped with a warning instead of killing the transport. Regression: real-process stub agent reproducing the captured OpenClaw frame order (updates first, then the session/new response), asserting the prompt completes and the deferred updates are applied. --- .../acp/acp-client-request-handler.ts | 40 +++ src/runtimes/acp/acp-driver-backend.ts | 8 +- tests/acp-early-session-update.test.ts | 244 ++++++++++++++++++ 3 files changed, 290 insertions(+), 2 deletions(-) create mode 100644 tests/acp-early-session-update.test.ts diff --git a/src/runtimes/acp/acp-client-request-handler.ts b/src/runtimes/acp/acp-client-request-handler.ts index f0de3e4..7e2124f 100644 --- a/src/runtimes/acp/acp-client-request-handler.ts +++ b/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 { + 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(operation: () => Promise): Promise { return this.#updateInbox.withReplay(operation); } @@ -237,6 +264,19 @@ export class AcpClientRequestHandler { params: SessionNotification, scope: AcpSessionUpdateScope, ): Promise { + // 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)) { diff --git a/src/runtimes/acp/acp-driver-backend.ts b/src/runtimes/acp/acp-driver-backend.ts index 41c7f5b..f384967 100644 --- a/src/runtimes/acp/acp-driver-backend.ts +++ b/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>> { + async #setupSession( + context: AgentDriverContext, + signal: AbortSignal, + ): Promise>> { 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; diff --git a/tests/acp-early-session-update.test.ts b/tests/acp-early-session-update.test.ts new file mode 100644 index 0000000..587480e --- /dev/null +++ b/tests/acp-early-session-update.test.ts @@ -0,0 +1,244 @@ +import { describe, expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; + +import { createAgentDriverContext } from "../src/core/agent-driver-backend"; +import { createBufferedSinkLogger } from "../src/observability"; +import type { DriverBootPayload } from "../src/protocol/boot"; +import type { DriverEventInput } from "../src/protocol/events"; +import { createDriverHostIntegrationSnapshotFromBootExecution } from "../src/protocol/host-integration"; +import type { RunId } from "../src/protocol/id"; +import { createDriverStartInputFromBootPayload } from "../src/protocol/start"; +import { AcpDriverBackend } from "../src/runtimes/acp/acp-driver-backend"; +import { driverBootPayload, DRIVER_TEST_IDS } from "./driver-boot-payload-fixture"; + +// Real-process stub with the OpenClaw Gateway bridge's observed wire order: +// session/update notifications for the newly created session are written +// BEFORE the session/new response. Captured from openclaw acp 2026.7.1-2. +const BRIDGE_AGENT = String.raw` +let buffer = ""; +const send = (message) => process.stdout.write(JSON.stringify(message) + "\n"); +const sessionId = "acp-bridge-session-1"; +const handle = (message) => { + if (!("method" in message) || !("id" in message)) return; + switch (message.method) { + case "initialize": + send({ + id: message.id, + jsonrpc: "2.0", + result: { + agentCapabilities: { + loadSession: true, + promptCapabilities: { audio: false, embeddedContext: true, image: true }, + sessionCapabilities: { close: {}, list: {} }, + }, + agentInfo: { name: "openclaw-acp-stub", version: "0.0.0" }, + authMethods: [], + protocolVersion: 1, + }, + }); + return; + case "session/new": { + if (process.env.BRIDGE_STUB_EARLY === "foreign") { + send({ + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId: "some-other-session", + update: { sessionUpdate: "session_info_update", title: "foreign" }, + }, + }); + } else { + send({ + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId, + update: { sessionUpdate: "session_info_update", title: "acp-bridge:stub" }, + }, + }); + send({ + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId, + update: { + availableCommands: [{ description: "Show help.", name: "help" }], + sessionUpdate: "available_commands_update", + }, + }, + }); + } + send({ id: message.id, jsonrpc: "2.0", result: { sessionId } }); + return; + } + case "session/prompt": + send({ + jsonrpc: "2.0", + method: "session/update", + params: { + sessionId, + update: { + content: { text: "OPENCLAW-COPILOT-E2E", type: "text" }, + sessionUpdate: "agent_message_chunk", + }, + }, + }); + send({ id: message.id, jsonrpc: "2.0", result: { stopReason: "end_turn" } }); + return; + case "session/close": + send({ id: message.id, jsonrpc: "2.0", result: {} }); + return; + default: + send({ + error: { code: -32601, message: "method not found" }, + id: message.id, + jsonrpc: "2.0", + }); + } +}; +process.stdin.setEncoding("utf8"); +process.stdin.on("data", (chunk) => { + buffer += chunk; + let index = buffer.indexOf("\n"); + while (index >= 0) { + const line = buffer.slice(0, index).trim(); + buffer = buffer.slice(index + 1); + if (line.length > 0) handle(JSON.parse(line)); + index = buffer.indexOf("\n"); + } +}); +`; + +async function createBridgeHarness(early: "foreign" | "match") { + const root = await mkdtemp(join(tmpdir(), "driver-acp-early-update-")); + const boot = { + ...driverBootPayload, + execution: { + ...driverBootPayload.execution, + environment: { + variables: { BRIDGE_STUB_EARLY: early }, + }, + session: { + ...driverBootPayload.execution.session, + context: { + ...driverBootPayload.execution.session.context, + homePath: join(root, "home"), + sessionOrganizationPath: root, + }, + cwd: root, + }, + }, + runtime: "acp-fallback", + runtimeTransport: "acp-fallback", + } satisfies DriverBootPayload; + const payload = createDriverStartInputFromBootPayload(boot); + const logger = createBufferedSinkLogger({ + level: "debug", + service: "acp-early-update-test", + sink: async () => {}, + }); + let acceptedSeq = 0; + const publishedEvents: DriverEventInput[] = []; + const context = createAgentDriverContext({ + eventSink: { + pushEvents: async ({ events }) => { + publishedEvents.push(...events); + return { + accepted: events.map((event) => ({ seq: ++acceptedSeq, type: event.kind })), + }; + }, + }, + logger, + payload, + permission: { request: async () => "reject_once" }, + ports: { + hostIntegration: { + snapshot: async () => createDriverHostIntegrationSnapshotFromBootExecution(boot.execution), + }, + skill: { materialize: async () => [] }, + }, + }); + const backend = new AcpDriverBackend(payload); + const previousCommand = process.env["MOSOO_ACP_FALLBACK_COMMAND"]; + const previousArgs = process.env["MOSOO_ACP_FALLBACK_ARGS"]; + process.env["MOSOO_ACP_FALLBACK_COMMAND"] = process.execPath; + process.env["MOSOO_ACP_FALLBACK_ARGS"] = JSON.stringify(["-e", BRIDGE_AGENT]); + + try { + await backend.start(context, new AbortController().signal); + } finally { + if (previousCommand === undefined) { + delete process.env["MOSOO_ACP_FALLBACK_COMMAND"]; + } else { + process.env["MOSOO_ACP_FALLBACK_COMMAND"] = previousCommand; + } + + if (previousArgs === undefined) { + delete process.env["MOSOO_ACP_FALLBACK_ARGS"]; + } else { + process.env["MOSOO_ACP_FALLBACK_ARGS"] = previousArgs; + } + } + + return { + backend, + context, + async destroy() { + await backend.stop(context, "test cleanup", new AbortController().signal).catch(() => {}); + await logger.destroy(); + await rm(root, { force: true, recursive: true }); + }, + events: publishedEvents, + }; +} + +describe("ACP early session updates", () => { + test("completes a prompt when the agent notifies before the session/new response", async () => { + const harness = await createBridgeHarness("match"); + + try { + await expect( + harness.backend.handleInput( + harness.context, + { text: "Reply exactly: OPENCLAW-COPILOT-E2E" }, + DRIVER_TEST_IDS.runId as RunId, + ), + ).resolves.toBeUndefined(); + + const kinds = harness.events.map((event) => event.kind); + expect(kinds).toContain("session.created"); + expect(kinds).toContain("run.completed"); + const delta = harness.events.find((event) => event.kind === "message.delta"); + expect((delta?.payload as { contentDelta?: string } | undefined)?.contentDelta).toBe( + "OPENCLAW-COPILOT-E2E", + ); + // The deferred pre-registration updates are applied, not lost. + expect(kinds).toContain("session.info.updated"); + expect(kinds).toContain("session.commands.updated"); + } finally { + await harness.destroy(); + } + }); + + test("drops a pre-registration update for a foreign session without failing the turn", async () => { + const harness = await createBridgeHarness("foreign"); + + try { + await expect( + harness.backend.handleInput( + harness.context, + { text: "Reply exactly: OPENCLAW-COPILOT-E2E" }, + DRIVER_TEST_IDS.runId as RunId, + ), + ).resolves.toBeUndefined(); + + const kinds = harness.events.map((event) => event.kind); + expect(kinds).toContain("run.completed"); + expect(kinds).not.toContain("session.info.updated"); + } finally { + await harness.destroy(); + } + }); +}); From caabd10b5a5284e571e44dbdfb82265ba19a81c8 Mon Sep 17 00:00:00 2001 From: Chen Xiefan Date: Fri, 28 Aug 2026 15:08:31 +0800 Subject: [PATCH 2/2] fix(acp): observe process exit independently of stdio Stdio "close" is not a reliable exit signal for the ACP agent process: the transport keeps the child's stdout locked behind web-stream readers, so "close" can stay pending after the process is gone, and a descendant holding inherited pipes delays it indefinitely. stopAcpAgentProcess then reported "ACP agent process did not exit after force kill." for a child that was already dead, failing otherwise-successful turns. Register both "exit" and "close", short-circuit on recorded exit metadata, and re-check it after a timed-out wait. Port of f3ccdb2d6f30 from the main line onto the production lineage. --- src/runtimes/acp/acp-agent-process.ts | 38 ++++++++++++++++----------- tests/acp-agent-process.test.ts | 26 ++++++++++++------ 2 files changed, 41 insertions(+), 23 deletions(-) diff --git a/src/runtimes/acp/acp-agent-process.ts b/src/runtimes/acp/acp-agent-process.ts index 2d9b090..14e7066 100644 --- a/src/runtimes/acp/acp-agent-process.ts +++ b/src/runtimes/acp/acp-agent-process.ts @@ -15,7 +15,7 @@ export type AcpAgentProcess = ChildProcessByStdio; const ACP_AGENT_EXIT_TIMEOUT_MS = 1_500; const ACP_AGENT_FORCE_KILL_TIMEOUT_MS = 500; -const agentProcessCloseTasks = new WeakMap>(); +const agentProcessExitTasks = new WeakMap>(); 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((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(); + 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 { - 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; diff --git a/tests/acp-agent-process.test.ts b/tests/acp-agent-process.test.ts index baa0a76..9be42d0 100644 --- a/tests/acp-agent-process.test.ts +++ b/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) {