diff --git a/src/main/acp/acp-agent-connection.test.ts b/src/main/acp/acp-agent-connection.test.ts new file mode 100644 index 000000000000..46fad1e969f2 --- /dev/null +++ b/src/main/acp/acp-agent-connection.test.ts @@ -0,0 +1,281 @@ +import { EventEmitter } from 'node:events' +import { PassThrough } from 'node:stream' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { spawnProcess } from '../../shared/child-process/run-process' +import { PROVIDER_SUPERVISOR_MAX_STOP_MS } from '../provider-process/provider-process-supervisor' +import { ROOT_ONLY_GRACEFUL_EXIT_MS } from '../provider-process/provider-process-close' +import type { terminateProviderProcessTree } from '../provider-process/provider-process-teardown' +import { + createAcpAgentConnection, + type AcpAgentConnection, + type AcpAgentConnectionOptions +} from './acp-agent-connection' +import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support' + +const teardown = vi.hoisted(() => + vi.fn(async () => 'unverifiable') +) +vi.mock('../provider-process/provider-process-teardown', () => ({ + terminateProviderProcessTree: teardown +})) + +const opened: { connection: AcpAgentConnection; exit: () => void }[] = [] +const grace = + process.platform === 'win32' ? ROOT_ONLY_GRACEFUL_EXIT_MS : PROVIDER_SUPERVISOR_MAX_STOP_MS +const start = { cwd: '/execution-host/folder', mcpServers: [] } +const prompt = [{ type: 'text', text: 'hello' }] as const + +function fixture( + options: AcpAgentConnectionOptions = {}, + behavior: { pid?: number | null; exitOnEnd?: boolean } = {} +) { + const agent = new AcpScriptedAgent() + const child = Object.assign(new EventEmitter(), { + pid: behavior.pid === null ? undefined : (behavior.pid ?? 9_999_999), + stdout: agent.stdout, + stdin: agent.stdin, + stderr: new PassThrough(), + kill: vi.fn(() => true) + }) + const spawn = vi.fn(() => { + // oxlint-disable-next-line typescript/consistent-type-assertions -- SAFETY: The supervised connection reads events, pid, piped stdio and kill; this fixture supplies each. + return child as unknown as ReturnType + }) + agent.on('initialize', (frame) => agent.reply(frame, { protocolVersion: 1 })) + agent.on('session/new', (frame) => agent.reply(frame, { sessionId: 'session-1' })) + if (behavior.exitOnEnd !== false) { + child.stdin.once('finish', () => child.emit('exit', 0, null)) + } + const connection = createAcpAgentConnection( + { + command: 'fixture-acp-agent', + args: ['--acp'], + cwd: start.cwd, + env: { ACP_ACCOUNT_HOME: '/host/account', STRIPPED: 'overlay' }, + envToDelete: ['STRIPPED'] + }, + options, + spawn + ) + opened.push({ connection, exit: () => child.emit('exit', 0, null) }) + return { agent, child, spawn, connection } +} + +afterEach(async () => { + for (const { connection, exit } of opened.splice(0)) { + exit() + await connection.close() + } + teardown.mockReset().mockResolvedValue('unverifiable') + vi.useRealTimers() +}) + +describe('ACP process-owning connection', () => { + it('spawns through the supervisor with the host launch and exposes typed session calls', async () => { + const { connection, spawn, child, agent } = fixture() + await connection.spawned + expect(connection.pid).toBe(child.pid) + expect(connection.rootVerdict).toBe('live') + expect(connection.lastCloseResult).toBeNull() + expect(spawn).toHaveBeenCalledExactlyOnceWith( + expect.objectContaining({ + cwd: start.cwd, + env: expect.objectContaining({ ACP_ACCOUNT_HOME: '/host/account' }), + stdio: ['pipe', 'pipe', 'pipe'] + }) + ) + expect(spawn.mock.calls[0][0].env).not.toHaveProperty('STRIPPED') + expect(await connection.start(start)).toMatchObject({ sessionId: 'session-1' }) + agent.on('session/prompt', (frame) => agent.reply(frame, { stopReason: 'end_turn' })) + expect(await connection.prompt([...prompt])).toEqual({ stopReason: 'end_turn' }) + connection.pauseReading() + expect(child.stdout.isPaused()).toBe(true) + connection.resumeReading() + expect(child.stdout.isPaused()).toBe(false) + expect(await connection.close()).toBe(true) + expect(connection.lastCloseResult).toEqual({ root: 'exited', tree: null }) + }) + + it('settles prompts and permission signals on proven exit with stdout still open', async () => { + const onExit = vi.fn() + const onClose = vi.fn() + const permission = deferred() + const { connection, child, agent } = fixture({ + onExit, + onClose, + onPermission: (_request, context) => { + permission.resolve(context.signal) + return new Promise(() => {}) + } + }) + await connection.start(start) + const rejected = expect(connection.prompt([...prompt])).rejects.toThrow('provider failed') + void agent.request('approval', 'session/request_permission', { + sessionId: 'session-1', + toolCall: { toolCallId: 'tool-1' }, + options: [{ optionId: 'allow', kind: 'allow_once', name: 'Allow' }] + }) + const signal = await permission.promise + child.stderr.write('provider failed\n') + child.emit('exit', 7, null) + child.emit('close', 7, null) + await rejected + expect(signal.aborted).toBe(true) + expect(child.stdout.readableEnded).toBe(false) + expect(connection.exited).toBe(true) + expect(onClose).not.toHaveBeenCalled() + expect(onExit).toHaveBeenCalledExactlyOnceWith(expect.any(Error), { + expected: false, + exit: { code: 7, signal: null, processless: false } + }) + const late = vi.fn() + connection.onExit(late) + expect(late).toHaveBeenCalledOnce() + await expect(connection.close()).resolves.toBe(true) + }) + + it('keeps cancel writable after stdout EOF and waits for process exit evidence', async () => { + const onClose = vi.fn() + const onExit = vi.fn() + const { connection, child, agent } = fixture({ onClose, onExit }) + await connection.start(start) + const rejected = expect(connection.prompt([...prompt])).rejects.toThrow('connection closed') + child.stdout.end() + await tick() + expect(connection.closed).toBe(false) + expect(connection.rootVerdict).toBe('live') + await connection.cancel() + expect(agent.frames.at(-1)?.method).toBe('session/cancel') + expect(onClose).not.toHaveBeenCalled() + expect(onExit).not.toHaveBeenCalled() + expect(await connection.close()).toBe(true) + await rejected + expect(onExit.mock.calls[0][1].expected).toBe(true) + }) + + it('reports broken stdin immediately but reports exit only after the host observes it', async () => { + vi.useFakeTimers() + const onClose = vi.fn() + const onExit = vi.fn() + const { connection, child } = fixture({ onClose, onExit }, { exitOnEnd: false }) + await connection.start(start) + const rejected = expect(connection.prompt([...prompt])).rejects.toThrow('broken pipe') + child.stdin.emit('error', new Error('broken pipe')) + child.emit('close', 0, null) + await rejected + expect(connection.closed).toBe(true) + expect(connection.rootVerdict).toBe('live') + expect(onClose).toHaveBeenCalledOnce() + expect(onExit).not.toHaveBeenCalled() + child.emit('exit', 1, null) + await vi.advanceTimersByTimeAsync(0) + expect(connection.rootVerdict).toBe('exited') + expect(onExit.mock.calls[0][1].expected).toBe(false) + expect(vi.getTimerCount()).toBe(0) + }) + + it('joins physical close and retries unproven exit without spawning a replacement', async () => { + vi.useFakeTimers() + const onExit = vi.fn() + const { connection, child, spawn } = fixture({ onExit }, { exitOnEnd: false }) + const first = connection.close() + const joined = connection.close() + await vi.advanceTimersByTimeAsync(grace + 1_000) + expect(await first).toBe(false) + expect(await joined).toBe(false) + expect(connection.rootVerdict).toBe('live') + expect(teardown).toHaveBeenCalledOnce() + teardown.mockImplementationOnce(async () => { + child.emit('exit', null, 'SIGKILL') + return 'unverifiable' + }) + const retry = connection.close() + await vi.advanceTimersByTimeAsync(grace + 1_000) + expect(await retry).toBe(true) + expect(connection.processTreeUnproven).toBe(true) + expect(spawn).toHaveBeenCalledOnce() + expect(onExit.mock.calls[0][1].expected).toBe(true) + expect(vi.getTimerCount()).toBe(0) + }) + + it.each(['live', 'unverifiable'] as const)( + 'retains %s tree evidence when root exits between completed close attempts', + async (tree) => { + vi.useFakeTimers() + const { connection, child } = fixture({}, { exitOnEnd: false }) + teardown.mockResolvedValueOnce(tree) + const close = connection.close() + await vi.advanceTimersByTimeAsync(grace + 1_000) + expect(await close).toBe(false) + child.emit('exit', null, 'SIGKILL') + expect(connection.processTreeUnproven).toBe(true) + expect(await connection.close()).toBe(true) + expect(connection.processTreeUnproven).toBe(true) + expect(connection.lastCloseResult).toEqual({ root: 'live', tree }) + expect(teardown).toHaveBeenCalledOnce() + } + ) + + it('isolates early and late exit subscribers and continues delivering exit', () => { + const onDiagnostic = vi.fn(() => { + throw new Error('diagnostic') + }) + const { connection, child } = fixture({ onDiagnostic }) + const failing = (): void => { + throw new Error('observer') + } + const early = vi.fn() + connection.onExit(failing) + connection.onExit(early) + expect(() => child.emit('exit', 0, null)).not.toThrow() + expect(early).toHaveBeenCalledExactlyOnceWith({ code: 0, signal: null, processless: false }) + expect(() => connection.onExit(failing)).not.toThrow() + const late = vi.fn() + connection.onExit(late) + expect(late).toHaveBeenCalledOnce() + expect(onDiagnostic).toHaveBeenCalledTimes(2) + }) + + it('settles a failed spawn but requires processless close evidence before reporting exit', async () => { + const onExit = vi.fn() + const { connection, child, agent } = fixture({ onExit }, { pid: null, exitOnEnd: false }) + agent.on('initialize', () => {}) + const rejected = expect(connection.initialize()).rejects.toThrow('ENOENT') + child.emit('error', new Error('ENOENT')) + await connection.spawned + await rejected + expect(connection.rootVerdict).toBe('unverifiable') + expect(onExit).not.toHaveBeenCalled() + child.emit('close', null, null) + expect(connection.rootVerdict).toBe('exited') + expect(onExit.mock.calls[0][1].exit.processless).toBe(true) + expect(await connection.close()).toBe(true) + }) + + it('owns stderr failure cleanup and isolates failing exit callbacks', async () => { + const { connection, child } = fixture({ + onExit: () => { + throw new Error('listener') + } + }) + await connection.start(start) + const rejected = expect(connection.prompt([...prompt])).rejects.toThrow('stderr failure') + child.stderr.destroy(new Error('stderr failure')) + await rejected + await tick() + expect(connection.exited).toBe(true) + expect(await connection.close()).toBe(true) + }) + + it('rejects invalid peer options before starting any process', () => { + const spawn = vi.fn() + expect(() => + createAcpAgentConnection( + { command: 'fixture', args: [] }, + { peer: { maxPendingRequests: 0 } }, + spawn + ) + ).toThrow('positive finite integers') + expect(spawn).not.toHaveBeenCalled() + }) +}) diff --git a/src/main/acp/acp-agent-connection.ts b/src/main/acp/acp-agent-connection.ts new file mode 100644 index 000000000000..d4f7eaf2ec25 --- /dev/null +++ b/src/main/acp/acp-agent-connection.ts @@ -0,0 +1,150 @@ +import { spawnProcess } from '../../shared/child-process/run-process' +import { + spawnManagedProviderProcess, + type ManagedProviderProcess, + type ProviderProcessExit +} from '../provider-process/managed-provider-process' +import type { ProviderProcessLaunch } from '../provider-process/provider-process-launch' +import { AcpConnectionClosedError } from './acp-errors' +import { resolveAcpPeerOptions, type AcpPeerOptions } from './acp-peer-limits' +import { AcpSessionRuntime, type AcpSessionRuntimeOptions } from './acp-session-runtime' + +export type AcpAgentConnectionOptions = Omit & { + peer?: Omit + /** Process exit evidence, including expected closes and processless spawn failures. */ + onExit?: (error: Error, context: { expected: boolean; exit: ProviderProcessExit }) => void +} + +/** The execution host owns the child and protocol lifetime; the adapter owns turns and Stop. */ +export function createAcpAgentConnection( + launch: ProviderProcessLaunch, + options: AcpAgentConnectionOptions = {}, + spawnImpl: typeof spawnProcess = spawnProcess +): AcpAgentConnection { + return new AcpAgentConnection(launch, options, spawnImpl) +} + +export class AcpAgentConnection extends AcpSessionRuntime { + private readonly managed: ManagedProviderProcess + private readonly lifecycle: { closing: boolean; error?: Error } + private readonly diagnoseLifecycle: (message: string) => void + readonly spawned: Promise + + constructor( + launch: ProviderProcessLaunch, + options: AcpAgentConnectionOptions = {}, + spawnImpl: typeof spawnProcess = spawnProcess + ) { + // Validate before spawning so invalid limits cannot leave an unowned child. + const peer = resolveAcpPeerOptions({ ...options.peer, closeOnInputEnd: false }) + const managed = spawnManagedProviderProcess(launch, { spawnImpl, site: 'acp-agent-teardown' }) + const lifecycle: { closing: boolean; error?: Error } = { closing: false } + const diagnose = (message: string): void => { + try { + options.onDiagnostic?.(message) + } catch { + /* Diagnostics cannot interrupt cleanup. */ + } + } + super(managed.child.stdout, managed.child.stdin, { + ...options, + peer, + onClose: (error) => { + lifecycle.error ??= error + if (lifecycle.closing || managed.rootVerdict === 'exited') { + return + } + void managed.close().then( + (result) => { + if (result.root !== 'exited') { + diagnose('ACP agent exit was not proven after transport failure') + } + }, + (failure: unknown) => diagnose(`ACP agent cleanup failed: ${String(failure)}`) + ) + options.onClose?.(error) + } + }) + this.managed = managed + this.lifecycle = lifecycle + this.diagnoseLifecycle = diagnose + const { child } = managed + this.spawned = new Promise((resolve) => { + if (child.pid !== undefined) { + resolve() + return + } + const done = (): void => { + child.removeListener('spawn', done) + child.removeListener('error', done) + resolve() + } + child.once('spawn', done) + child.once('error', done) + }) + child.on('error', (error) => super.close(error)) + child.stderr.on('error', (error) => super.close(error)) + managed.onExit((exit) => { + const error = + lifecycle.error ?? + new AcpConnectionClosedError(managed.stderrTail().trim() || `${launch.command} exited`) + super.close(error) + try { + options.onExit?.(error, { expected: lifecycle.closing, exit }) + } catch (failure) { + diagnose(`ACP exit listener failed: ${String(failure)}`) + } + }) + } + + get pid(): number | undefined { + return this.managed.child.pid + } + + get exited(): boolean { + return this.managed.rootVerdict === 'exited' + } + + get rootVerdict(): ManagedProviderProcess['rootVerdict'] { + return this.managed.rootVerdict + } + + /** Retained evidence from the last close attempt; a null tree means no observation. */ + get lastCloseResult(): Readonly { + return this.managed.lastCloseResult + } + + /** Reports retained cleanup uncertainty; false does not prove descendant exit. */ + get processTreeUnproven(): boolean { + const result = this.lastCloseResult + return this.exited && (result?.tree === 'unverifiable' || result?.tree === 'live') + } + + stderrTail(): string { + return this.managed.stderrTail().trim() + } + + onExit(listener: (exit: ProviderProcessExit) => void): void { + this.managed.onExit((exit) => { + try { + listener(exit) + } catch (failure) { + this.diagnoseLifecycle(`ACP exit listener failed: ${String(failure)}`) + } + }) + } + + pauseReading(): void { + this.managed.child.stdout.pause() + } + + resumeReading(): void { + this.managed.child.stdout.resume() + } + + override close(error?: Error): Promise { + this.lifecycle.closing ||= this.managed.rootVerdict !== 'exited' + super.close(error) + return this.managed.close().then((result) => result.root === 'exited') + } +} diff --git a/src/main/acp/acp-json-rpc-peer.ts b/src/main/acp/acp-json-rpc-peer.ts index b4d57e246cd5..2467cb345cce 100644 --- a/src/main/acp/acp-json-rpc-peer.ts +++ b/src/main/acp/acp-json-rpc-peer.ts @@ -11,7 +11,8 @@ import { AcpRequestTimeoutError } from './acp-errors' import { AcpIncomingRequests } from './acp-incoming-requests' -import { bounded, requestTimeout } from './acp-peer-limits' +import { requestTimeout, resolveAcpPeerOptions, type AcpPeerOptions } from './acp-peer-limits' +export type { AcpPeerOptions } from './acp-peer-limits' import { settleOversizedAcpLine } from './acp-oversized-lines' import { AcpWriteQueue } from './acp-write-queue' import { detachAcpStreamErrorHandler } from './acp-stdio-error-boundary' @@ -40,14 +41,6 @@ export type AcpPeerHandlers = { onDiagnostic?: (message: string) => void onClose?: (error: Error) => void } -export type AcpPeerOptions = { - maxLineBytes?: number - maxQueuedWriteBytes?: number - maxPendingRequests?: number - maxIncomingRequests?: number - // Caller-selected outgoing deadlines only; omitted or null means unlimited. - requestTimeoutMs?: number | null -} type Pending = { method: string resolve: (value: unknown) => void @@ -66,6 +59,7 @@ export class AcpJsonRpcPeer { private readonly maxPending: number private readonly maxIncoming: number private readonly timeoutMs: number | null + private readonly closeOnInputEnd: boolean constructor( private readonly input: Readable, @@ -73,14 +67,14 @@ export class AcpJsonRpcPeer { private readonly handlers: AcpPeerHandlers = {}, options: AcpPeerOptions = {} ) { - this.maxLineBytes = bounded(options.maxLineBytes, 16 * 1024 * 1024) - this.maxPending = bounded(options.maxPendingRequests, 128) - this.maxIncoming = bounded(options.maxIncomingRequests, 128) - this.timeoutMs = requestTimeout(options.requestTimeoutMs) - this.writer = new AcpWriteQueue( - output, - bounded(options.maxQueuedWriteBytes, 32 * 1024 * 1024), - (error) => this.close(error) + const limits = resolveAcpPeerOptions(options) + this.maxLineBytes = limits.maxLineBytes + this.maxPending = limits.maxPendingRequests + this.maxIncoming = limits.maxIncomingRequests + this.timeoutMs = limits.requestTimeoutMs + this.closeOnInputEnd = limits.closeOnInputEnd + this.writer = new AcpWriteQueue(output, limits.maxQueuedWriteBytes, (error) => + this.close(error) ) this.incoming = new AcpIncomingRequests( handlers.onRequest, @@ -104,13 +98,17 @@ export class AcpJsonRpcPeer { ) input.setEncoding('utf8') input.on('data', this.onData) - input.on('end', this.onEnd) - input.on('close', this.onEnd) + input.on('end', this.onInputEnd) + input.on('close', this.onInputEnd) input.on('error', this.onError) output.on('close', this.onEnd) output.on('finish', this.onEnd) output.on('error', this.onError) - if (input.destroyed || input.readableEnded || output.destroyed || !output.writable) { + if ( + (this.closeOnInputEnd && (input.destroyed || input.readableEnded)) || + output.destroyed || + !output.writable + ) { this.onEnd() } } @@ -178,8 +176,8 @@ export class AcpJsonRpcPeer { } this.terminalError = error this.input.removeListener('data', this.onData) - this.input.removeListener('end', this.onEnd) - this.input.removeListener('close', this.onEnd) + this.input.removeListener('end', this.onInputEnd) + this.input.removeListener('close', this.onInputEnd) detachAcpStreamErrorHandler(this.input, this.onError) this.output.removeListener('close', this.onEnd) this.output.removeListener('finish', this.onEnd) @@ -207,6 +205,11 @@ export class AcpJsonRpcPeer { } } private readonly onEnd = (): void => this.close() + private readonly onInputEnd = (): void => { + if (this.closeOnInputEnd) { + this.close() + } + } private readonly onError = (error: Error): void => this.close(error) private diagnose(message: string): void { try { diff --git a/src/main/acp/acp-peer-limits.ts b/src/main/acp/acp-peer-limits.ts index 51f98809472e..51963e4791aa 100644 --- a/src/main/acp/acp-peer-limits.ts +++ b/src/main/acp/acp-peer-limits.ts @@ -1,5 +1,26 @@ import { isSafeTimerDelayMs } from '../../shared/timer-delay' +export type AcpPeerOptions = { + maxLineBytes?: number + maxQueuedWriteBytes?: number + maxPendingRequests?: number + maxIncomingRequests?: number + requestTimeoutMs?: number | null + /** A managed connection waits for process exit rather than treating stdout EOF as exit. */ + closeOnInputEnd?: boolean +} + +export function resolveAcpPeerOptions(options: AcpPeerOptions = {}): Required { + return { + maxLineBytes: bounded(options.maxLineBytes, 16 * 1024 * 1024), + maxQueuedWriteBytes: bounded(options.maxQueuedWriteBytes, 32 * 1024 * 1024), + maxPendingRequests: bounded(options.maxPendingRequests, 128), + maxIncomingRequests: bounded(options.maxIncomingRequests, 128), + requestTimeoutMs: requestTimeout(options.requestTimeoutMs), + closeOnInputEnd: options.closeOnInputEnd ?? true + } +} + export function bounded(value: number | undefined, fallback: number): number { if (value === undefined) { return fallback diff --git a/src/main/acp/acp-permission-requests.test.ts b/src/main/acp/acp-permission-requests.test.ts index d2f15611725b..d04649beb361 100644 --- a/src/main/acp/acp-permission-requests.test.ts +++ b/src/main/acp/acp-permission-requests.test.ts @@ -99,11 +99,19 @@ describe('ACP permission requests', () => { expect(diagnostics).toEqual([`Answered ACP permission request cancelled: ${problem}`]) }) - it('answers cancelled when the cancel lands before the permission handler starts', async () => { - const asked = vi.fn((_request: RequestPermissionRequest) => new Promise(() => {})) + it('lets the owner decline a permission after cancel, before the handler starts', async () => { + let stopping = false + const asked = vi.fn((_request: RequestPermissionRequest) => ({ + outcome: stopping + ? { outcome: 'cancelled' as const } + : { outcome: 'selected' as const, optionId: 'allow' } + })) const { agent, runtime } = fixture({ onPermission: asked }) await runtime.start(startOptions) - runtime.subscribe(() => void runtime.cancel()) + runtime.subscribe(() => { + stopping = true + void runtime.cancel() + }) const answer = new Promise((resolve) => { agent.stdin.on('data', (chunk: string) => { if (chunk.includes('"id":5')) { @@ -133,15 +141,17 @@ describe('ACP permission requests', () => { .join('') ) expect(await answer).toMatchObject({ id: 5, result: { outcome: { outcome: 'cancelled' } } }) - expect(asked).not.toHaveBeenCalled() + expect(stopping).toBe(true) + expect(asked).toHaveBeenCalledOnce() }) - it('lets a turn the agent began itself ask for permission and cancels it with session/cancel', async () => { + it('keeps an autonomous permission with its owner after session/cancel', async () => { + const decision = deferred<{ outcome: { outcome: 'cancelled' } }>() const asked = deferred() const { agent, runtime } = fixture({ onPermission: (_request, context) => { asked.resolve(context.signal) - return new Promise(() => {}) + return decision.promise } }) await runtime.start(startOptions) @@ -152,7 +162,8 @@ describe('ACP permission requests', () => { }) const signal = await asked.promise await runtime.cancel() - expect(signal.aborted).toBe(true) + expect(signal.aborted).toBe(false) + decision.resolve({ outcome: { outcome: 'cancelled' } }) expect(await answer).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } }) await tick() expect(agent.frames.filter((frame) => frame.method === 'session/cancel')).toHaveLength(1) diff --git a/src/main/acp/acp-prompt-cancel.ts b/src/main/acp/acp-prompt-cancel.ts deleted file mode 100644 index 8938d10aea3d..000000000000 --- a/src/main/acp/acp-prompt-cancel.ts +++ /dev/null @@ -1,94 +0,0 @@ -import { AcpRequestTimeoutError } from './acp-errors' -import type { PromptResponse } from './generated/acp-protocol.generated' - -export type ActivePrompt = { - response: Promise - cancelling: boolean - cancelPromise?: Promise - steerCancel?: Promise -} - -/** How a cancel reaches the agent: the `session/cancel` write and the connection it rides on. */ -export type AcpCancelChannel = { - send: () => Promise - /** Cancels the agent's open requests (permissions, vendor hooks) so each gets its answer. */ - cancelIncomingRequests: () => void - close: (error: Error) => void - closed: () => boolean - timeoutMs: number -} - -/** A Stop's cancel. Sends `session/cancel` whenever the session runs: the agent may be in a turn it - * began itself. With Orca's prompt running, also waits (bounded) for that prompt to settle, and - * past the bound closes the connection, which ends the agent. Repeated asks share one wait. */ -export function cancelAcpPromptForStop( - active: ActivePrompt | undefined, - channel: AcpCancelChannel -): Promise { - channel.cancelIncomingRequests() - if (!active) { - return channel.send() - } - active.cancelling = true - active.cancelPromise ??= confirmAcpPromptCancel(active, channel) - return active.cancelPromise -} - -/** Sends `session/cancel` and waits, bounded, for Orca's prompt to settle; past the bound it closes. */ -async function confirmAcpPromptCancel( - active: ActivePrompt, - connection: AcpCancelChannel -): Promise { - let timer: ReturnType | undefined - const unconfirmed = new Promise((_resolve, reject) => { - timer = setTimeout(() => { - const error = new AcpRequestTimeoutError('session/cancel') - connection.close(error) - reject(error) - }, connection.timeoutMs) - }) - try { - await Promise.race([connection.send(), unconfirmed]).catch((error) => { - // Only a cancel that reached the agent is shared; a failed write is retried next call. - active.cancelPromise = undefined - active.cancelling = false - throw error - }) - await Promise.race([ - active.response.catch((error) => { - if (connection.closed()) { - throw error - } - }), - unconfirmed - ]) - } finally { - clearTimeout(timer) - } -} - -/** A steer's cancel: asks the agent once to end Orca's running prompt so the steer can follow it, - * and resolves when the notification is written. Never bounded and never closes: the prompt's own - * reply ends it, however long that takes, and a later Stop still bounds. A prompt that fails - * instead leaves a session the steer must not be sent into until the caller rebuilds it. With no - * prompt of Orca's there is nothing to end; a Stop already cancelling it owns that cancel. */ -export function requestAcpSteerCancel( - active: ActivePrompt | undefined, - channel: Pick -): Promise { - if (!active) { - return Promise.resolve() - } - if (active.steerCancel || active.cancelling) { - return active.steerCancel ?? Promise.resolve() - } - active.cancelling = true - channel.cancelIncomingRequests() - active.steerCancel = channel.send().catch((error: unknown) => { - // Only a cancel that reached the agent counts; a failed write may be asked again. - active.cancelling = false - active.steerCancel = undefined - throw error - }) - return active.steerCancel -} diff --git a/src/main/acp/acp-session-lifecycle.test.ts b/src/main/acp/acp-session-lifecycle.test.ts index 7230870a8c02..a3020620c1e0 100644 --- a/src/main/acp/acp-session-lifecycle.test.ts +++ b/src/main/acp/acp-session-lifecycle.test.ts @@ -3,7 +3,7 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import { AcpJsonRpcPeer } from './acp-json-rpc-peer' import { AcpSessionRuntime, type AcpSessionRuntimeOptions } from './acp-session-runtime' import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support' -import { AcpConnectionClosedError, AcpRequestTimeoutError } from './acp-errors' +import { AcpConnectionClosedError, AcpRpcError } from './acp-errors' import type { RequestPermissionResponse } from './generated/acp-protocol.generated' const opened: { close: () => void }[] = [] @@ -219,21 +219,6 @@ describe('ACP caller-owned waits', () => { await rejected }) - it('clamps cancellation confirmation waits to the safe timer range', async () => { - vi.useFakeTimers() - const { agent, runtime } = fixture({ cancelTimeoutMs: 3_000_000_000 }) - agent.on('session/prompt', () => {}) - await runtime.start(startOptions) - const rejected = expect(runtime.prompt([...prompt])).rejects.toBeInstanceOf( - AcpRequestTimeoutError - ) - const cancelled = expect(runtime.cancel()).rejects.toBeInstanceOf(AcpRequestTimeoutError) - await vi.advanceTimersByTimeAsync(1) - expect(agent.frames.at(-1)?.method).toBe('session/cancel') - await vi.advanceTimersByTimeAsync(2_147_483_646) - await Promise.all([rejected, cancelled]) - }) - it('lets initialize and explicit authentication wait for slow first-run startup and login', async () => { vi.useFakeTimers() const { agent, runtime } = fixture({ peer: { requestTimeoutMs: 20 } }) @@ -248,16 +233,17 @@ describe('ACP caller-owned waits', () => { await expect(authenticated).resolves.toEqual({}) }) - it('lets cancel abort vendor hooks so each sends its own answer, or request-cancelled if it throws', async () => { + it('lets request owners withdraw vendor hooks after cancel', async () => { + const withdrawal = new AbortController() const { agent, runtime } = fixture({ - onRequest: (method, _params, context) => + onRequest: (method) => new Promise((resolve, reject) => { // '_vendor/silent' ignores the abort: the runtime never answers for it. if (method !== '_vendor/silent') { - context.signal.addEventListener('abort', () => + withdrawal.signal.addEventListener('abort', () => method === '_vendor/plan' ? resolve({ outcome: 'abandoned' }) - : reject(new Error('stop')) + : reject(new AcpRpcError(-32800, 'stop')) ) } }) @@ -272,6 +258,7 @@ describe('ACP caller-owned waits', () => { void agent.request('silent', '_vendor/silent', {}) await tick() await runtime.cancel() + withdrawal.abort() expect(await question).toMatchObject({ error: { code: -32800 } }) expect(await plan).toMatchObject({ result: { outcome: 'abandoned' } }) await pending @@ -280,18 +267,19 @@ describe('ACP caller-owned waits', () => { expect(agent.frames.some((frame) => frame.id === 'silent')).toBe(false) }) - it('keeps a handler answer that finishes saving after the cancel', async () => { + it('keeps a handler answer that finishes saving after its owner withdraws requests', async () => { + const withdrawal = new AbortController() // A real I/O hop, the shape of a journal write the handler commits before replying. const save = (): Promise => readFile(import.meta.filename).then(() => undefined) const userAnswer = deferred() const { agent, runtime } = fixture({ - onRequest: (method, _params, context) => + onRequest: (method) => new Promise((resolve) => { if (method === '_vendor/plan') { // The user already approved; the save started before the stop and replies after it. void userAnswer.promise.then(save).then(() => resolve({ outcome: 'approved' })) } else { - context.signal.addEventListener( + withdrawal.signal.addEventListener( 'abort', () => void save().then(() => resolve({ outcome: 'abandoned' })) ) @@ -308,6 +296,7 @@ describe('ACP caller-owned waits', () => { await tick() userAnswer.resolve() await runtime.cancel() + withdrawal.abort() expect(await plan).toMatchObject({ result: { outcome: 'approved' } }) expect(await question).toMatchObject({ result: { outcome: 'abandoned' } }) await pending @@ -316,7 +305,7 @@ describe('ACP caller-owned waits', () => { expect(agent.frames.filter((frame) => frame.id === 'question')).toHaveLength(1) }) - it('settles requests when the process owner closes on exit even if stdout stays open', async () => { + it('settles stream-level requests on close even if stdout stays open', async () => { const { agent, runtime } = fixture() agent.on('session/prompt', () => {}) await runtime.start(startOptions) @@ -326,7 +315,7 @@ describe('ACP caller-owned waits', () => { await rejected expect(agent.stdout.readableEnded).toBe(false) }) - it.each(['answer', 'cancel'] as const)( + it.each(['answer', 'close'] as const)( 'keeps a permission pending after prompt completion until caller %s', async (action) => { const decision = deferred() @@ -354,13 +343,13 @@ describe('ACP caller-owned waits', () => { result: { outcome: { outcome: 'selected', optionId: 'allow' } } }) } else { - await runtime.cancel() - expect(await response).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } }) + runtime.close() + expect(signal.aborted).toBe(true) } } ) - it('confirms cancellation when the agent settles the prompt with an error', async () => { + it('writes cancel independently of the agent settling the prompt with an error', async () => { const { agent, runtime } = fixture() agent.on('session/prompt', (frame) => { agent.on('session/cancel', () => agent.fail(frame, -32800, 'Request cancelled')) diff --git a/src/main/acp/acp-session-runtime.test.ts b/src/main/acp/acp-session-runtime.test.ts index ef5369e8589f..6b1dff7cc1b8 100644 --- a/src/main/acp/acp-session-runtime.test.ts +++ b/src/main/acp/acp-session-runtime.test.ts @@ -1,16 +1,11 @@ import { afterEach, describe, expect, it, vi } from 'vitest' -import { - AcpAuthRequiredError, - AcpConnectionClosedError, - AcpRequestTimeoutError, - AcpRpcError -} from './acp-errors' +import { AcpAuthRequiredError, AcpConnectionClosedError, AcpRpcError } from './acp-errors' import { AcpSessionRuntime, type AcpSessionRuntimeOptions, type AcpSessionEvent } from './acp-session-runtime' -import { AcpScriptedAgent, deferred, tick } from './acp-scripted-agent.test-support' +import { AcpScriptedAgent, deferred } from './acp-scripted-agent.test-support' import type { AgentCapabilities, RequestPermissionResponse @@ -162,10 +157,9 @@ describe('ACP session runtime', () => { }) }) - it('cancels open and late permissions and waits for the agent cancellation response', async () => { + it('leaves open and late permission decisions with their owner after cancel', async () => { const requested = deferred() const decision = deferred() - const responses: unknown[] = [] const { runtime, agent } = fixture( {}, { @@ -175,35 +169,23 @@ describe('ACP session runtime', () => { } } ) - agent.on('session/prompt', (frame) => { - void agent - .request('permission', 'session/request_permission', permission) - .then((response) => responses.push(response.result)) - agent.on('session/cancel', (cancel) => { - expect(cancel.id).toBeUndefined() - void agent - .request('late-permission', 'session/request_permission', permission) - .then((response) => { - responses.push(response.result) - agent.reply(frame, { stopReason: 'cancelled' }) - }) - }) - }) + agent.on('session/prompt', () => {}) await runtime.start(startOptions) const prompt = runtime.prompt([...textPrompt]) + const open = agent.request('open', 'session/request_permission', permission) const signal = await requested.promise - const cancel = runtime.cancel() - expect(runtime.cancel()).toBe(cancel) - expect(await prompt).toEqual({ stopReason: 'cancelled' }) - await cancel - expect(signal.aborted).toBe(true) - expect(responses).toEqual([ - { outcome: { outcome: 'cancelled' } }, - { outcome: { outcome: 'cancelled' } } - ]) - decision.resolve({ outcome: { outcome: 'selected', optionId: 'allow' } }) - await tick() - expect(agent.frames.filter((frame) => frame.id === 'permission')).toHaveLength(1) + await runtime.cancel() + expect(signal.aborted).toBe(false) + const late = agent.request('late', 'session/request_permission', permission) + decision.resolve({ outcome: { outcome: 'cancelled' } }) + expect(await open).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } }) + expect(await late).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } }) + const frame = agent.frames.find((frame) => frame.method === 'session/prompt') + if (!frame) { + throw new Error('missing prompt') + } + agent.reply(frame, { stopReason: 'cancelled' }) + await prompt }) it('surfaces authentication-required errors with code and data', async () => { @@ -409,88 +391,24 @@ describe('ACP session runtime', () => { await expect(runtime.setMode('plan')).rejects.toBeInstanceOf(AcpConnectionClosedError) }) - it('closes an unconfirmed cancellation rather than allowing another prompt', async () => { + it('writes cancel immediately without waiting, coalescing, timing out, or closing', async () => { vi.useFakeTimers() - const { runtime, agent } = fixture({}, { cancelTimeoutMs: 100 }) - agent.on('session/prompt', () => {}) + const { runtime, agent } = fixture() + const frame = deferred[0]>() + agent.on('session/prompt', (prompt) => frame.resolve(prompt)) await runtime.start(startOptions) const prompt = runtime.prompt([...textPrompt]) - const rejectedPrompt = expect(prompt).rejects.toBeInstanceOf(AcpRequestTimeoutError) - const cancelled = expect(runtime.cancel()).rejects.toBeInstanceOf(AcpRequestTimeoutError) - await vi.advanceTimersByTimeAsync(100) - await Promise.all([rejectedPrompt, cancelled]) - await expect(runtime.prompt([...textPrompt])).rejects.toBeInstanceOf(AcpRequestTimeoutError) - }) - - describe("a steer's cancel", () => { - const cancels = (agent: AcpScriptedAgent) => - agent.frames.filter((frame) => frame.method === 'session/cancel') - - it("asks once, never times out or closes, and the prompt's own reply ends it", async () => { - vi.useFakeTimers() - const signals: AbortSignal[] = [] - const { runtime, agent } = fixture( - {}, - { - cancelTimeoutMs: 100, - // Would allow every request it is asked; a steer's cancel keeps late ones from it. - onPermission: (_request, context) => { - signals.push(context.signal) - return new Promise(() => {}) - } - } - ) - const promptFrame = deferred[0]>() - agent.on('session/prompt', (frame) => promptFrame.resolve(frame)) - await runtime.start(startOptions) - const prompt = runtime.prompt([...textPrompt]) - const frame = await promptFrame.promise - const open = agent.request('open', 'session/request_permission', permission) - await vi.waitFor(() => expect(signals).toHaveLength(1)) - - const steer = runtime.requestSteerCancel() - expect(runtime.requestSteerCancel()).toBe(steer) - await steer - await vi.advanceTimersByTimeAsync(1_000) - - expect(cancels(agent)).toHaveLength(1) - expect(signals[0]?.aborted).toBe(true) - expect(await open).toMatchObject({ result: { outcome: { outcome: 'cancelled' } } }) - expect(await agent.request('late', 'session/request_permission', permission)).toMatchObject({ - result: { outcome: { outcome: 'cancelled' } } - }) - expect(signals).toHaveLength(1) - agent.reply(frame, { stopReason: 'cancelled' }) - expect(await prompt).toEqual({ stopReason: 'cancelled' }) - // The connection is still open, so the steer's own prompt follows. - agent.on('session/prompt', (next) => agent.reply(next, { stopReason: 'end_turn' })) - expect(await runtime.prompt([...textPrompt])).toEqual({ stopReason: 'end_turn' }) - }) - - it('leaves a later Stop bounded, closing the connection when the prompt never settles', async () => { - vi.useFakeTimers() - const { runtime, agent } = fixture({}, { cancelTimeoutMs: 100 }) - agent.on('session/prompt', () => {}) - await runtime.start(startOptions) - const prompt = runtime.prompt([...textPrompt]) - const rejectedPrompt = expect(prompt).rejects.toBeInstanceOf(AcpRequestTimeoutError) - - await runtime.requestSteerCancel() - const stopped = expect(runtime.cancel()).rejects.toBeInstanceOf(AcpRequestTimeoutError) - await vi.advanceTimersByTimeAsync(100) - - await Promise.all([rejectedPrompt, stopped]) - expect(cancels(agent)).toHaveLength(2) - await expect(runtime.prompt([...textPrompt])).rejects.toBeInstanceOf(AcpRequestTimeoutError) - }) - - it("writes nothing with no prompt of Orca's running", async () => { - const { runtime, agent } = fixture() - await expect(runtime.requestSteerCancel()).resolves.toBeUndefined() - await runtime.start(startOptions) - await runtime.requestSteerCancel() - await tick() - expect(cancels(agent)).toHaveLength(0) - }) + const sent = await frame.promise + await runtime.cancel() + await runtime.cancel() + expect(agent.frames.filter((frame) => frame.method === 'session/cancel')).toHaveLength(2) + expect(vi.getTimerCount()).toBe(0) + await vi.advanceTimersByTimeAsync(120_000) + expect(runtime.closed).toBe(false) + await expect(runtime.prompt([...textPrompt])).rejects.toThrow('already in progress') + agent.reply(sent, { stopReason: 'cancelled' }) + expect(await prompt).toEqual({ stopReason: 'cancelled' }) + agent.on('session/prompt', (next) => agent.reply(next, { stopReason: 'end_turn' })) + expect(await runtime.prompt([...textPrompt])).toEqual({ stopReason: 'end_turn' }) }) }) diff --git a/src/main/acp/acp-session-runtime.ts b/src/main/acp/acp-session-runtime.ts index eb6ceb821947..e205c1f67c46 100644 --- a/src/main/acp/acp-session-runtime.ts +++ b/src/main/acp/acp-session-runtime.ts @@ -1,6 +1,5 @@ import type { Readable, Writable } from 'node:stream' import type { z } from 'zod' -import { MAX_TIMER_DELAY_MS } from '../../shared/timer-delay' import { AcpAgentError, AcpAuthRequiredError, @@ -14,12 +13,6 @@ import { type AcpPermissionHandler } from './acp-permission-requests' import { readAcpSessionEvent, type AcpSessionEvent } from './acp-session-events' -import { - cancelAcpPromptForStop, - requestAcpSteerCancel, - type AcpCancelChannel, - type ActivePrompt -} from './acp-prompt-cancel' import { setupAcpSession, type AcpSessionStarted, @@ -55,7 +48,6 @@ const withMeta = (meta: Meta): { _meta?: Meta } => (meta ? { _meta: meta } : {}) export type AcpSessionRuntimeOptions = { clientInfo?: InitializeRequest['clientInfo'] peer?: AcpPeerOptions - cancelTimeoutMs?: number onPermission?: AcpPermissionHandler /** Agent requests other than permissions. The handler owns its request: once `context.signal` * aborts, send the agent's own cancelled reply, finish an answer already in progress, or throw @@ -68,13 +60,14 @@ export type AcpSessionRuntimeOptions = { onClose?: (error: Error) => void } +/** Stream-level protocol seam; production agents use createAcpAgentConnection. */ export class AcpSessionRuntime { private readonly peer: AcpJsonRpcPeer private readonly listeners = new Set<(event: AcpSessionEvent) => void>() private initialized?: Promise private starting?: Promise private started?: AcpSessionStarted - private activePrompt?: ActivePrompt + private activePrompt?: Promise private reportedUpdateAnomaly = false constructor( @@ -82,10 +75,6 @@ export class AcpSessionRuntime { output: Writable, private readonly options: AcpSessionRuntimeOptions = {} ) { - const timeout = options.cancelTimeoutMs - if (timeout !== undefined && (!Number.isSafeInteger(timeout) || timeout <= 0)) { - throw new Error('ACP timeouts must be positive finite timer durations') - } this.peer = new AcpJsonRpcPeer( input, output, @@ -99,6 +88,10 @@ export class AcpSessionRuntime { ) } + get closed(): boolean { + return this.peer.closed + } + subscribe(listener: (event: AcpSessionEvent) => void): () => void { this.listeners.add(listener) return () => { @@ -165,32 +158,27 @@ export class AcpSessionRuntime { throw new Error('ACP prompt already in progress') } const params: PromptRequest = { sessionId: this.sessionId(), prompt, ...withMeta(meta) } - const active: ActivePrompt = { - cancelling: false, - response: this.call('session/prompt', params, PromptResponseSchema) - } - this.activePrompt = active - active.response = active.response.finally(() => { - if (this.activePrompt === active) { + const response = this.call('session/prompt', params, PromptResponseSchema) + this.activePrompt = response + try { + return await response + } finally { + if (this.activePrompt === response) { this.activePrompt = undefined } - }) - return active.response + } } - /** A Stop's cancel (`cancelAcpPromptForStop`): bounded, and past the bound it closes. A prompt - * that settles in time leaves the agent running; the Stop's owner ends its process. */ + /** Stop and steering share this notification; the adapter owns their completion policy. */ cancel(options: { meta?: CancelNotification['_meta'] } = {}): Promise { - const channel = this.cancelChannel(options.meta) - return channel - ? cancelAcpPromptForStop(this.activePrompt, channel) - : Promise.reject(new Error('ACP session has not started')) - } - - /** A steer's cancel (`requestAcpSteerCancel`): never bounded and never closes. */ - requestSteerCancel(options: { meta?: CancelNotification['_meta'] } = {}): Promise { - const channel = this.cancelChannel(options.meta) - return channel ? requestAcpSteerCancel(this.activePrompt, channel) : Promise.resolve() + if (!this.started) { + return Promise.reject(new Error('ACP session has not started')) + } + const params: CancelNotification = { + sessionId: this.started.sessionId, + ...withMeta(options.meta) + } + return this.peer.notify('session/cancel', params) } async setMode(modeId: string, meta?: Meta): Promise { @@ -229,26 +217,11 @@ export class AcpSessionRuntime { return this.call('session/set_config_option', request, SetSessionConfigOptionResponseSchema) } - // The process owner must call close on child exit, even if descendants keep stdio open. close(error?: Error): void { this.peer.close(error) this.listeners.clear() } - private cancelChannel(meta: CancelNotification['_meta']): AcpCancelChannel | null { - if (!this.started) { - return null - } - const params: CancelNotification = { sessionId: this.started.sessionId, ...withMeta(meta) } - return { - send: () => this.peer.notify('session/cancel', params), - cancelIncomingRequests: () => this.peer.cancelIncomingRequests(), - close: (error) => this.peer.close(error), - closed: () => this.peer.closed, - timeoutMs: Math.min(this.options.cancelTimeoutMs ?? 10_000, MAX_TIMER_DELAY_MS) - } - } - private sessionId(): string { if (!this.started) { throw new Error('ACP session has not started') @@ -287,7 +260,7 @@ export class AcpSessionRuntime { throw new AcpRpcError(-32602, 'Invalid ACP permission request') } // Whether a turn the agent began itself may ask is the caller's call; the runtime cannot see it. - if (this.activePrompt?.cancelling || request.sessionId !== this.started?.sessionId) { + if (request.sessionId !== this.started?.sessionId) { return { outcome: { outcome: 'cancelled' } } } return answerAcpPermission(request, context, this.options.onPermission, diagnose)