From bd846d435b267c61c27227bb5fa80e18524a2aac Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 22 Jul 2026 00:22:42 +0800 Subject: [PATCH 1/2] fix(acp-bridge): close prompt-terminal follow-ups from PR #7400 self-review Keep a removed RUNNING prompt visible to the teardown flush via a removed flag so its terminal still publishes when the session closes before the agent cooperates; gate broadcastTurnError's session turn-state mutation to running prompts; propagate the typed PromptDeadlineExceededError from the pre-dispatch abort check; document the deadline FIFO-release overlap trade-off, the trailing prompt_cancelled after flush, and the result.then/finally ordering invariant; route the dedup log to the debug channel; drop the prompt-deadline re-export that pulled the bridge into a leaf module. Fixes #7451 --- packages/acp-bridge/src/bridge.test.ts | 59 ++++++++++ packages/acp-bridge/src/bridge.ts | 107 ++++++++++++++---- packages/acp-bridge/src/bridgeTypes.ts | 7 ++ packages/cli/src/serve/server.ts | 6 +- .../cli/src/serve/server/prompt-deadline.ts | 8 -- 5 files changed, 152 insertions(+), 35 deletions(-) diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 5ba6187e8dc..a8f48309a60 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -6821,6 +6821,13 @@ describe('createAcpSessionBridge', () => { await new Promise((r) => setTimeout(r, 60)); expect(terminalsFor(events, 'prompt-queued-deadline')).toHaveLength(1); + // A queued prompt never ran, so its deadline terminal must not + // advertise a session-level turnError nor arm the retry path — those + // belong to the ACTIVE turn only. + expect( + bridge.getSessionSummary(session.sessionId).turnError, + ).toBeUndefined(); + await bridge.shutdown(); // Shutdown flushed the still-wedged head prompt exactly once, and the // queued prompt's residual FIFO abort stayed latched. @@ -6831,6 +6838,11 @@ describe('createAcpSessionBridge', () => { expect((headTerms[0]?.data as { code?: string }).code).toBe( 'daemon_shutdown', ); + // The caller sees the same typed rejection as a running-prompt expiry + // — the pre-dispatch abort check (reached once shutdown released the + // wedged head) propagates the deadline reason instead of a generic + // AbortError. + await expect(p2).rejects.toBeInstanceOf(PromptDeadlineExceededError); }); it('keeps a detached session draining until the last pending prompt settles (DAEMON-005)', async () => { @@ -6934,6 +6946,53 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('still publishes a terminal for a removed RUNNING prompt when the session closes before the agent cooperates', async () => { + const handle = wedgeChannel(); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const events: BridgeEvent[] = []; + subscribe(bridge, session.sessionId, events); + + const p1 = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'wedge' }], + }, + undefined, + { promptId: 'prompt-removed-running' }, + ); + p1.catch(() => {}); + await new Promise((r) => setTimeout(r, 20)); + + // Remove the RUNNING prompt — the wedged agent ignores the cancel, + // so no terminal has been published yet. + expect( + bridge.removePendingPrompt(session.sessionId, 'prompt-removed-running'), + ).toEqual({ removed: true }); + // The API no longer shows the prompt… + expect(bridge.getPendingPrompts(session.sessionId)).toHaveLength(0); + // …and a repeat removal is a no-op. + expect( + bridge.removePendingPrompt(session.sessionId, 'prompt-removed-running'), + ).toEqual({ removed: false }); + expect(terminalsFor(events, 'prompt-removed-running')).toHaveLength(0); + + // Session closes before the agent ever settles: the teardown flush + // must still see the removed-but-unsettled prompt and publish its + // terminal before the bus closes. + await bridge.closeSession(session.sessionId); + + const closedIdx = events.findIndex((e) => e.type === 'session_closed'); + expect(closedIdx).toBeGreaterThan(-1); + const terms = terminalsFor(events, 'prompt-removed-running'); + expect(terms).toHaveLength(1); + expect(terms[0]?.type).toBe('turn_error'); + expect((terms[0]?.data as { code?: string }).code).toBe('session_closed'); + expect(events.indexOf(terms[0]!)).toBeLessThan(closedIdx); + await bridge.shutdown(); + }); + it('flushes error terminals for active and queued prompts before session_died on killSession (DAEMON-005)', async () => { const handle = wedgeChannel(); const bridge = makeBridge({ channelFactory: async () => handle.channel }); diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 32f5564c741..73a72454484 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -1086,6 +1086,7 @@ function broadcastTurnError( err: unknown, promptId: string | undefined, originatorClientId: string | undefined, + mutateTurnState: boolean, ): void { const message = extractErrorMessage(err); const code = extractErrorCode(err); @@ -1099,12 +1100,20 @@ function broadcastTurnError( (promptId ? ` promptId=${JSON.stringify(promptId)}` : ''), ); } - entry.retryAllowed = true; - entry.turnError = { - message, - ...(code ? { code } : {}), - ...(errorKind ? { errorKind } : {}), - }; + // Session-scoped turn state (`turnError` is surfaced by the summary, + // `retryAllowed` is consumed by the retry-admission check) must only + // reflect the ACTIVE turn's failure. A queued prompt's terminal (deadline + // expiry, teardown flush) publishes the event alone — otherwise a queued + // failure would advertise a `turnError` for a turn that never ran and + // arm a retry the active prompt didn't earn. + if (mutateTurnState) { + entry.retryAllowed = true; + entry.turnError = { + message, + ...(code ? { code } : {}), + ...(errorKind ? { errorKind } : {}), + }; + } try { entry.events.publish({ type: 'turn_error', @@ -1149,7 +1158,10 @@ function publishPromptTerminal( terminal: PromptTerminal, ): void { if (pendingEntry.terminalPublished) { - writeStderrLine( + // Dedup here is the designed steady state, not an anomaly: deadline + // expiry, queued removal, and teardown flush each race the prompt's + // natural settle, so the loser lands here on every such turn. + writeServeDebugLine( `publishPromptTerminal: suppressed duplicate ${terminal.kind} terminal ` + `for prompt ${pendingEntry.promptId} (session ${entry.sessionId})`, ); @@ -1180,6 +1192,13 @@ function publishPromptTerminal( terminal.err, pendingEntry.promptId, originatorClientId, + // Only a running prompt's failure is the active turn's failure. The + // `state === 'running'` gate (not `activePromptId`) is deliberate: + // on the normal settle path `settleActivePromptState` runs in + // `promptPromise.finally` BEFORE the terminal is published, so + // `activePromptId` is already cleared when a genuine active failure + // lands here. + pendingEntry.state === 'running', ); } } @@ -1191,7 +1210,10 @@ function publishPromptTerminal( * check instead of being promoted to running. Must run before * `entry.events.close()` — the bus swallows publishes afterwards. Any * later settle of the same prompts re-enters `publishPromptTerminal` and - * is deduped by the latch. + * is deduped by the latch. For a running prompt the abort fires the + * existing `onAbort` listener while the bus is still open, so a trailing + * `prompt_cancelled` after the terminal frame is expected — consumers + * settling on the terminal by `promptId` are unaffected. */ function flushPromptTerminals( entry: SessionEntry, @@ -5020,7 +5042,14 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // racing the (possibly wedged) `promptPromise`, and the agent is // best-effort cancelled through the existing abort path. The channel // is NOT killed — it may be shared by other sessions; reclaiming a - // wedged agent's channel is a tracked follow-up. + // wedged agent's channel is a tracked follow-up. Releasing the FIFO + // while the wedged call is still outstanding also means the next + // prompt overlaps it on the same ACP session: an agent that ignored + // `cancel()` but keeps streaming will interleave its stale + // `session/update`s with the new turn's output. Accepted trade-off — + // the alternative (poisoning the session until the old call settles) + // would give up the "follow-up prompt dispatches normally" recovery + // property the deadline exists to provide. const deadlineMs = context?.deadlineMs; const hasDeadline = typeof deadlineMs === 'number' && @@ -5108,6 +5137,14 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // already aborted this entry, skip the running transition and // the `pending_prompt_started` event entirely. if (pendingAbort.signal.aborted) { + // A deadline that expired while this prompt was still queued + // aborted with the typed error; surface it to the caller so + // queued and running expiry reject identically. + if ( + pendingAbort.signal.reason instanceof PromptDeadlineExceededError + ) { + throw pendingAbort.signal.reason; + } throw new DOMException('Prompt aborted', 'AbortError'); } // If this prompt was queued behind another, promote it to @@ -5357,6 +5394,11 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { } }), ); + // Do not reorder — this `result.then` must stay registered before the + // `result.finally` below: handlers on the same promise run in + // registration order and the broadcasts are synchronous, which is what + // guarantees the terminal frame precedes the deferred + // close-on-prompt-complete in `result.finally`. result.then( (promptResult) => { publishPromptTerminal(entry, pendingEntry, { @@ -5388,8 +5430,10 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); // Remove this prompt from the pending list and publish a // completed event so SSE subscribers can update their queue view. - // If `removePendingPrompt` already spliced this entry and - // published its own terminal event, skip to avoid a duplicate. + // A removed RUNNING prompt is still on the list (see + // `removePendingPrompt`) — splice it now, but skip the `completed` + // event: its `pending_prompt_completed{state:'removed'}` already + // announced the queue-view change. const listIdx = entry.pendingPromptList.indexOf(pendingEntry); if (listIdx !== -1) { entry.pendingPromptList.splice(listIdx, 1); @@ -5397,7 +5441,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { // (and thus had an `added` event). The first prompt on an idle // session starts immediately without `added`, so publishing // `completed` would produce an unpaired event. - if (isQueued) { + if (isQueued && !pendingEntry.removed) { try { entry.events.publish({ type: 'pending_prompt_completed', @@ -7037,15 +7081,17 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { if (!entry) throw new SessionNotFoundError(sessionId); // Authorize the caller against this session — mirrors /prompt. resolveTrustedClientId(entry, context?.clientId); - return entry.pendingPromptList.map((p) => ({ - promptId: p.promptId, - text: p.text, - queuedAt: p.queuedAt, - state: p.state, - ...(p.originatorClientId !== undefined - ? { originatorClientId: p.originatorClientId } - : {}), - })); + return entry.pendingPromptList + .filter((p) => !p.removed) + .map((p) => ({ + promptId: p.promptId, + text: p.text, + queuedAt: p.queuedAt, + state: p.state, + ...(p.originatorClientId !== undefined + ? { originatorClientId: p.originatorClientId } + : {}), + })); }, removePendingPrompt(sessionId, promptId, context) { @@ -7058,6 +7104,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ); if (idx === -1) return { removed: false }; const target = entry.pendingPromptList[idx]; + // A running prompt already removed once is invisible to the API — + // repeat removals are no-ops. + if (target.removed) return { removed: false }; writeStderrLine( `[pending-prompt] session=${sessionId} removing promptId=${promptId} state=${target.state}`, ); @@ -7067,8 +7116,20 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { target.abortController.abort( new DOMException('Prompt removed by user', 'AbortError'), ); - // Remove from the list immediately so the API reflects the change. - entry.pendingPromptList.splice(idx, 1); + if (target.state === 'queued') { + // A queued prompt never dispatches once aborted — safe to drop + // from the list immediately. + entry.pendingPromptList.splice(idx, 1); + } else { + // A RUNNING prompt must stay on the list (hidden from + // `getPendingPrompts` via the `removed` flag) until it settles + // through `result.finally`. Splicing it here would make it + // invisible to `flushPromptTerminals`: if the session then closes + // before the agent cooperates with the cancel, the prompt's + // terminal would be published into an already-closed bus and + // silently dropped. + target.removed = true; + } // Keep the admission slot until this prompt's FIFO node reaches the head // and settles through the original result.finally() path. Otherwise a // client could enqueue/delete queued prompts repeatedly while one turn is diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 81c3930426c..a5c1fa7867b 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -599,6 +599,13 @@ export interface PendingPromptEntry { * later publish attempts for the same prompt are suppressed. */ terminalPublished?: boolean; + /** + * Set when `removePendingPrompt` cancels a RUNNING prompt. The entry + * stays on `pendingPromptList` (hidden from `getPendingPrompts`) until + * the prompt settles, so the teardown flush can still publish its + * terminal if the session closes before the agent cooperates. + */ + removed?: boolean; } /** diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 56175ecf1dd..0fb295b6499 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -227,10 +227,8 @@ export { resolveBoundWorkspacesFromIdeEnv, resolveBridgeFsFactory, } from './server/fs-factory.js'; -export { - PromptDeadlineExceededError, - resolvePromptDeadlineMs, -} from './server/prompt-deadline.js'; +export { PromptDeadlineExceededError } from './acp-session-bridge.js'; +export { resolvePromptDeadlineMs } from './server/prompt-deadline.js'; export { detectFromLoopback } from './server/request-helpers.js'; export { InvalidCursorError, diff --git a/packages/cli/src/serve/server/prompt-deadline.ts b/packages/cli/src/serve/server/prompt-deadline.ts index 3242f80ec61..82d8ace4064 100644 --- a/packages/cli/src/serve/server/prompt-deadline.ts +++ b/packages/cli/src/serve/server/prompt-deadline.ts @@ -4,14 +4,6 @@ * SPDX-License-Identifier: Apache-2.0 */ -/** - * Rejected by the bridge's `sendPrompt` when a prompt exceeds its - * wallclock deadline. The class itself lives in the acp-bridge package - * (the bridge owns the deadline race since DAEMON-003); re-exported here - * so existing `server.ts` / test imports keep working. - */ -export { PromptDeadlineExceededError } from '../acp-session-bridge.js'; - /** * Resolve the effective per-prompt wallclock from the server flag + * an optional request body override. Returns `undefined` when no From 666f6cd95ded118cc443bc8f47f457ceabed3d3e Mon Sep 17 00:00:00 2001 From: Qwen Code Bot Date: Wed, 22 Jul 2026 02:56:41 +0000 Subject: [PATCH 2/2] test(acp-bridge): cover promote-then-remove-then-settle duplicate completed guard (#7453) --- packages/acp-bridge/src/bridge.test.ts | 92 ++++++++++++++++++++++++++ 1 file changed, 92 insertions(+) diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index a8f48309a60..2240c1317d4 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -6993,6 +6993,98 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('does not publish a duplicate completed event when a promoted-then-removed running prompt settles', async () => { + let releaseFirst: (() => void) | undefined; + const firstDone = new Promise((r) => { + releaseFirst = r; + }); + let releaseSecond: (() => void) | undefined; + const secondDone = new Promise((r) => { + releaseSecond = r; + }); + const handle = makeChannel({ + promptImpl: async (req: PromptRequest) => { + const text = (req.prompt[0] as { text?: string }).text; + if (text === 'blocker') await firstDone; + if (text === 'queued then running') await secondDone; + return { stopReason: 'end_turn' } as PromptResponse; + }, + }); + const bridge = makeBridge({ channelFactory: async () => handle.channel }); + const session = await bridge.spawnOrAttach({ workspaceCwd: WS_A }); + const events: BridgeEvent[] = []; + subscribe(bridge, session.sessionId, events); + + // Prompt 1 starts running immediately; prompt 2 queues behind it. + const p1 = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'blocker' }], + }, + undefined, + { promptId: 'prompt-blocker' }, + ); + const p2 = bridge.sendPrompt( + session.sessionId, + { + sessionId: session.sessionId, + prompt: [{ type: 'text', text: 'queued then running' }], + }, + undefined, + { promptId: 'prompt-promoted' }, + ); + await new Promise((r) => setTimeout(r, 20)); + + // Prompt 2 is queued behind prompt 1. + expect( + bridge + .getPendingPrompts(session.sessionId) + .find((p) => p.promptId === 'prompt-promoted')?.state, + ).toBe('queued'); + + // Release prompt 1 so prompt 2 promotes to running. + releaseFirst!(); + await p1; + await new Promise((r) => setTimeout(r, 20)); + expect( + bridge + .getPendingPrompts(session.sessionId) + .find((p) => p.promptId === 'prompt-promoted')?.state, + ).toBe('running'); + + // Remove the now-running prompt 2. + expect( + bridge.removePendingPrompt(session.sessionId, 'prompt-promoted'), + ).toEqual({ removed: true }); + + // Let prompt 2 settle cooperatively. + releaseSecond!(); + await p2; + await new Promise((r) => setTimeout(r, 20)); + + // Exactly one pending_prompt_completed for prompt-promoted: the + // 'removed' one from removePendingPrompt. The result.finally path + // must NOT publish a second 'completed' event because the + // isQueued && !pendingEntry.removed guard suppresses it. + const completedForPromoted = events.filter( + (e) => + e.type === 'pending_prompt_completed' && + (e as BridgeEvent & { data: { promptId: string } }).data.promptId === + 'prompt-promoted', + ); + expect(completedForPromoted).toHaveLength(1); + expect( + (completedForPromoted[0] as BridgeEvent & { data: { state: string } }) + .data.state, + ).toBe('removed'); + + // The formal terminal is still published exactly once. + expect(terminalsFor(events, 'prompt-promoted')).toHaveLength(1); + + await bridge.shutdown(); + }); + it('flushes error terminals for active and queued prompts before session_died on killSession (DAEMON-005)', async () => { const handle = wedgeChannel(); const bridge = makeBridge({ channelFactory: async () => handle.channel });