From 1502b7f90f77aaecdac5908159a6cdf6aa1d65fb Mon Sep 17 00:00:00 2001 From: Dominic <33105147+DAnaeto@users.noreply.github.com> Date: Tue, 3 Nov 2026 05:08:51 -0500 Subject: [PATCH] fix(acp): fail a prompt to a disconnected ACP agent instead of hanging (#8406) When the ACP agent process exited while idle, the session kept its dead connection and accepted the next prompt. The ACP SDK logs the failed write but never settles the prompt request, so the turn stayed running forever with no error. startTurn now refuses a connection whose signal has aborted, and a turn whose transport closes before its response fails with "ACP agent disconnected" (with the exit code or signal when known). A response that was already delivered still completes the turn, and a crash that the exit handler reports keeps its "exited unexpectedly" error. Fixes #7070 --- .../server/agent/providers/acp-agent.test.ts | 290 +++++++++++++++++- .../src/server/agent/providers/acp-agent.ts | 48 ++- 2 files changed, 324 insertions(+), 14 deletions(-) diff --git a/packages/server/src/server/agent/providers/acp-agent.test.ts b/packages/server/src/server/agent/providers/acp-agent.test.ts index 7e50b32cf5..ea0c1702e9 100644 --- a/packages/server/src/server/agent/providers/acp-agent.test.ts +++ b/packages/server/src/server/agent/providers/acp-agent.test.ts @@ -1,5 +1,5 @@ import { type ChildProcess, type ChildProcessWithoutNullStreams } from "node:child_process"; -import { EventEmitter } from "node:events"; +import { EventEmitter, once } from "node:events"; import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { createRequire } from "node:module"; import { tmpdir } from "node:os"; @@ -99,13 +99,34 @@ describe("buildACPClientCapabilities", () => { interface ACPSessionInternals { sessionId: string | null; - connection: { prompt: (...args: unknown[]) => Promise }; + connection: { + prompt: (...args: unknown[]) => Promise; + signal: AbortSignal; + }; activeForegroundTurnId: string | null; configOptions: SessionConfigOption[]; translateSessionUpdate(update: SessionUpdate): AgentStreamEvent[]; acpMcpServers(): unknown[]; } +function openPromptConnection( + prompt: (...args: unknown[]) => Promise, +): ACPSessionInternals["connection"] { + return { + prompt: prompt as ACPSessionInternals["connection"]["prompt"], + signal: new AbortController().signal, + }; +} + +function terminalEvents(events: AgentStreamEvent[]): AgentStreamEvent[] { + return events.filter( + (event) => + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled", + ); +} + interface ACPModelSelectionInternals { sessionId: string | null; connection: { @@ -3053,7 +3074,7 @@ describe("ACPAgentSession", () => { ); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event as { type: string; turnId?: string }); @@ -3091,7 +3112,7 @@ describe("ACPAgentSession", () => { ); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event); @@ -3140,7 +3161,7 @@ describe("ACPAgentSession", () => { const prompt = vi.fn(() => new Promise(() => {})); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event); @@ -3167,7 +3188,7 @@ describe("ACPAgentSession", () => { const prompt = vi.fn(() => new Promise(() => {})); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event); @@ -3211,7 +3232,7 @@ describe("ACPAgentSession", () => { ); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event); @@ -3272,7 +3293,7 @@ describe("ACPAgentSession", () => { const prompt = vi.fn(() => new Promise(() => {})); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event); @@ -3311,7 +3332,7 @@ describe("ACPAgentSession", () => { const prompt = vi.fn(() => new Promise(() => {})); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event); @@ -3364,7 +3385,7 @@ describe("ACPAgentSession", () => { const prompt = vi.fn(() => new Promise(() => {})); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event); @@ -3413,7 +3434,7 @@ describe("ACPAgentSession", () => { ); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => { events.push(event as { type: string; turnId?: string; error?: string }); @@ -3446,7 +3467,7 @@ describe("ACPAgentSession", () => { ); asInternals(session).sessionId = "session-1"; - asInternals(session).connection = { prompt }; + asInternals(session).connection = openPromptConnection(prompt); session.subscribe((event) => events.push(event)); const { turnId } = await session.startTurn([ @@ -3551,6 +3572,148 @@ describe("ACPAgentSession", () => { error: expect.not.stringContaining("[object Object]"), }); }); + + function fakeACPAgent(prompt: Agent["prompt"]): Agent { + return { + async initialize() { + return { protocolVersion: PROTOCOL_VERSION, agentCapabilities: {} }; + }, + async newSession() { + return { sessionId: "session-1" }; + }, + prompt, + async authenticate() {}, + async cancel() {}, + }; + } + + async function connectInMemoryACPAgent( + agent: Agent, + options: { closeAfterPromptResponse?: boolean } = {}, + ): Promise<{ + agentConnection: AgentSideConnection; + connection: ClientSideConnection; + sessionId: string; + closeTransport: () => Promise; + }> { + const clientToAgent = new TransformStream(); + const agentToClient = new TransformStream(); + const agentOutput = agentToClient.writable.getWriter(); + const decoder = new TextDecoder(); + const agentConnection = new AgentSideConnection( + () => agent, + ndJsonStream( + new WritableStream({ + async write(chunk) { + await agentOutput.write(chunk); + if (options.closeAfterPromptResponse && decoder.decode(chunk).includes("stopReason")) { + await agentOutput.close(); + } + }, + }), + clientToAgent.readable, + ), + ); + const connection = new ClientSideConnection( + () => ({ + async requestPermission() { + return { outcome: { outcome: "cancelled" } }; + }, + async sessionUpdate() {}, + }), + ndJsonStream(clientToAgent.writable, agentToClient.readable), + ); + await connection.initialize({ + protocolVersion: PROTOCOL_VERSION, + clientCapabilities: {}, + clientInfo: { name: "Paseo test", version: "dev" }, + }); + const { sessionId } = await connection.newSession({ + cwd: "/tmp/paseo-acp-test", + mcpServers: [], + }); + return { + agentConnection, + connection, + sessionId, + closeTransport: () => agentOutput.close(), + }; + } + + test("startTurn fails instead of hanging when the ACP connection has closed", async () => { + const agent = await connectInMemoryACPAgent( + fakeACPAgent(async () => ({ stopReason: "end_turn" })), + ); + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + asInternals(session).sessionId = agent.sessionId; + asInternals(session).connection = agent.connection; + + await agent.closeTransport(); + await agent.connection.closed; + + await expect(session.startTurn("second")).rejects.toThrow("ACP agent disconnected"); + expect(events.some((event) => event.type === "turn_started")).toBe(false); + expect(asInternals(session).activeForegroundTurnId).toBeNull(); + }, 2_000); + + test("a prompt whose ACP transport closes mid-turn fails the turn", async () => { + let promptReceived!: () => void; + const received = new Promise((resolve) => { + promptReceived = resolve; + }); + const agent = await connectInMemoryACPAgent( + fakeACPAgent(() => { + promptReceived(); + return new Promise(() => {}); + }), + ); + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + asInternals(session).sessionId = agent.sessionId; + asInternals(session).connection = agent.connection; + + const { turnId } = await session.startTurn("hello"); + await received; + await agent.closeTransport(); + + await vi.waitFor( + () => + expect(terminalEvents(events)).toEqual([ + expect.objectContaining({ + type: "turn_failed", + turnId, + error: "ACP agent disconnected", + }), + ]), + { timeout: 1_000 }, + ); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect(terminalEvents(events)).toHaveLength(1); + expect(asInternals(session).activeForegroundTurnId).toBeNull(); + }, 3_000); + + test("a turn answered just before its ACP transport closes still completes", async () => { + const agent = await connectInMemoryACPAgent( + fakeACPAgent(async () => ({ stopReason: "end_turn" })), + { closeAfterPromptResponse: true }, + ); + const session = createSession(); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + asInternals(session).sessionId = agent.sessionId; + asInternals(session).connection = agent.connection; + + const { turnId } = await session.startTurn("hello"); + await agent.connection.closed; + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(terminalEvents(events)).toEqual([ + expect.objectContaining({ type: "turn_completed", turnId }), + ]); + }, 3_000); }); interface ACPCloseInternals { @@ -4490,6 +4653,109 @@ describe("ACPAgentSession close() with an unresponsive provider", () => { }, 15_000); }); +const EXITING_ACP_AGENT = ` +import { Readable, Writable } from "node:stream"; +const { AgentSideConnection, PROTOCOL_VERSION, ndJsonStream } = await import(process.env.ACP_SDK_URL); +new AgentSideConnection( + () => ({ + async initialize() { + return { protocolVersion: PROTOCOL_VERSION, agentCapabilities: {} }; + }, + async newSession() { + return { sessionId: "exiting-session" }; + }, + async authenticate() {}, + async cancel() {}, + async prompt({ prompt }) { + if (prompt[0]?.text === "crash") { + process.stderr.write("fixture crash\\n"); + process.exit(3); + } + setTimeout(() => process.exit(0), 100); + return { stopReason: "end_turn" }; + }, + }), + ndJsonStream(Writable.toWeb(process.stdout), Readable.toWeb(process.stdin)), +); +`; + +describe("ACPAgentSession after its provider process exits", () => { + let dir: string | null = null; + let session: ACPAgentSession | null = null; + + afterEach(async () => { + await session?.close(); + session = null; + if (dir) await rm(dir, { recursive: true, force: true }); + dir = null; + }); + + async function startExitingProvider(): Promise { + dir = await mkdtemp(path.join(tmpdir(), "paseo-acp-exiting-")); + const agentScript = path.join(dir, "agent.mjs"); + await writeFile(agentScript, EXITING_ACP_AGENT); + session = new ACPAgentSession( + { provider: "exiting-acp", cwd: dir }, + { + provider: "exiting-acp", + logger: createTestLogger(), + defaultCommand: [process.execPath, agentScript], + defaultModes: [], + capabilities: { supportsStreaming: true, supportsSessionPersistence: true }, + launchEnv: { + ACP_SDK_URL: pathToFileURL( + createRequire(import.meta.url).resolve("@agentclientprotocol/sdk"), + ).href, + }, + }, + ); + await session.initializeNewSession(); + return session; + } + + test("startTurn after the provider exits while idle fails instead of hanging", async () => { + const provider = await startExitingProvider(); + const events: AgentStreamEvent[] = []; + provider.subscribe((event) => events.push(event)); + const internals = asInternals<{ child: ChildProcess; connection: ClientSideConnection }>( + provider, + ); + const { child, connection } = internals; + + const { turnId } = await provider.startTurn("first"); + await vi.waitFor(() => + expect(terminalEvents(events)).toEqual([ + expect.objectContaining({ type: "turn_completed", turnId }), + ]), + ); + if (child.exitCode === null) await once(child, "exit"); + await connection.closed; + + await expect(provider.startTurn("second")).rejects.toThrow( + "ACP agent disconnected (exit code 0)", + ); + expect(terminalEvents(events)).toHaveLength(1); + }, 10_000); + + test("a provider that crashes mid-turn still reports its exit status", async () => { + const provider = await startExitingProvider(); + const events: AgentStreamEvent[] = []; + provider.subscribe((event) => events.push(event)); + + const { turnId } = await provider.startTurn("crash"); + await vi.waitFor(() => expect(terminalEvents(events)).toHaveLength(1), { timeout: 3_000 }); + await new Promise((resolve) => setTimeout(resolve, 50)); + + expect(terminalEvents(events)).toEqual([ + expect.objectContaining({ + type: "turn_failed", + turnId, + error: "ACP agent exited unexpectedly (3)", + }), + ]); + }, 10_000); +}); + // An ACP agent that advertises two models (or none, with ACP_ADVERTISE_MODELS=none) but, // like Cline, runs any model id it is given. const PERMISSIVE_MODEL_ACP_AGENT = ` diff --git a/packages/server/src/server/agent/providers/acp-agent.ts b/packages/server/src/server/agent/providers/acp-agent.ts index 0f7411d8f7..09662aaa1c 100644 --- a/packages/server/src/server/agent/providers/acp-agent.ts +++ b/packages/server/src/server/agent/providers/acp-agent.ts @@ -1,10 +1,12 @@ import { ACPProviderOptionsSchema } from "./acp-options.js"; import { type ChildProcess, type ChildProcessWithoutNullStreams } from "node:child_process"; import { randomUUID } from "node:crypto"; +import { once } from "node:events"; import fs from "node:fs/promises"; import { homedir } from "node:os"; import path from "node:path"; import { Readable, Writable } from "node:stream"; +import { setImmediate } from "node:timers/promises"; import { terminateWithTreeKill } from "../../../utils/tree-kill.js"; import type { ProcessTerminator } from "../../../utils/tree-kill.js"; @@ -166,6 +168,16 @@ function rejectOnProcessFailure(child: ChildProcess, stderrChunks: string[]): Pr }); } +function describeACPDisconnect(child: ChildProcess | null): string { + if (child?.signalCode) { + return `ACP agent disconnected (signal ${child.signalCode})`; + } + if (child?.exitCode != null) { + return `ACP agent disconnected (exit code ${child.exitCode})`; + } + return "ACP agent disconnected"; +} + function isRecord(value: unknown): value is Record { return value != null && typeof value === "object" && !Array.isArray(value); } @@ -323,6 +335,7 @@ const ACP_PROBE_CLOSE_TIMEOUT_MS = 2_000; // request when the provider never answers, so these calls must be bounded // or the provider process is never terminated. const ACP_CLOSE_REQUEST_TIMEOUT_MS = 2_000; +const ACP_DISCONNECT_EXIT_GRACE_MS = 1_000; const ACP_IMPORT_HISTORY_LOAD_TIMEOUT_MS = 30_000; const ACP_IMPORT_HISTORY_BUDGET_MS = 60_000; @@ -1978,6 +1991,9 @@ export class ACPAgentSession implements AgentSession, ACPClient { if (!this.connection || !this.sessionId) { throw new Error(`${this.provider} session is not initialized`); } + if (this.connection.signal.aborted) { + throw new Error(describeACPDisconnect(this.child)); + } if (this.activeForegroundTurnId) { throw new Error("A foreground turn is already active"); } @@ -1992,7 +2008,11 @@ export class ACPAgentSession implements AgentSession, ACPClient { this.pushEvent({ type: "turn_started", provider: this.provider, turnId }); this.emitSubmittedUserMessage(prompt, messageId, turnId, options?.clientMessageId); - void this.connection + const connection = this.connection; + const child = this.child; + const failOnDisconnect = () => void this.failDisconnectedTurn(turnId, child); + connection.signal.addEventListener("abort", failOnDisconnect, { once: true }); + void connection .prompt({ sessionId: this.sessionId, messageId, @@ -2012,11 +2032,35 @@ export class ACPAgentSession implements AgentSession, ACPClient { diagnostic: this.collectDiagnostic(summary.diagnostic ?? summary.message), turnId, }); - }); + }) + .finally(() => connection.signal.removeEventListener("abort", failOnDisconnect)); return { turnId }; } + // The SDK aborts the connection without rejecting the pending prompt, and before the child's + // exit event, so wait for the exit status before failing the turn. + private async failDisconnectedTurn(turnId: string, child: ChildProcess | null): Promise { + await setImmediate(); + if (child && child.exitCode === null && child.signalCode === null) { + await once(child, "exit", { + signal: AbortSignal.timeout(ACP_DISCONNECT_EXIT_GRACE_MS), + }).catch(() => undefined); + } + if (this.closed || this.activeForegroundTurnId !== turnId) { + return; + } + this.synthesizeCanceledToolCalls(); + const error = describeACPDisconnect(child); + this.finishTurn({ + type: "turn_failed", + provider: this.provider, + error, + diagnostic: this.collectDiagnostic(error), + turnId, + }); + } + subscribe(callback: (event: AgentStreamEvent) => void): () => void { this.subscribers.add(callback); if (this.sessionId) {