From e784e4dce5f5178b5e22a3d3376599144d8bea1b Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Tue, 14 Jul 2026 09:12:32 +0800 Subject: [PATCH] fix: await ACP client callbacks during shutdown --- packages/support/acp-snapshot/src/launcher.ts | 59 ++++++++++++------- .../acp-snapshot/tests/harness.spec.ts | 36 +++++++++++ 2 files changed, 75 insertions(+), 20 deletions(-) diff --git a/packages/support/acp-snapshot/src/launcher.ts b/packages/support/acp-snapshot/src/launcher.ts index 815753ae5f..30702a5888 100644 --- a/packages/support/acp-snapshot/src/launcher.ts +++ b/packages/support/acp-snapshot/src/launcher.ts @@ -65,7 +65,7 @@ export interface LaunchedAcpTestAgent { stderr(): string /** Resolve when a future session update matches the predicate. */ waitForUpdate(match: (update: SessionNotification['update']) => boolean): Promise - /** Gracefully close stdin, or send a signal, then wait for process exit, inherited stdio closure, and ACP parser drain. */ + /** Gracefully close stdin, or send a signal, then wait for process exit, inherited stdio closure, ACP parsing, and client callbacks. */ close(signal?: NodeJS.Signals): Promise } @@ -137,29 +137,41 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe Writable.toWeb(child.stdin) as WritableStream, Readable.toWeb(passthrough) as ReadableStream, ) + const inFlightClientCallbacks = new Set>() + const trackClientCallback = (callback: () => T | PromiseLike): Promise => { + const pending = Promise.resolve().then(callback) + inFlightClientCallbacks.add(pending) + void pending.then( + () => { inFlightClientCallbacks.delete(pending) }, + () => { inFlightClientCallbacks.delete(pending) }, + ) + return pending + } + const requestPermission = options.requestPermission + ?? (() => Promise.resolve({ outcome: { outcome: 'cancelled' as const } })) const makeClient = (_agent: AcpAgent): Client => ({ sessionUpdate(params: SessionNotification): Promise { - updates.push(params.update) - for (let index = updateWaiters.length - 1; index >= 0; index--) { - const waiter = updateWaiters[index] - /* v8 ignore next 1 -- index is bounded by the array length */ - if (waiter === undefined) continue - let matches: boolean - try { - matches = waiter.match(params.update) - } catch (error: unknown) { + return trackClientCallback(() => { + updates.push(params.update) + for (let index = updateWaiters.length - 1; index >= 0; index--) { + const waiter = updateWaiters[index] + /* v8 ignore next 1 -- index is bounded by the array length */ + if (waiter === undefined) continue + let matches: boolean + try { + matches = waiter.match(params.update) + } catch (error: unknown) { + updateWaiters.splice(index, 1) + waiter.reject(error) + continue + } + if (!matches) continue updateWaiters.splice(index, 1) - waiter.reject(error) - continue + waiter.resolve(params.update) } - if (!matches) continue - updateWaiters.splice(index, 1) - waiter.resolve(params.update) - } - return Promise.resolve() + }) }, - requestPermission: options.requestPermission - ?? (() => Promise.resolve({ outcome: { outcome: 'cancelled' } })), + requestPermission: params => trackClientCallback(() => requestPermission(params)), }) const client = new ClientSideConnection(makeClient, stream) // `exit` only reports the parent process's status. Descendants may retain @@ -168,7 +180,14 @@ export function launchAcpTestAgent(options: AcpTestLaunchOptions): LaunchedAcpTe // `closed` follows parser exhaustion. Capture both eagerly so a caller that // invokes close after process exit still joins the complete drain boundary. const stdioClosed = new Promise(resolve => child.once('close', () => { resolve() })) - const drained = Promise.all([stdioClosed, client.closed]).then(() => undefined) + const drained = Promise.all([stdioClosed, client.closed]).then(async () => { + // The ACP SDK's readable loop dispatches client callbacks without awaiting + // them. Once `closed` settles no new callbacks can start, but callbacks + // already in flight still belong to this launch's teardown boundary. + while (inFlightClientCallbacks.size > 0) { + await Promise.allSettled([...inFlightClientCallbacks]) + } + }) // A caller may await a pending update without calling close(). Make natural // stream exhaustion terminal for those waiters too, but only after the // parser has dispatched every buffered frame. diff --git a/packages/support/acp-snapshot/tests/harness.spec.ts b/packages/support/acp-snapshot/tests/harness.spec.ts index 051db6a0ba..902f48d2cf 100644 --- a/packages/support/acp-snapshot/tests/harness.spec.ts +++ b/packages/support/acp-snapshot/tests/harness.spec.ts @@ -1,4 +1,5 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { once } from 'node:events' import { tmpdir } from 'node:os' import { delimiter, join } from 'node:path' import { fileURLToPath } from 'node:url' @@ -112,6 +113,41 @@ describe('runScenario', () => { expect(launched.stderr()).toContain('late inherited stderr') }) + it('waits for in-flight client callbacks after the ACP stream closes', { timeout: 20_000 }, async () => { + const { dir, fixtureFile } = await scenario({ permissionProbe: true }) + let releasePermission: (() => void) | undefined + const permissionReleased = new Promise((resolve) => { releasePermission = resolve }) + let markPermissionStarted: (() => void) | undefined + const permissionStarted = new Promise((resolve) => { markPermissionStarted = resolve }) + let permissionFinished = false + const launched = launchAcpTestAgent({ + agent: AGENT, + cwd: dir, + env: { DSH_SNAPSHOT_FILE: fixtureFile }, + async requestPermission() { + markPermissionStarted?.() + await permissionReleased + permissionFinished = true + return { outcome: { outcome: 'cancelled' } } + }, + }) + await launched.client.initialize({ protocolVersion: PROTOCOL_VERSION, clientCapabilities: {} }) + const { sessionId } = await launched.client.newSession({ cwd: dir, mcpServers: [] }) + void launched.client.prompt({ sessionId, prompt: [{ type: 'text', text: 'go' }] }).catch(() => undefined) + await permissionStarted + + const childClosed = once(launched.child, 'close') + let closeSettled = false + const closing = launched.close('SIGKILL').then(() => { closeSettled = true }) + await childClosed + await launched.client.closed + expect(closeSettled).toBe(false) + + releasePermission?.() + await closing + expect(permissionFinished).toBe(true) + }) + it('drives a full turn: initialize (terminal caps), session, prompt, permission stub, harvest', { timeout: 20_000 }, async () => { const { fixtureFile } = await scenario({ permissionProbe: true,