diff --git a/packages/agent-runtime/src/pi/bridge/__tests__/bridge.test.ts b/packages/agent-runtime/src/pi/bridge/__tests__/bridge.test.ts index 788b5aee9f..30bd551c4f 100644 --- a/packages/agent-runtime/src/pi/bridge/__tests__/bridge.test.ts +++ b/packages/agent-runtime/src/pi/bridge/__tests__/bridge.test.ts @@ -212,6 +212,15 @@ function threadEvents( return assembleCapturedThreadEvents(messages); } +/** Turn lifecycle only, without the provider diagnostics around it. */ +function turnEvents( + messages: readonly BridgeJsonRpcOutputMessage[], +): ThreadEvent[] { + return threadEvents(messages).filter((event) => + event.type.startsWith("turn/"), + ); +} + interface ControlledPiAgentSession { abort: ReturnType; bindExtensions: ReturnType; @@ -220,6 +229,7 @@ interface ControlledPiAgentSession { emit(event: AgentSessionEvent): void; extensionRunner: { emit: ReturnType }; finishAbort(): void; + finishPrompt(): void; getActiveToolNames: ReturnType; getContextUsage: ReturnType; hasExtensionHandlers: ReturnType; @@ -233,6 +243,7 @@ interface ControlledPiAgentSession { function createControlledPiAgentSession(): ControlledPiAgentSession { let finishAbort: (() => void) | undefined; + let finishPrompt: (() => void) | undefined; let extensionShutdownHandler: (() => void) | undefined; const listeners: ControlledPiAgentSessionListener[] = []; const abort = vi.fn( @@ -263,11 +274,30 @@ function createControlledPiAgentSession(): ControlledPiAgentSession { finishAbort(); finishAbort = undefined; }, + finishPrompt() { + if (!finishPrompt) { + throw new Error("Expected Pi prompt to be running"); + } + finishPrompt(); + finishPrompt = undefined; + }, getActiveToolNames: vi.fn(() => []), getContextUsage: vi.fn(() => undefined), hasExtensionHandlers: vi.fn(() => false), isStreaming: false, - prompt: vi.fn(async () => {}), + // Pi accepts the prompt in preflight and only settles when the run it + // started ends, so a dispatched prompt stays open until a test finishes it. + prompt: vi.fn( + async ( + _text: string, + options?: { preflightResult?: (accepted: boolean) => void }, + ) => { + options?.preflightResult?.(true); + await new Promise((resolve) => { + finishPrompt = resolve; + }); + }, + ), requestExtensionShutdown(): void { if (!extensionShutdownHandler) { throw new Error("Expected Pi extension shutdown handler to be bound"); @@ -290,11 +320,12 @@ function createControlledPiAgentSession(): ControlledPiAgentSession { function createQueueUpdateEvent( steering: readonly string[], + followUp: readonly string[] = [], ): AgentSessionEvent { return { type: "queue_update", steering, - followUp: [], + followUp, }; } @@ -1137,9 +1168,10 @@ describe("pi bridge", () => { ); await bridge.flushWork(); - expect(piSession.prompt).toHaveBeenCalledWith("interrupting steer", { - streamingBehavior: "steer", - }); + expect(piSession.prompt).toHaveBeenCalledWith( + "interrupting steer", + expect.objectContaining({ streamingBehavior: "steer" }), + ); await expect(bridge.waitForResponse(22)).resolves.toMatchObject({ id: 22, result: { threadId: "thread-steer-consumption" }, @@ -1170,12 +1202,16 @@ describe("pi bridge", () => { ]), ); await bridge.waitForResponse(51); + piSession.finishPrompt(); await bridge.flushWork(); // The accepted input opened a turn the SDK never worked on; the settle // signal must still close it, or the runtime waits forever. expect(threadEvents(bridge.messages)).toContainEqual( - expect.objectContaining({ type: "turn/completed", status: "completed" }), + expect.objectContaining({ + type: "turn/completed", + status: "completed", + }), ); } finally { bridge.restore(); @@ -1264,6 +1300,66 @@ describe("pi bridge", () => { } }); + it("holds a turn/start pi queued behind a live run until pi reads it", async () => { + const bridge = createBridgeJsonRpcTestHarness(handleLine); + const piSession = createControlledPiAgentSession(); + piSession.isStreaming = true; + // Pi queues a prompt that arrives while a run is still live and returns + // straight away: the dispatch call settling is not the run settling. + piSession.prompt.mockImplementation( + async ( + _text: string, + options?: { preflightResult?: (accepted: boolean) => void }, + ) => { + piSession.emit(createQueueUpdateEvent([], ["queued prompt"])); + options?.preflightResult?.(true); + }, + ); + mockCreateAgentSession.mockImplementation(async () => ({ + session: piSession, + })); + + try { + bridge.sendRequest( + 80, + "thread/start", + sessionParams({ threadId: "thread-queued-turn" }), + ); + await bridge.waitForResponse(80); + + bridge.sendRequest( + 81, + "turn/start", + turnStartParams("thread-queued-turn", [ + { type: "text", text: "queued prompt" }, + ]), + ); + await bridge.flushWork(); + await bridge.flushWork(); + + // Accepting queued input lets the queue-time settle report claim it and + // complete an empty turn for a message pi has not read (#2014). + expect(turnEvents(bridge.messages)).toEqual([]); + + piSession.emit(createQueueUpdateEvent([], [])); + await expect(bridge.waitForResponse(81)).resolves.toMatchObject({ + id: 81, + result: { threadId: "thread-queued-turn" }, + }); + + piSession.emit({ type: "agent_start" }); + await bridge.flushWork(); + + // The acceptance lands in the turn pi opened for the input it read. + expect(turnEvents(bridge.messages)).toEqual([ + expect.objectContaining({ type: "turn/started" }), + expect.objectContaining({ type: "turn/input/accepted" }), + ]); + } finally { + bridge.restore(); + } + }); + it("emits an error when a queued steer is not consumed before agent end", async () => { const bridge = createBridgeJsonRpcTestHarness(handleLine); const piSession = createControlledPiAgentSession(); diff --git a/packages/agent-runtime/src/pi/bridge/__tests__/sdk-session.test.ts b/packages/agent-runtime/src/pi/bridge/__tests__/sdk-session.test.ts index 132bcb58b8..78c3094f4c 100644 --- a/packages/agent-runtime/src/pi/bridge/__tests__/sdk-session.test.ts +++ b/packages/agent-runtime/src/pi/bridge/__tests__/sdk-session.test.ts @@ -235,14 +235,34 @@ function emitSessionEvent(event: AgentSessionEvent): void { function createQueueUpdateEvent( steering: readonly string[], + followUp: readonly string[] = [], ): AgentSessionEvent { return { type: "queue_update", steering, - followUp: [], + followUp, }; } +/** + * Every dispatch installs pi's preflight hook: it is how the session learns + * that pi took an input it did not queue. + */ +function withPreflight( + options: Record = {}, +): Record { + return { ...options, preflightResult: expect.any(Function) }; +} + +/** Report pi's preflight acceptance for the most recent dispatch. */ +function reportPreflightAccepted(accepted = true): void { + const call = mockPrompt.mock.calls.at(-1); + const options = call?.[1] as + | { preflightResult?: (accepted: boolean) => void } + | undefined; + options?.preflightResult?.(accepted); +} + function createAgentEndEvent(willRetry = false): AgentSessionEvent { return { type: "agent_end", @@ -284,6 +304,7 @@ async function flushDeferredSteerSettlement(): Promise { describe("PiSdkSession", () => { beforeEach(() => { vi.clearAllMocks(); + mockPrompt.mockReset(); mockSessionState.isStreaming = false; mockSessionEventListeners.length = 0; mockGetActiveToolNames.mockReturnValue([]); @@ -669,8 +690,8 @@ describe("PiSdkSession", () => { ); await session.start(); - await session.prompt("first follow-up"); - await session.prompt("second follow-up"); + await session.prompt("first follow-up").settled; + await session.prompt("second follow-up").settled; expect(mockSetActiveToolsByName).toHaveBeenCalledTimes(2); expect(mockSetActiveToolsByName).toHaveBeenNthCalledWith(1, [ @@ -690,11 +711,57 @@ describe("PiSdkSession", () => { const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), vi.fn()); await session.start(); - await session.prompt("queued follow-up"); + session.prompt("queued follow-up"); + + expect(mockPrompt).toHaveBeenCalledWith( + "queued follow-up", + withPreflight({ streamingBehavior: "followUp" }), + ); + }); + + it("accepts a queued follow-up prompt only once pi reads it", async () => { + mockSessionState.isStreaming = true; + mockPrompt.mockImplementationOnce(async () => { + emitSessionEvent(createQueueUpdateEvent([], ["expanded follow-up"])); + reportPreflightAccepted(); + }); + const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), vi.fn()); + + await session.start(); + const dispatch = session.prompt("queued follow-up"); + let consumed = false; + void dispatch.consumed.then(() => { + consumed = true; + }); + await flushAsyncWork(); + + // Pi queued the prompt behind the live run: it has not read the input, and + // the run it lands in reports its own settlement. + expect(consumed).toBe(false); + await expect(dispatch.settled).resolves.toBeNull(); + + emitSessionEvent(createQueueUpdateEvent([], [])); + await expect(dispatch.consumed).resolves.toBeUndefined(); + }); - expect(mockPrompt).toHaveBeenCalledWith("queued follow-up", { - streamingBehavior: "followUp", + it("accepts an unqueued prompt when pi reports preflight acceptance", async () => { + let releaseRun: (() => void) | undefined; + mockPrompt.mockImplementationOnce(async () => { + reportPreflightAccepted(); + await new Promise((resolve) => { + releaseRun = resolve; + }); }); + const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), vi.fn()); + + await session.start(); + const dispatch = session.prompt("direct prompt"); + + // Pi started the run with the input, so the turn is accepted long before + // the run it started settles. + await expect(dispatch.consumed).resolves.toBeUndefined(); + releaseRun?.(); + await expect(dispatch.settled).resolves.toEqual({}); }); it("resolves queued steer once the SDK accepts it and monitors consumption", async () => { @@ -712,9 +779,10 @@ describe("PiSdkSession", () => { }); await steerPromise; - expect(mockPrompt).toHaveBeenCalledWith("interrupting steer", { - streamingBehavior: "steer", - }); + expect(mockPrompt).toHaveBeenCalledWith( + "interrupting steer", + withPreflight({ streamingBehavior: "steer" }), + ); expect(steerAccepted).toBe(true); expect(onDone).not.toHaveBeenCalled(); @@ -731,9 +799,10 @@ describe("PiSdkSession", () => { await session.start(); await session.steer("handled steer"); - expect(mockPrompt).toHaveBeenCalledWith("handled steer", { - streamingBehavior: "steer", - }); + expect(mockPrompt).toHaveBeenCalledWith( + "handled steer", + withPreflight({ streamingBehavior: "steer" }), + ); }); it("rejects steer consumption when the SDK prompt rejects", async () => { @@ -814,6 +883,36 @@ describe("PiSdkSession", () => { ); }); + it("keeps a queued follow-up pending past the agent end that continues into it", async () => { + mockSessionState.isStreaming = true; + mockPrompt.mockImplementationOnce(async () => { + emitSessionEvent(createQueueUpdateEvent([], ["queued follow-up"])); + }); + const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), vi.fn()); + + await session.start(); + const dispatch = session.prompt("queued follow-up"); + let settledConsumption: "consumed" | "failed" | undefined; + void dispatch.consumed.then( + () => { + settledConsumption = "consumed"; + }, + () => { + settledConsumption = "failed"; + }, + ); + + // Pi drains its follow-up queue by continuing the same run after + // agent_end, so that event is terminal for steering only. + emitSessionEvent(createAgentEndEvent()); + await flushDeferredSteerSettlement(); + + expect(settledConsumption).toBeUndefined(); + + emitSessionEvent(createQueueUpdateEvent([], [])); + await expect(dispatch.consumed).resolves.toBeUndefined(); + }); + it("keeps queued steer consumption pending when auto retry starts", async () => { mockSessionState.isStreaming = true; mockPrompt.mockImplementationOnce(async () => { @@ -867,11 +966,19 @@ describe("PiSdkSession", () => { const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), vi.fn()); await session.start(); - await session.prompt("idle follow-up"); + session.prompt("idle follow-up"); await session.steer("idle steer"); - expect(mockPrompt).toHaveBeenNthCalledWith(1, "idle follow-up", {}); - expect(mockPrompt).toHaveBeenNthCalledWith(2, "idle steer", {}); + expect(mockPrompt).toHaveBeenNthCalledWith( + 1, + "idle follow-up", + withPreflight(), + ); + expect(mockPrompt).toHaveBeenNthCalledWith( + 2, + "idle steer", + withPreflight(), + ); }); it("reports pending steer consumption failure when the session closes", async () => { @@ -891,7 +998,7 @@ describe("PiSdkSession", () => { expect(onDone).toHaveBeenCalledTimes(1); expect(onDone).toHaveBeenCalledWith( expect.objectContaining({ - message: "Pi SDK session stopped before steer consumed", + message: "Pi SDK session stopped before input was consumed", }), ); }); @@ -904,7 +1011,7 @@ describe("PiSdkSession", () => { const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), onDone); await session.start(); - await session.prompt("retry after auth storage miss"); + await session.prompt("retry after auth storage miss").settled; expect(mockPrompt).toHaveBeenCalledTimes(9); expect(onDone).not.toHaveBeenCalled(); @@ -917,7 +1024,12 @@ describe("PiSdkSession", () => { const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), onDone); await session.start(); - await session.prompt("fail after retry budget"); + const dispatch = session.prompt("fail after retry budget"); + void dispatch.consumed.catch(() => undefined); + await expect(dispatch.settled).resolves.toEqual({ error: authError }); + await expect(dispatch.consumed).rejects.toThrow( + "No API key found for anthropic.", + ); expect(mockPrompt).toHaveBeenCalledTimes(9); expect(onDone).toHaveBeenCalledTimes(1); @@ -927,7 +1039,7 @@ describe("PiSdkSession", () => { it("stays processing across retryable agent-end events", async () => { const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), vi.fn()); await session.start(); - await session.prompt("retry me"); + session.prompt("retry me"); emitSessionEvent(createAgentEndEvent(true)); expect(session.getIsProcessing()).toBe(true); @@ -939,7 +1051,7 @@ describe("PiSdkSession", () => { it("stays processing while Pi performs post-turn streaming work", async () => { const session = new PiSdkSession({ cwd: "/tmp/project" }, vi.fn(), vi.fn()); await session.start(); - await session.prompt("trigger auto compaction"); + session.prompt("trigger auto compaction"); emitSessionEvent(createAgentEndEvent()); mockSessionState.isStreaming = true; diff --git a/packages/agent-runtime/src/pi/bridge/bridge.conformance.test.ts b/packages/agent-runtime/src/pi/bridge/bridge.conformance.test.ts index bb9d3d6c69..9447fc5c56 100644 --- a/packages/agent-runtime/src/pi/bridge/bridge.conformance.test.ts +++ b/packages/agent-runtime/src/pi/bridge/bridge.conformance.test.ts @@ -142,40 +142,48 @@ function createScriptedPiAgentSession(): ScriptedPiAgentSession { getContextUsage: vi.fn(() => undefined), hasExtensionHandlers: vi.fn(() => false), isStreaming: false, - prompt: vi.fn(async (promptText: string) => { - // A prompt the agent handles without emitting a single SDK event: the - // bridge's own pi/prompt/settled report is then the only signal that can - // settle the turn (#1431). - if (promptText === ZERO_WORK_PROMPT_TEXT) { - return; - } - scriptedTurnCounter += 1; - const text = `hello from turn ${scriptedTurnCounter}`; - emit(asPiSdkEvent({ type: "agent_start" })); - emit( - asPiSdkEvent({ - type: "message_update", - assistantMessageEvent: { - type: "text_delta", - contentIndex: 0, - delta: text, - }, - }), - ); - emit( - asPiSdkEvent({ - type: "agent_end", - messages: [ - { - role: "assistant", - content: [{ type: "text", text }], - usage: { input: 12, output: 5 }, + prompt: vi.fn( + async ( + promptText: string, + options?: { preflightResult?: (accepted: boolean) => void }, + ) => { + // Pi accepts a prompt it is about to run in preflight, which is what + // tells the bridge the input was consumed rather than queued. + options?.preflightResult?.(true); + // A prompt the agent handles without emitting a single SDK event: the + // bridge's own pi/prompt/settled report is then the only signal that + // can settle the turn (#1431). + if (promptText === ZERO_WORK_PROMPT_TEXT) { + return; + } + scriptedTurnCounter += 1; + const text = `hello from turn ${scriptedTurnCounter}`; + emit(asPiSdkEvent({ type: "agent_start" })); + emit( + asPiSdkEvent({ + type: "message_update", + assistantMessageEvent: { + type: "text_delta", + contentIndex: 0, + delta: text, }, - ], - willRetry: false, - }), - ); - }), + }), + ); + emit( + asPiSdkEvent({ + type: "agent_end", + messages: [ + { + role: "assistant", + content: [{ type: "text", text }], + usage: { input: 12, output: 5 }, + }, + ], + willRetry: false, + }), + ); + }, + ), sessionManager: { getLeafId: vi.fn(() => "pi-conformance-checkpoint") }, setActiveToolsByName: vi.fn(), subscribe: vi.fn((listener: (event: AgentSessionEvent) => void) => { diff --git a/packages/agent-runtime/src/pi/bridge/bridge.ts b/packages/agent-runtime/src/pi/bridge/bridge.ts index 49fa8a9f2f..e446a1fb5b 100644 --- a/packages/agent-runtime/src/pi/bridge/bridge.ts +++ b/packages/agent-runtime/src/pi/bridge/bridge.ts @@ -635,7 +635,7 @@ async function handleRequest( await handleThreadFork(request.id, request.params); break; case "turn/start": - handleTurnStart(request.id, request.params); + await handleTurnStart(request.id, request.params); break; case "turn/steer": await handleTurnSteer(request.id, request.params); @@ -834,27 +834,33 @@ async function handleThreadFork( ); } +/** + * Dispatch turn input and report the settlement of the run it starts. The + * returned promise resolves once pi consumed the input. + */ function startPiPrompt( threadSession: ThreadSession, threadId: string, text: string, images: ImageContent[], -): void { - void threadSession.session - .prompt(text, images.length > 0 ? images : undefined) - .then( - () => - reportPromptSettled({ - sessionSerial: threadSession.sessionSerial, - threadId, - }), - (error: unknown) => - reportPromptSettled({ - error, - sessionSerial: threadSession.sessionSerial, - threadId, - }), - ); +): Promise { + const dispatch = threadSession.session.prompt( + text, + images.length > 0 ? images : undefined, + ); + void dispatch.settled.then((outcome) => { + // Input pi queued into a run it did not start has no settlement of its + // own. Reporting one anyway settles whichever turn is open when it lands. + if (outcome === null) { + return; + } + reportPromptSettled({ + ...(outcome.error !== undefined ? { error: outcome.error } : {}), + sessionSerial: threadSession.sessionSerial, + threadId, + }); + }); + return dispatch.consumed; } /** @@ -885,8 +891,10 @@ function startPiCompaction( } /** - * Accepted-input correlation (turn/input/accepted): the assembler owns the - * queue-until-turn-opens behavior, so the bridge only reports the acceptance. + * Accepted-input correlation (turn/input/accepted): acceptance means pi + * consumed the input, never that bb handed it over, so every caller reports it + * only after pi read the input. The assembler owns the queue-until-turn-opens + * behavior, so the bridge only reports the acceptance. */ function recordAcceptedTurnInput(params: TurnStartParams): void { sendThreadDeltas(params.threadId, [ @@ -894,7 +902,10 @@ function recordAcceptedTurnInput(params: TurnStartParams): void { ]); } -function handleTurnStart(id: string | number, params: TurnStartParams): void { +async function handleTurnStart( + id: string | number, + params: TurnStartParams, +): Promise { // Requests resolve the session by bb threadId — pi's stable session handle. const threadSession = sessions.get(params.threadId); if (!threadSession || threadSession.closing) { @@ -918,9 +929,18 @@ function handleTurnStart(id: string | number, params: TurnStartParams): void { return; } - recordAcceptedTurnInput(params); - startPiPrompt(threadSession, params.threadId, text, images); - sendResult(id, { threadId: params.threadId }); + try { + await startPiPrompt(threadSession, params.threadId, text, images); + // Like steer, a new turn is accepted only once pi read the input. Pi + // queues a prompt that arrives while a run is still unwinding, and that + // run's settle report would otherwise claim the queued input and complete + // an empty turn for a message pi has not answered yet. + recordAcceptedTurnInput(params); + sendResult(id, { threadId: params.threadId }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + sendError(id, -32000, message); + } } async function handleTurnSteer( @@ -949,8 +969,11 @@ async function handleTurnSteer( text, images.length > 0 ? images : undefined, ); - // A steer joins the active turn; its acceptance is reported only once - // the SDK actually accepted the queued input. + // A steer joins the turn the assembler already holds open, so its + // acceptance can never be the pending claim a stale terminal takes. It is + // reported once the SDK took the steering message: pi delivers steering + // only between assistant turns, and waiting for that would leave the + // steered message unrendered for the length of the running tool call. sendThreadDeltas(params.threadId, [ { kind: "input.accepted", clientRequestId: params.clientRequestId }, ]); diff --git a/packages/agent-runtime/src/pi/bridge/sdk-session.ts b/packages/agent-runtime/src/pi/bridge/sdk-session.ts index 32390fed65..f02b7bd9c9 100644 --- a/packages/agent-runtime/src/pi/bridge/sdk-session.ts +++ b/packages/agent-runtime/src/pi/bridge/sdk-session.ts @@ -38,25 +38,53 @@ type AppendSystemPromptOverride = (base: string[]) => string[]; interface RunPromptArgs { images?: ImageContent[]; + pending: PendingInputConsumption; streamingBehavior: PiStreamingBehavior; text: string; } -interface RunPromptResult { - steerConsumptionPromise: Promise | null; -} +/** + * Which of pi's two input queues holds a dispatch pi did not run immediately. + * A steering message interrupts the current assistant turn; a follow-up waits + * for it. They drain independently, so a dispatch is correlated against the + * queue it was placed in. + */ +type PiInputQueue = "followUp" | "steering"; -interface PendingSteerConsumption { +interface PendingInputConsumption { + queue: PiInputQueue; queuedText: string | null; reject: (error: Error) => void; resolve: () => void; } -interface TrackedSteerConsumption { - pending: PendingSteerConsumption; +interface TrackedInputConsumption { + pending: PendingInputConsumption; promise: Promise; } +/** The outcome of an agent run pi started for a dispatched input. */ +export interface PiPromptRunOutcome { + /** Omitted when the run finished without a fatal error. */ + error?: unknown; +} + +/** How pi took a dispatched turn input. */ +export interface PiInputDispatch { + /** + * Resolves once pi consumed the input — it started a run with it, or the + * queue that held it delivered it. Rejects when pi refused the input or the + * session ended before delivery. + */ + consumed: Promise; + /** + * The outcome of the run pi started for this input, or `null` when pi queued + * the input into a run it did not start. That run's own dispatch reports its + * settlement; a second report would settle whatever turn is open by then. + */ + settled: Promise; +} + type PiStreamingBehavior = NonNullable; const PI_TRANSIENT_AUTH_RETRY_DELAY_MS = 250; @@ -183,8 +211,11 @@ export class PiSdkSession { private isProcessing = false; private isCompacting = false; private manualCompactionCompletionCount = 0; - private readonly pendingSteerConsumptions: PendingSteerConsumption[] = []; - private lastObservedSteeringQueue: string[] = []; + private readonly pendingInputConsumptions: PendingInputConsumption[] = []; + private lastObservedQueues: Record = { + followUp: [], + steering: [], + }; private autoRetryInProgress = false; private terminalSteerSettlementTimeout: | ReturnType @@ -297,47 +328,89 @@ export class PiSdkSession { // Subscribe to session events this.unsubscribe = session.subscribe((event: AgentSessionEvent) => { this.trackProcessingState(event); - this.observeSteerConsumption(event); + this.observeInputConsumption(event); this.observeTerminalSteerSettlement(event); this.onEvent(event); }); } - async prompt(text: string, images?: ImageContent[]): Promise { - if (!this.session) return; - this.isProcessing = true; - try { - await this.runPromptWithTransientAuthRetry({ - images, - streamingBehavior: "followUp", - text, - }); - } catch (error) { - this.isProcessing = false; - this.rejectPendingSteerConsumptions( - "Pi SDK prompt failed before steer consumed", - ); - this.onDone(error); + /** + * Dispatch turn input. Pi either starts a run with it or, when a run is + * already live, queues it as a follow-up — so the caller learns consumption + * and settlement separately instead of treating the dispatch call returning + * as either one. + */ + prompt(text: string, images?: ImageContent[]): PiInputDispatch { + if (!this.session) { + const consumed = Promise.reject(new Error("No active Pi SDK session")); + // A caller that only watches settlement must not turn this into an + // unhandled rejection. + void consumed.catch(() => undefined); + return { consumed, settled: Promise.resolve(null) }; } + this.isProcessing = true; + const tracked = this.trackPendingInputConsumption("followUp"); + const settled = this.runPromptWithTransientAuthRetry({ + images, + pending: tracked.pending, + streamingBehavior: "followUp", + text, + }).then( + (): PiPromptRunOutcome | null => { + if (tracked.pending.queuedText !== null) { + return null; + } + // Pi handled the input without queueing it and without a preflight + // report (an SDK that predates the hook, or a prompt handled before + // preflight): the returned call is then the only consumption signal. + this.resolvePendingInputConsumption(tracked.pending); + return {}; + }, + (error: unknown): PiPromptRunOutcome | null => { + this.isProcessing = false; + const queued = tracked.pending.queuedText !== null; + this.rejectPendingInputConsumption(tracked.pending, asError(error)); + this.rejectPendingInputConsumptions( + "Pi SDK prompt failed before input was consumed", + ); + this.onDone(error); + return queued ? null : { error }; + }, + ); + return { consumed: tracked.promise, settled }; } + /** + * Steer the live run. Resolves once the SDK took the steering message, not + * once it read it: pi delivers steering between assistant turns, so a steer + * sent during a long tool call waits for that tool, and the runtime's + * 30-second command timeout would fail a steer that is still on its way. + * A steer the run ends without reading is reported through `onDone`. + */ async steer(text: string, images?: ImageContent[]): Promise { if (!this.session) { throw new Error("No active Pi SDK session"); } + const tracked = this.trackPendingInputConsumption("steering"); try { - const result = await this.runPromptWithTransientAuthRetry({ + await this.runPromptWithTransientAuthRetry({ images, + pending: tracked.pending, streamingBehavior: "steer", text, }); - if (result.steerConsumptionPromise) { - this.monitorSteerConsumption(result.steerConsumptionPromise); - } } catch (error) { + this.rejectPendingInputConsumption(tracked.pending, asError(error)); this.onDone(error); throw error; } + if (tracked.pending.queuedText === null) { + // Pi handled the steer without queueing it, so no delivery event is + // coming: the returned call is the consumption signal. + this.resolvePendingInputConsumption(tracked.pending); + return; + } + this.monitorSteerConsumption(tracked.promise); } async compact(): Promise { @@ -366,8 +439,8 @@ export class PiSdkSession { } detach(): void { - this.rejectPendingSteerConsumptions( - "Pi SDK session detached before steer consumed", + this.rejectPendingInputConsumptions( + "Pi SDK session detached before input was consumed", ); if (this.unsubscribe) { this.unsubscribe(); @@ -378,8 +451,8 @@ export class PiSdkSession { } stop(): void { - this.rejectPendingSteerConsumptions( - "Pi SDK session stopped before steer consumed", + this.rejectPendingInputConsumptions( + "Pi SDK session stopped before input was consumed", ); this.detach(); const session = this.session; @@ -389,8 +462,8 @@ export class PiSdkSession { async closeGracefully(timeoutMs: number): Promise { const session = this.session; - this.rejectPendingSteerConsumptions( - "Pi SDK session closed before steer consumed", + this.rejectPendingInputConsumptions( + "Pi SDK session closed before input was consumed", ); this.detach(); if (!session) { @@ -453,7 +526,9 @@ export class PiSdkSession { } } - private trackPendingSteerConsumption(): TrackedSteerConsumption { + private trackPendingInputConsumption( + queue: PiInputQueue, + ): TrackedInputConsumption { let resolvePromise: (() => void) | undefined; let rejectPromise: ((error: Error) => void) | undefined; const promise = new Promise((resolve, reject) => { @@ -461,39 +536,45 @@ export class PiSdkSession { rejectPromise = reject; }); if (!resolvePromise || !rejectPromise) { - throw new Error("Failed to track Pi steer consumption"); + throw new Error("Failed to track Pi input consumption"); } - const pending: PendingSteerConsumption = { + const pending: PendingInputConsumption = { + queue, queuedText: null, reject: rejectPromise, resolve: resolvePromise, }; - this.pendingSteerConsumptions.push(pending); + this.pendingInputConsumptions.push(pending); void promise.catch(() => undefined); return { pending, promise }; } - private observeSteerConsumption(event: AgentSessionEvent): void { + private observeInputConsumption(event: AgentSessionEvent): void { if (event.type !== "queue_update") { return; } + this.observeQueue("steering", event.steering); + this.observeQueue("followUp", event.followUp); + } - const addedQueuedTexts = listMultisetDifference( - event.steering, - this.lastObservedSteeringQueue, - ); + private observeQueue( + queue: PiInputQueue, + queuedTexts: readonly string[], + ): void { + const lastObserved = this.lastObservedQueues[queue]; + const addedQueuedTexts = listMultisetDifference(queuedTexts, lastObserved); const removedQueuedTexts = listMultisetDifference( - this.lastObservedSteeringQueue, - event.steering, + lastObserved, + queuedTexts, ); - this.lastObservedSteeringQueue = [...event.steering]; + this.lastObservedQueues[queue] = [...queuedTexts]; for (const queuedText of addedQueuedTexts) { // Pi queue_update exposes SDK-transformed text, so correlate by FIFO queue - // additions rather than by the raw BB steer text. - const pending = this.pendingSteerConsumptions.find( - (entry) => entry.queuedText === null, + // additions rather than by the raw BB input text. + const pending = this.pendingInputConsumptions.find( + (entry) => entry.queue === queue && entry.queuedText === null, ); if (!pending) { break; @@ -502,11 +583,13 @@ export class PiSdkSession { } for (const queuedText of removedQueuedTexts) { - const pending = this.pendingSteerConsumptions.find( - (entry) => entry.queuedText === queuedText, + // Pi drops a queued message from the queue when it reads it into the + // conversation, which is the only moment the input is truly consumed. + const pending = this.pendingInputConsumptions.find( + (entry) => entry.queue === queue && entry.queuedText === queuedText, ); if (pending) { - this.resolvePendingSteerConsumption(pending); + this.resolvePendingInputConsumption(pending); } } } @@ -528,16 +611,22 @@ export class PiSdkSession { if (event.type === "auto_retry_end") { this.autoRetryInProgress = false; if (!event.success) { - this.rejectPendingSteerConsumptions( + this.rejectPendingInputConsumptions( "Pi auto retry ended before steer was consumed", + "steering", ); } } } private scheduleTerminalSteerSettlement(): void { + // Only steering is terminal at agent_end: pi drains its follow-up queue by + // continuing the same run after that event, so a queued follow-up is still + // on its way to being read. if ( - this.pendingSteerConsumptions.length === 0 || + !this.pendingInputConsumptions.some( + (entry) => entry.queue === "steering", + ) || this.terminalSteerSettlementTimeout !== undefined ) { return; @@ -548,8 +637,9 @@ export class PiSdkSession { if (this.autoRetryInProgress) { return; } - this.rejectPendingSteerConsumptions( + this.rejectPendingInputConsumptions( "Pi turn ended before steer was consumed", + "steering", ); }, 0); } @@ -562,33 +652,41 @@ export class PiSdkSession { this.terminalSteerSettlementTimeout = undefined; } - private resolvePendingSteerConsumption( - pending: PendingSteerConsumption, + private resolvePendingInputConsumption( + pending: PendingInputConsumption, ): void { - const index = this.pendingSteerConsumptions.indexOf(pending); + const index = this.pendingInputConsumptions.indexOf(pending); if (index === -1) { return; } - this.pendingSteerConsumptions.splice(index, 1); + this.pendingInputConsumptions.splice(index, 1); pending.resolve(); } - private rejectPendingSteerConsumption( - pending: PendingSteerConsumption, + /** Reports whether this call was the one that settled the consumption. */ + private rejectPendingInputConsumption( + pending: PendingInputConsumption, error: Error, - ): void { - const index = this.pendingSteerConsumptions.indexOf(pending); + ): boolean { + const index = this.pendingInputConsumptions.indexOf(pending); if (index === -1) { - return; + return false; } - this.pendingSteerConsumptions.splice(index, 1); + this.pendingInputConsumptions.splice(index, 1); pending.reject(error); + return true; } - private rejectPendingSteerConsumptions(message: string): void { + private rejectPendingInputConsumptions( + message: string, + queue?: PiInputQueue, + ): void { this.clearTerminalSteerSettlement(); - const pendingSteers = this.pendingSteerConsumptions.splice(0); - for (const pending of pendingSteers) { + for (const pending of this.pendingInputConsumptions.splice(0)) { + if (queue !== undefined && pending.queue !== queue) { + this.pendingInputConsumptions.push(pending); + continue; + } pending.reject(new Error(message)); } } @@ -624,10 +722,11 @@ export class PiSdkSession { private async runPromptWithTransientAuthRetry( args: RunPromptArgs, - ): Promise { + ): Promise { for (let attempt = 0; ; attempt += 1) { try { - return await this.runPromptOnce(args); + await this.runPromptOnce(args); + return; } catch (error) { if ( !(error instanceof Error) || @@ -641,44 +740,35 @@ export class PiSdkSession { } } - private async runPromptOnce(args: RunPromptArgs): Promise { + private async runPromptOnce(args: RunPromptArgs): Promise { if (!this.session) { throw new Error("No active Pi SDK session"); } this.ensureCustomToolsActive(); - if (this.session.isStreaming) { - const steerConsumption = - args.streamingBehavior === "steer" - ? this.trackPendingSteerConsumption() - : null; - try { - await this.session.prompt(args.text, { - streamingBehavior: args.streamingBehavior, - ...(args.images && args.images.length > 0 - ? { images: args.images } - : {}), - }); - } catch (error) { - if (steerConsumption) { - this.rejectPendingSteerConsumption( - steerConsumption.pending, - error instanceof Error ? error : new Error(String(error)), - ); - } - throw error; - } - if (steerConsumption && steerConsumption.pending.queuedText === null) { - this.resolvePendingSteerConsumption(steerConsumption.pending); - } - return { steerConsumptionPromise: steerConsumption?.promise ?? null }; - } + const pending = args.pending; await this.session.prompt(args.text, { + ...(this.session.isStreaming + ? { streamingBehavior: args.streamingBehavior } + : {}), ...(args.images && args.images.length > 0 ? { images: args.images } : {}), + // Pi reports preflight acceptance after it queued the input and before + // it starts a run with it. Input pi did not queue is therefore consumed + // the moment preflight accepts it; queued input waits for its queue to + // deliver it. Nothing else reports the start of a run pi handles without + // emitting a single SDK event. + preflightResult: (accepted: boolean) => { + if (accepted && pending.queuedText === null) { + this.resolvePendingInputConsumption(pending); + } + }, }); - return { steerConsumptionPromise: null }; } } +function asError(error: unknown): Error { + return error instanceof Error ? error : new Error(String(error)); +} + type PiModel = NonNullable>; /** diff --git a/packages/host-daemon-contract/src/protocol.ts b/packages/host-daemon-contract/src/protocol.ts index e711023a09..0b30a18894 100644 --- a/packages/host-daemon-contract/src/protocol.ts +++ b/packages/host-daemon-contract/src/protocol.ts @@ -1,3 +1,12 @@ +// Version 141 extends the consumed-not-queued acceptance rule to the remaining +// providers. Pi reports `input.accepted` for a turn only once it read the +// input: a prompt pi queues behind a live run stays unaccepted, and the +// queue-time settle report that used to accompany it is gone, so it can no +// longer complete an empty turn for a message pi has not answered. ACP reports +// acceptance once the `session/prompt` request carrying the input goes out, so +// a steer the turn drops is no longer reported as accepted. Older daemons emit +// the queue-time semantics and produce those phantom turns. +// // Version 140 reports each daemon's browser-local editor helper port during // session open. The server uses those ports to let a remote browser discover // the helper on its own machine instead of assuming every machine uses the @@ -71,7 +80,7 @@ // // The version mismatch is what triggers the enrolled daemon's automatic update // instead of an `invalid-message` reconnect loop. -export const HOST_DAEMON_PROTOCOL_VERSION = 140 as const; +export const HOST_DAEMON_PROTOCOL_VERSION = 141 as const; /** * Absolute ceiling for any executable artifact delivered to a host daemon — diff --git a/packages/host-daemon-contract/test/contract.test.ts b/packages/host-daemon-contract/test/contract.test.ts index 1e1f4aff80..e808f2b4c7 100644 --- a/packages/host-daemon-contract/test/contract.test.ts +++ b/packages/host-daemon-contract/test/contract.test.ts @@ -1138,7 +1138,7 @@ describe("host-daemon command schemas", () => { // mixed version. Version 113 carried the Devin Desktop open target rename // and remains part of the protocol lineage. it("uses the current host-daemon protocol version", () => { - expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(140); + expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(141); expect(HOST_ARTIFACT_MAX_BYTES).toBe(256 * 1024 * 1024); }); diff --git a/plugins/provider-acp/src/bridge/bridge.test.ts b/plugins/provider-acp/src/bridge/bridge.test.ts index a3c9a3e801..3d5a4b9053 100644 --- a/plugins/provider-acp/src/bridge/bridge.test.ts +++ b/plugins/provider-acp/src/bridge/bridge.test.ts @@ -12,7 +12,10 @@ import { fileURLToPath } from "node:url"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { createStandaloneBuiltinCompactCommandInput } from "@bb/domain"; import type { DynamicTool, ReasoningLevel } from "@bb/domain"; -import { PROVIDER_BRIDGE_PROTOCOL_VERSION } from "@bb/provider-bridge-protocol"; +import { + PROVIDER_BRIDGE_PROTOCOL_VERSION, + THREAD_DELTA_NOTIFICATION_METHOD, +} from "@bb/provider-bridge-protocol"; import { captureBridgeJsonRpcOutput, type BridgeJsonRpcOutputMessage, @@ -100,6 +103,16 @@ function threadEvents(): Record[] { Record[]; } +/** The delta kinds the bridge put on the wire, in emission order. */ +function emittedDeltaKinds(): string[] { + return notifications(THREAD_DELTA_NOTIFICATION_METHOD).flatMap((message) => { + const params = message.params as + | { deltas?: { kind?: string }[] } + | undefined; + return (params?.deltas ?? []).map((delta) => delta.kind ?? ""); + }); +} + function threadEventsOfType(type: string): Record[] { return threadEvents().filter((event) => event.type === type); } @@ -1910,6 +1923,51 @@ describe("acp bridge", () => { expect(threadEventsOfType("thread/compacted")).toEqual([]); }); + it("accepts turn input only after the prompt carrying it goes out", async () => { + const { providerThreadId } = await startThread(); + const turnId = sendTurnRequest("turn/start", providerThreadId, { + input: [{ type: "text", text: "hello there", mentions: [] }], + }); + await waitForResponse(turnId); + await waitForTurnCompleted(); + + // Acceptance means the `session/prompt` request carrying the input went + // out, so the bridge emits it after opening the turn. Emitting it first + // leaves a pending claim on bb's side that any stale terminal can take, + // which is the class #2013 fixed for Claude (#2014). + const deltaKinds = emittedDeltaKinds(); + expect(deltaKinds.indexOf("input.accepted")).toBe( + deltaKinds.indexOf("turn.open") + 1, + ); + }); + + it("never accepts a queued steer the stopped turn did not send", async () => { + const { providerThreadId } = await startThread(); + const turnId = sendTurnRequest("turn/start", providerThreadId, { + input: [{ type: "text", text: "hang", mentions: [] }], + }); + await waitForResponse(turnId); + + const steerId = sendTurnRequest("turn/steer", providerThreadId, { + expectedTurnId: "turn-1", + input: [{ type: "text", text: "never sent", mentions: [] }], + }); + await waitForResponse(steerId); + const stopId = sendRequest("thread/stop", { + threadId: bbThreadIdFor(providerThreadId), + providerThreadId, + intent: "interrupt", + activeTurnId: null, + }); + await waitForResponse(stopId); + await waitForTurnCompleted(); + + // The stop dropped the queued steer before it reached the agent, so the + // turn reports the one input the agent was actually given, not two. + expect(threadEventsOfType("turn/input/accepted")).toHaveLength(1); + startedProviderThreadIds.pop(); + }); + it("rejects steers when no turn is active", async () => { const { providerThreadId } = await startThread(); const steerId = sendTurnRequest("turn/steer", providerThreadId, { diff --git a/plugins/provider-acp/src/bridge/bridge.ts b/plugins/provider-acp/src/bridge/bridge.ts index 53e98784ad..d6b519d495 100644 --- a/plugins/provider-acp/src/bridge/bridge.ts +++ b/plugins/provider-acp/src/bridge/bridge.ts @@ -137,6 +137,24 @@ interface PendingAcpPermission { options: AcpPermissionOption[]; } +/** + * Turn input bb handed the bridge, waiting to reach the agent. ACP has no + * provider acknowledgement to correlate acceptance against, so the designed + * correlation point is the `session/prompt` request that carries the input: + * before that the agent has not seen the input at all, and a queued steer can + * still be dropped by a failed or stopping turn. + */ +interface AcpPendingTurnInput { + clientRequestId: string; + input: PromptInput[]; + /** + * The command to answer once the input reaches the agent, or `null` for a + * steer, which is answered at queue time. Waiting for the cancelled prompt + * to be reissued would risk the runtime's 30-second command timeout. + */ + requestId: AcpBridgeRequestId | null; +} + interface AcpThreadSession { bbThreadId: string; providerThreadId: string; @@ -154,7 +172,7 @@ interface AcpThreadSession { * the provider-local `"compaction"` maintenance prompt, or none. */ activePromptKind: "turn" | "compaction" | null; - queuedInputs: PromptInput[][]; + queuedInputs: AcpPendingTurnInput[]; /** True while a session/prompt request is outstanding. */ promptRequestPending: boolean; /** True after a steer sent session/cancel for the current prompt. */ @@ -202,6 +220,8 @@ const { send, sendResult, sendError } = createBridgeIo< BridgeNotification | BridgeRuntimeRequest >(); +type AcpBridgeRequestId = Parameters[0]; + function sendNotification( method: string, params: Record, @@ -1793,7 +1813,10 @@ async function stopSession(session: AcpThreadSession): Promise { return; } session.stopping = true; - session.queuedInputs = []; + dropQueuedTurnInputs( + session, + "ACP session stopped before the steer was sent", + ); cancelPendingPermissions(session); if (session.activePromptKind !== null && !session.connection.exited) { @@ -1825,7 +1848,10 @@ function releaseSession(session: AcpThreadSession): void { return; } session.stopping = true; - session.queuedInputs = []; + dropQueuedTurnInputs( + session, + "ACP session released before the steer was sent", + ); cancelPendingPermissions(session); session.connection.kill(); removeSession(session); @@ -1851,12 +1877,58 @@ function requestSteerCancel(session: AcpThreadSession): void { }); } +/** + * Accepted-input correlation (turn/input/accepted): the input reached the + * agent, so bb can attach it to the turn it runs in. Reporting acceptance + * before the `session/prompt` request goes out would claim an input the agent + * may never be given. + */ +function acceptTurnInput( + session: AcpThreadSession, + pending: AcpPendingTurnInput, +): void { + sendThreadDeltas(session.bbThreadId, [ + { kind: "input.accepted", clientRequestId: pending.clientRequestId }, + ]); + const requestId = takeTurnInputRequestId(pending); + if (requestId !== null) { + sendResult(requestId, { threadId: session.bbThreadId }); + } +} + +/** + * Report input the turn ended without ever sending. Reply, never drop (#853): + * a command still waiting on the input fails instead of hanging, and no + * acceptance is reported for a turn the agent never received it in. + */ +function dropTurnInput(pending: AcpPendingTurnInput, reason: string): void { + const requestId = takeTurnInputRequestId(pending); + if (requestId !== null) { + sendError(requestId, -32000, reason); + } +} + +/** Answers a command at most once, whatever else happens to the input. */ +function takeTurnInputRequestId( + pending: AcpPendingTurnInput, +): AcpBridgeRequestId | null { + const requestId = pending.requestId; + pending.requestId = null; + return requestId; +} + +function dropQueuedTurnInputs(session: AcpThreadSession, reason: string): void { + for (const pending of session.queuedInputs.splice(0)) { + dropTurnInput(pending, reason); + } +} + function finishTurn( session: AcpThreadSession, stopReason: z.infer, ): void { session.activePromptKind = null; - session.queuedInputs = []; + dropQueuedTurnInputs(session, "ACP turn ended before the steer was sent"); session.promptRequestPending = false; session.cancelRequested = false; emitForSession(session, ACP_TURN_COMPLETED_METHOD, { @@ -1865,16 +1937,20 @@ function finishTurn( }); } -function runTurn(session: AcpThreadSession, firstInput: PromptInput[]): void { +function runTurn( + session: AcpThreadSession, + firstInput: AcpPendingTurnInput, +): void { session.activePromptKind = "turn"; emitForSession(session, ACP_TURN_STARTED_METHOD, { threadId: session.bbThreadId, }); session.turnSettled = (async () => { - let input = firstInput; + let pending = firstInput; for (;;) { if (session.stopping) { + dropTurnInput(pending, "ACP session is stopping"); finishTurn(session, "cancelled"); return; } @@ -1887,10 +1963,14 @@ function runTurn(session: AcpThreadSession, firstInput: PromptInput[]): void { method: "session/prompt", params: { sessionId: session.providerThreadId, - prompt: buildPromptContentBlocks(session, input), + prompt: buildPromptContentBlocks(session, pending.input), }, resultSchema: acpPromptResultSchema, }); + // The agent has the input now, and the turn it runs in is already + // open, so the acceptance names that turn instead of waiting as a + // pending claim any stale terminal could take (#2014). + acceptTurnInput(session, pending); // A steer that stacked behind the cancelled prompt still needs its own // cancel; otherwise this prompt can hang and strand the later input. if (session.queuedInputs.length > 0) { @@ -1900,7 +1980,12 @@ function runTurn(session: AcpThreadSession, firstInput: PromptInput[]): void { stopReason = result.stopReason; } catch (error) { session.promptRequestPending = false; - session.queuedInputs = []; + // Answered already unless the request never went out at all. + dropTurnInput(pending, "ACP turn failed before the prompt was sent"); + dropQueuedTurnInputs( + session, + "ACP turn failed before the steer was sent", + ); session.cancelRequested = false; // An exited agent already produced an error notification from the // connection's exit handler; only report in-protocol prompt failures. @@ -1921,7 +2006,7 @@ function runTurn(session: AcpThreadSession, firstInput: PromptInput[]): void { if (!session.stopping) { const next = session.queuedInputs.shift(); if (next) { - input = next; + pending = next; continue; } } @@ -1953,7 +2038,10 @@ function runTurn(session: AcpThreadSession, firstInput: PromptInput[]): void { * every other stop reason or prompt rejection fails the turn with the agent's * own reason rather than being reported as a shrunk context. */ -function startCompaction(session: AcpThreadSession): void { +function startCompaction( + session: AcpThreadSession, + pending: AcpPendingTurnInput, +): void { session.activePromptKind = "compaction"; emitForSession(session, ACP_COMPACTION_STARTED_METHOD, { threadId: session.bbThreadId, @@ -1968,15 +2056,17 @@ function startCompaction(session: AcpThreadSession): void { session.turnSettled = undefined; }; - session.turnSettled = session.connection - .request({ - method: "session/prompt", - params: { - sessionId: session.providerThreadId, - prompt: [{ type: "text", text: "/compact" }], - }, - resultSchema: acpPromptResultSchema, - }) + const promptResult = session.connection.request({ + method: "session/prompt", + params: { + sessionId: session.providerThreadId, + prompt: [{ type: "text", text: "/compact" }], + }, + resultSchema: acpPromptResultSchema, + }); + acceptTurnInput(session, pending); + + session.turnSettled = promptResult .then((result) => { finish( result.stopReason === "end_turn" @@ -2320,21 +2410,22 @@ async function handleRequest( sendError(request.id, -32000, "A turn is already active"); return; } - // Accepted-input correlation (turn/input/accepted): the assembler owns - // the queue-until-turn-opens behavior, so the bridge only reports the - // acceptance. - sendThreadDeltas(session.bbThreadId, [ - { kind: "input.accepted", clientRequestId: params.clientRequestId }, - ]); + const pending: AcpPendingTurnInput = { + clientRequestId: params.clientRequestId, + input: params.input, + requestId: request.id, + }; + // Both paths answer the command and report the acceptance once the + // `session/prompt` request carrying the input has gone out. + // // A standalone builtin `/compact` mention is bb's manual-compaction // request, not model input: it runs the agent's own compaction command // instead of becoming a prompt. if (isStandaloneBuiltinCompactCommand(params.input)) { - startCompaction(session); + startCompaction(session, pending); } else { - runTurn(session, params.input); + runTurn(session, pending); } - sendResult(request.id, { threadId: params.threadId }); return; } @@ -2353,12 +2444,15 @@ async function handleRequest( ); return; } - // A steer joins the active turn: the assembler emits the acceptance - // into the turn it holds open. - sendThreadDeltas(session.bbThreadId, [ - { kind: "input.accepted", clientRequestId: params.clientRequestId }, - ]); - session.queuedInputs.push(params.input); + // A steer joins the active turn, but the agent only learns about it + // when the cancelled prompt is reissued with it. The command answers now + // — the bridge has the input — while the acceptance waits for that + // reissue, so a steer the turn drops is never reported as accepted. + session.queuedInputs.push({ + clientRequestId: params.clientRequestId, + input: params.input, + requestId: null, + }); requestSteerCancel(session); sendResult(request.id, { threadId: params.threadId }); return;