diff --git a/cli/src/agent/backends/acp/AcpStdioTransport.test.ts b/cli/src/agent/backends/acp/AcpStdioTransport.test.ts index 7fe00b06..c800ee85 100644 --- a/cli/src/agent/backends/acp/AcpStdioTransport.test.ts +++ b/cli/src/agent/backends/acp/AcpStdioTransport.test.ts @@ -5,6 +5,12 @@ const guard = vi.hoisted(() => ({ unregister: vi.fn() })); +const spawnState = vi.hoisted(() => ({ + exitHandlers: [] as Array<(code: number | null, signal: NodeJS.Signals | null) => void>, + stdinWrite: vi.fn<(chunk: string) => boolean>(() => true), + exitCode: null as number | null +})); + vi.mock('./agentCliGuard', () => ({ registerActiveAcpTransport: guard.register, unregisterActiveAcpTransport: guard.unregister @@ -12,8 +18,12 @@ vi.mock('./agentCliGuard', () => ({ vi.mock('node:child_process', () => ({ spawn: vi.fn(() => { + spawnState.exitHandlers = []; const handlers = new Map void>>(); const proc = { + get exitCode() { + return spawnState.exitCode; + }, stdout: { setEncoding: vi.fn(), on: vi.fn((event: string, handler: (...args: unknown[]) => void) => { @@ -26,12 +36,15 @@ vi.mock('node:child_process', () => ({ handlers.set(`stderr:${event}`, [...(handlers.get(`stderr:${event}`) ?? []), handler]); }) }, - stdin: { end: vi.fn(), write: vi.fn() }, + stdin: { + end: vi.fn(), + write: (chunk: string) => spawnState.stdinWrite(chunk) + }, on: vi.fn((event: string, handler: (...args: unknown[]) => void) => { - handlers.set(`proc:${event}`, [...(handlers.get(`proc:${event}`) ?? []), handler]); if (event === 'exit') { - queueMicrotask(() => handler(0, null)); + spawnState.exitHandlers.push(handler as (code: number | null, signal: NodeJS.Signals | null) => void); } + handlers.set(`proc:${event}`, [...(handlers.get(`proc:${event}`) ?? []), handler]); }), kill: vi.fn() }; @@ -45,6 +58,10 @@ describe('AcpStdioTransport agent CLI guard', () => { afterEach(() => { guard.register.mockClear(); guard.unregister.mockClear(); + spawnState.stdinWrite.mockReset(); + spawnState.stdinWrite.mockReturnValue(true); + spawnState.exitCode = null; + spawnState.exitHandlers = []; }); test('registers cross-process guard only for Cursor agent command', async () => { @@ -64,3 +81,39 @@ describe('AcpStdioTransport agent CLI guard', () => { } }); }); + +describe('AcpStdioTransport closed stdin writes', () => { + afterEach(() => { + spawnState.stdinWrite.mockReset(); + spawnState.stdinWrite.mockReturnValue(true); + spawnState.exitCode = null; + spawnState.exitHandlers = []; + }); + + test('rejects new requests after the ACP process exits instead of throwing from stdin.write', async () => { + const transport = new AcpStdioTransport({ command: 'gemini' }); + spawnState.exitCode = 1; + spawnState.stdinWrite.mockImplementation(() => { + throw new Error('WritableIterable is closed'); + }); + + for (const handler of spawnState.exitHandlers) { + handler(1, null); + } + + await expect(transport.sendRequest('session/new')).rejects.toThrow( + 'ACP process exited (code=1, signal=null)' + ); + expect(() => transport.sendNotification('session/cancel', {})).not.toThrow(); + }); + + test('rejects pending requests when stdin.write throws', async () => { + spawnState.stdinWrite.mockImplementation(() => { + throw new Error('WritableIterable is closed'); + }); + + const transport = new AcpStdioTransport({ command: 'gemini' }); + await expect(transport.sendRequest('initialize')).rejects.toThrow('WritableIterable is closed'); + await expect(transport.sendRequest('session/new')).rejects.toThrow('WritableIterable is closed'); + }); +}); diff --git a/cli/src/agent/backends/acp/AcpStdioTransport.ts b/cli/src/agent/backends/acp/AcpStdioTransport.ts index 06d3e033..e157bc82 100644 --- a/cli/src/agent/backends/acp/AcpStdioTransport.ts +++ b/cli/src/agent/backends/acp/AcpStdioTransport.ts @@ -63,6 +63,8 @@ export class AcpStdioTransport { private nextId = 1; private protocolError: Error | null = null; private guardReleased = false; + private closed = false; + private closeError: Error | null = null; constructor(options: { command: string; @@ -94,14 +96,14 @@ export class AcpStdioTransport { this.releaseAgentCliGuard(); const message = `ACP process exited (code=${code ?? 'null'}, signal=${signal ?? 'null'})`; logger.debug(message); - this.rejectAllPending(new Error(message)); + this.markClosed(new Error(message)); }); this.process.on('error', (error) => { this.releaseAgentCliGuard(); logger.debug('[ACP] Process error', error); const message = error instanceof Error ? error.message : String(error); - this.rejectAllPending(new Error( + this.markClosed(new Error( `Failed to spawn ${options.command}: ${message}. Is it installed and on PATH?`, { cause: error } )); @@ -124,6 +126,10 @@ export class AcpStdioTransport { static readonly DEFAULT_TIMEOUT_MS = 120_000; async sendRequest(method: string, params?: unknown, options?: { timeoutMs?: number }): Promise { + if (this.closed) { + return Promise.reject(this.closeError ?? new Error('ACP transport is closed')); + } + const id = this.nextId++; const payload: JsonRpcRequest = { jsonrpc: '2.0', @@ -167,6 +173,10 @@ export class AcpStdioTransport { } sendNotification(method: string, params?: unknown): void { + if (this.closed) { + return; + } + const payload: JsonRpcNotification = { jsonrpc: '2.0', method, @@ -179,7 +189,7 @@ export class AcpStdioTransport { this.process.stdin.end(); await killProcessByChildProcess(this.process); this.releaseAgentCliGuard(); - this.rejectAllPending(new Error('ACP transport closed')); + this.markClosed(new Error('ACP transport closed')); } private releaseAgentCliGuard(): void { @@ -302,8 +312,27 @@ export class AcpStdioTransport { } private writePayload(payload: JsonRpcRequest | JsonRpcNotification | JsonRpcResponse): void { - const serialized = JSON.stringify(payload); - this.process.stdin.write(`${serialized}\n`); + if (this.closed) { + return; + } + + try { + const serialized = JSON.stringify(payload); + this.process.stdin.write(`${serialized}\n`); + } catch (error) { + const writeError = error instanceof Error ? error : new Error(String(error)); + this.markClosed(writeError); + } + } + + private markClosed(error: Error): void { + if (this.closed) { + return; + } + + this.closed = true; + this.closeError = error; + this.rejectAllPending(error); } private rejectAllPending(error: Error): void { diff --git a/cli/src/cursor/runCursor.test.ts b/cli/src/cursor/runCursor.test.ts new file mode 100644 index 00000000..0893e0e9 --- /dev/null +++ b/cli/src/cursor/runCursor.test.ts @@ -0,0 +1,121 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const mockCursorSession = vi.hoisted(() => ({ + setPermissionMode: vi.fn(), + setModel: vi.fn(), + pushKeepAlive: vi.fn(), + stopKeepAlive: vi.fn(), + canApplyModelConfig: vi.fn(() => false) +})); + +const harness = vi.hoisted(() => ({ + bootstrapArgs: [] as Array>, + loopArgs: [] as Array>, + loopError: null as Error | null, + session: { + onUserMessage: vi.fn(), + onCancelQueuedMessage: vi.fn(), + sendSessionEvent: vi.fn(), + rpcHandlerManager: { + registerHandler: vi.fn() + } + }, + metadata: { sessionId: 'cursor-session-1' } +})); + +vi.mock('@/agent/sessionFactory', () => ({ + bootstrapSession: vi.fn(async (options: Record) => { + harness.bootstrapArgs.push(options); + return { + api: {}, + session: harness.session, + metadata: harness.metadata + }; + }), + bootstrapExistingSession: vi.fn(async (options: Record) => { + harness.bootstrapArgs.push(options); + return { + api: {}, + session: harness.session, + metadata: harness.metadata + }; + }) +})); + +vi.mock('./loop', () => ({ + loop: vi.fn(async (options: Record) => { + harness.loopArgs.push(options); + if (harness.loopError) { + throw harness.loopError; + } + const onSessionReady = options.onSessionReady as ((session: unknown) => void) | undefined; + onSessionReady?.(mockCursorSession); + }) +})); + +vi.mock('@/claude/registerKillSessionHandler', () => ({ + registerKillSessionHandler: vi.fn() +})); + +const lifecycleMock = vi.hoisted(() => ({ + registerProcessHandlers: vi.fn(), + cleanupAndExit: vi.fn(async () => {}), + markCrash: vi.fn(), + setExitCode: vi.fn(), + setArchiveReason: vi.fn(), + setSessionEndReason: vi.fn() +})); + +vi.mock('@/agent/runnerLifecycle', () => ({ + createModeChangeHandler: vi.fn(() => vi.fn()), + createRunnerLifecycle: vi.fn(() => lifecycleMock), + setControlledByUser: vi.fn() +})); + +vi.mock('@/agent/localHandoff', () => ({ + registerLocalHandoffHandler: vi.fn() +})); + +vi.mock('@/ui/logger', () => ({ + logger: { + debug: vi.fn() + } +})); + +vi.mock('@/utils/attachmentFormatter', () => ({ + formatMessageWithAttachments: vi.fn((text: string) => text) +})); + +import { runCursor } from './runCursor'; + +describe('runCursor', () => { + beforeEach(() => { + harness.bootstrapArgs.length = 0; + harness.loopArgs.length = 0; + harness.loopError = null; + harness.session.onUserMessage.mockReset(); + harness.session.onCancelQueuedMessage.mockReset(); + harness.session.sendSessionEvent.mockReset(); + harness.session.rpcHandlerManager.registerHandler.mockReset(); + lifecycleMock.registerProcessHandlers.mockClear(); + lifecycleMock.cleanupAndExit.mockClear(); + lifecycleMock.markCrash.mockClear(); + lifecycleMock.setExitCode.mockClear(); + lifecycleMock.setArchiveReason.mockClear(); + lifecycleMock.setSessionEndReason.mockClear(); + }); + + it('surfaces loop-level ACP failures to the web UI before archiving', async () => { + harness.loopError = new Error('WritableIterable is closed'); + + await runCursor({ startedBy: 'runner' }); + + expect(harness.session.sendSessionEvent).toHaveBeenCalledWith({ + type: 'message', + message: 'Cursor Agent failed: WritableIterable is closed' + }); + expect(lifecycleMock.markCrash).toHaveBeenCalledWith(harness.loopError); + expect(lifecycleMock.setSessionEndReason).not.toHaveBeenCalledWith('completed'); + expect(lifecycleMock.cleanupAndExit).toHaveBeenCalled(); + }); +}); diff --git a/cli/src/cursor/runCursor.ts b/cli/src/cursor/runCursor.ts index d861b133..f5508f34 100644 --- a/cli/src/cursor/runCursor.ts +++ b/cli/src/cursor/runCursor.ts @@ -186,6 +186,11 @@ export async function runCursor(opts: { }); } catch (error) { crashed = true; + const errMsg = error instanceof Error ? error.message : String(error); + session.sendSessionEvent({ + type: 'message', + message: `Cursor Agent failed: ${errMsg}` + }); lifecycle.markCrash(error); logger.debug('[cursor] Loop error:', error); } finally {