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) {