diff --git a/packages/support/acp-snapshot/src/launcher.ts b/packages/support/acp-snapshot/src/launcher.ts index 7f2ec9b0a7..253e67fd5a 100644 --- a/packages/support/acp-snapshot/src/launcher.ts +++ b/packages/support/acp-snapshot/src/launcher.ts @@ -114,18 +114,26 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe const rawBuffers: Buffer[] = [] const passthrough = new Readable({ read() {} }) - child.stdout.on('data', (buffer: Buffer) => { - rawBuffers.push(buffer) - passthrough.push(buffer) - }) - child.stdout.on('end', () => passthrough.push(null)) - const updates: SessionNotification['update'][] = [] const updateWaiters: { match: (update: SessionNotification['update']) => boolean resolve: (update: SessionNotification['update']) => void reject: (reason: unknown) => void }[] = [] + let updateStreamFailure: Error | undefined + const closeUpdateStream = (): void => { + if (updateStreamFailure !== undefined) return + updateStreamFailure = new Error('ACP test agent update stream closed before a matching session update arrived') + for (const waiter of updateWaiters.splice(0)) waiter.reject(updateStreamFailure) + } + child.stdout.on('data', (buffer: Buffer) => { + rawBuffers.push(buffer) + passthrough.push(buffer) + }) + child.stdout.on('end', () => { + passthrough.push(null) + closeUpdateStream() + }) const stream = ndJsonStream( Writable.toWeb(child.stdin) as WritableStream, Readable.toWeb(passthrough) as ReadableStream, @@ -163,10 +171,21 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe updates, rawStdout: () => Buffer.concat(rawBuffers).toString('utf8'), stderr: () => stderrChunks.join(''), - waitForUpdate: match => new Promise((resolve, reject) => updateWaiters.push({ match, resolve, reject })), + waitForUpdate(match): Promise { + if (updateStreamFailure !== undefined) return Promise.reject(updateStreamFailure) + return new Promise((resolve, reject) => updateWaiters.push({ match, resolve, reject })) + }, async close(signal?: NodeJS.Signals): Promise { - await spawned - if (!isRunning(child)) return + try { + await spawned + } catch (error: unknown) { + closeUpdateStream() + throw error + } + if (!isRunning(child)) { + closeUpdateStream() + return + } const exited = waitForExit(child) if (signal === undefined) child.stdin.end() else child.kill(signal) @@ -174,7 +193,10 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe exited.then((): undefined => undefined), childFailure, ]) - if (failure === undefined) return + if (failure === undefined) { + closeUpdateStream() + return + } // An `error` after spawn is not an exit edge: in particular, a failed // signal can leave the subprocess live. Force termination, await the @@ -182,6 +204,7 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe // callers may safely remove cwd/session resources after close rejects. child.kill('SIGKILL') await exited + closeUpdateStream() throw failure }, } diff --git a/packages/support/acp-snapshot/tests/harness.spec.ts b/packages/support/acp-snapshot/tests/harness.spec.ts index 3c1bb0d44a..192d36dc6c 100644 --- a/packages/support/acp-snapshot/tests/harness.spec.ts +++ b/packages/support/acp-snapshot/tests/harness.spec.ts @@ -73,7 +73,10 @@ describe('runScenario', () => { expect(launched.updates.some(update => update.sessionUpdate === 'agent_message_chunk')).toBe(true) expect(launched.rawStdout()).toContain('permission:{\\"outcome\\":\\"cancelled\\"}') expect(launched.stderr()).toContain('launcher stderr') + const unmatched = expect(launched.waitForUpdate(() => false)).rejects.toThrow(/update stream closed/) await launched.close() + await unmatched + await expect(launched.waitForUpdate(() => true)).rejects.toThrow(/update stream closed/) await launched.close('SIGKILL') // The minimal shape needs no environment or config override.