fix: settle ACP update waiters on shutdown
This commit is contained in:
@@ -114,18 +114,26 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe
|
|||||||
|
|
||||||
const rawBuffers: Buffer[] = []
|
const rawBuffers: Buffer[] = []
|
||||||
const passthrough = new Readable({ read() {} })
|
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 updates: SessionNotification['update'][] = []
|
||||||
const updateWaiters: {
|
const updateWaiters: {
|
||||||
match: (update: SessionNotification['update']) => boolean
|
match: (update: SessionNotification['update']) => boolean
|
||||||
resolve: (update: SessionNotification['update']) => void
|
resolve: (update: SessionNotification['update']) => void
|
||||||
reject: (reason: unknown) => 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(
|
const stream = ndJsonStream(
|
||||||
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
|
Writable.toWeb(child.stdin) as WritableStream<Uint8Array>,
|
||||||
Readable.toWeb(passthrough) as ReadableStream<Uint8Array>,
|
Readable.toWeb(passthrough) as ReadableStream<Uint8Array>,
|
||||||
@@ -163,10 +171,21 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe
|
|||||||
updates,
|
updates,
|
||||||
rawStdout: () => Buffer.concat(rawBuffers).toString('utf8'),
|
rawStdout: () => Buffer.concat(rawBuffers).toString('utf8'),
|
||||||
stderr: () => stderrChunks.join(''),
|
stderr: () => stderrChunks.join(''),
|
||||||
waitForUpdate: match => new Promise((resolve, reject) => updateWaiters.push({ match, resolve, reject })),
|
waitForUpdate(match): Promise<SessionNotification['update']> {
|
||||||
|
if (updateStreamFailure !== undefined) return Promise.reject(updateStreamFailure)
|
||||||
|
return new Promise((resolve, reject) => updateWaiters.push({ match, resolve, reject }))
|
||||||
|
},
|
||||||
async close(signal?: NodeJS.Signals): Promise<void> {
|
async close(signal?: NodeJS.Signals): Promise<void> {
|
||||||
await spawned
|
try {
|
||||||
if (!isRunning(child)) return
|
await spawned
|
||||||
|
} catch (error: unknown) {
|
||||||
|
closeUpdateStream()
|
||||||
|
throw error
|
||||||
|
}
|
||||||
|
if (!isRunning(child)) {
|
||||||
|
closeUpdateStream()
|
||||||
|
return
|
||||||
|
}
|
||||||
const exited = waitForExit(child)
|
const exited = waitForExit(child)
|
||||||
if (signal === undefined) child.stdin.end()
|
if (signal === undefined) child.stdin.end()
|
||||||
else child.kill(signal)
|
else child.kill(signal)
|
||||||
@@ -174,7 +193,10 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe
|
|||||||
exited.then((): undefined => undefined),
|
exited.then((): undefined => undefined),
|
||||||
childFailure,
|
childFailure,
|
||||||
])
|
])
|
||||||
if (failure === undefined) return
|
if (failure === undefined) {
|
||||||
|
closeUpdateStream()
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
// An `error` after spawn is not an exit edge: in particular, a failed
|
// An `error` after spawn is not an exit edge: in particular, a failed
|
||||||
// signal can leave the subprocess live. Force termination, await the
|
// 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.
|
// callers may safely remove cwd/session resources after close rejects.
|
||||||
child.kill('SIGKILL')
|
child.kill('SIGKILL')
|
||||||
await exited
|
await exited
|
||||||
|
closeUpdateStream()
|
||||||
throw failure
|
throw failure
|
||||||
},
|
},
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -73,7 +73,10 @@ describe('runScenario', () => {
|
|||||||
expect(launched.updates.some(update => update.sessionUpdate === 'agent_message_chunk')).toBe(true)
|
expect(launched.updates.some(update => update.sessionUpdate === 'agent_message_chunk')).toBe(true)
|
||||||
expect(launched.rawStdout()).toContain('permission:{\\"outcome\\":\\"cancelled\\"}')
|
expect(launched.rawStdout()).toContain('permission:{\\"outcome\\":\\"cancelled\\"}')
|
||||||
expect(launched.stderr()).toContain('launcher stderr')
|
expect(launched.stderr()).toContain('launcher stderr')
|
||||||
|
const unmatched = expect(launched.waitForUpdate(() => false)).rejects.toThrow(/update stream closed/)
|
||||||
await launched.close()
|
await launched.close()
|
||||||
|
await unmatched
|
||||||
|
await expect(launched.waitForUpdate(() => true)).rejects.toThrow(/update stream closed/)
|
||||||
await launched.close('SIGKILL')
|
await launched.close('SIGKILL')
|
||||||
|
|
||||||
// The minimal shape needs no environment or config override.
|
// The minimal shape needs no environment or config override.
|
||||||
|
|||||||
Reference in New Issue
Block a user