diff --git a/apps/server/test/internal/internal-event-append-ownership.test.ts b/apps/server/test/internal/internal-event-append-ownership.test.ts index 445881e58f..2c9d986fb5 100644 --- a/apps/server/test/internal/internal-event-append-ownership.test.ts +++ b/apps/server/test/internal/internal-event-append-ownership.test.ts @@ -1,5 +1,5 @@ import { eq } from "drizzle-orm"; -import { events, getThread } from "@bb/db"; +import { events, getThread, listQueuedThreadMessages } from "@bb/db"; import { threadScope, turnScope } from "@bb/domain"; import { groupHostDaemonEvents, @@ -18,6 +18,7 @@ import { seedEvent, seedHostSession, seedProjectWithSource, + seedQueuedMessage, seedThread, seedThreadRuntimeState, } from "../helpers/seed.js"; @@ -405,6 +406,75 @@ describe("internal event append ownership", () => { } }); + it("auto-sends a queued message after a zero-work turn completes", async () => { + const { environment, harness, session, thread } = await setupEventRoute(); + try { + seedThreadRuntimeState(harness.deps, { + environmentId: environment.id, + inputText: "/clear", + providerThreadId: "provider-zero-work", + threadId: thread.id, + }); + seedQueuedMessage(harness.deps, { + content: [ + { + type: "text", + text: "queued behind zero-work turn", + mentions: [], + }, + ], + threadId: thread.id, + }); + + const response = await postEventBatch({ + harness, + sessionId: session.id, + events: [ + { + threadId: thread.id, + event: { + type: "turn/started", + threadId: thread.id, + providerThreadId: "provider-zero-work", + scope: turnScope("turn-zero-work"), + }, + }, + { + threadId: thread.id, + event: { + type: "turn/completed", + threadId: thread.id, + providerThreadId: "provider-zero-work", + scope: turnScope("turn-zero-work"), + status: "completed", + }, + }, + ], + }); + + expect(response.status).toBe(200); + const queuedTurn = await waitForQueuedCommand( + harness, + ({ command }) => + command.type === "turn.submit" && command.threadId === thread.id, + ); + expect(queuedTurn.command).toMatchObject({ + type: "turn.submit", + input: [ + { + type: "text", + text: "queued behind zero-work turn", + mentions: [], + }, + ], + }); + expect(listQueuedThreadMessages(harness.db, thread.id)).toEqual([]); + expect(getThread(harness.db, thread.id)?.status).toBe("active"); + } finally { + await harness.cleanup(); + } + }); + it("rejects unowned thread events without blocking owned events in the same batch", async () => { const { harness, session, thread } = await setupEventRoute(); try { diff --git a/packages/agent-runtime/src/acp/adapter.test.ts b/packages/agent-runtime/src/acp/adapter.test.ts index 043bd92d2c..61d72f38dc 100644 --- a/packages/agent-runtime/src/acp/adapter.test.ts +++ b/packages/agent-runtime/src/acp/adapter.test.ts @@ -774,6 +774,40 @@ describe("acp adapter event translation", () => { ]); }); + it("settles accepted input when completion arrives before an update", () => { + const adapter = createAdapter(); + adapter.translateAcceptedCommand({ + command: { + type: "turn/start", + clientRequestId: "creq_222222228e", + input: [promptTextInput({ text: "/agent-local-command" })], + options: fullProviderExecutionContext, + providerThreadId: "sess-1", + threadId: "thread-1", + }, + }); + + const events = adapter.translateEvent( + { + jsonrpc: "2.0", + method: "acp/turn/completed", + params: { threadId: "thread-1", stopReason: "end_turn" }, + }, + THREAD_CONTEXT, + ); + + expect(events.map((event) => event.type)).toEqual([ + "turn/started", + "turn/input/accepted", + "turn/completed", + ]); + expect(events.at(-1)).toMatchObject({ + type: "turn/completed", + scope: turnScope("turn-1"), + status: "completed", + }); + }); + it("translates ACP usage updates into exact context-window usage", () => { const adapter = createAdapter(); startTurn(adapter); diff --git a/packages/agent-runtime/src/acp/adapter.ts b/packages/agent-runtime/src/acp/adapter.ts index 13802cc0bc..981cfe4b65 100644 --- a/packages/agent-runtime/src/acp/adapter.ts +++ b/packages/agent-runtime/src/acp/adapter.ts @@ -63,6 +63,7 @@ import { createStandardAdapterMembers } from "../shared/standard-adapter-members import { completeStartedToolItem } from "../shared/tool-item-translation.js"; import { buildUnhandledProviderEvents } from "../shared/provider-unhandled-event.js"; import { createScopedItemIdFactory } from "../shared/scoped-item-ids.js"; +import { resolveProviderTerminalTurn } from "../shared/provider-terminal-turn.js"; import { createProviderTurnStateRegistry, finishOpenProviderTurn, @@ -987,11 +988,16 @@ export function createAcpProviderAdapter( state: AcpTurnState, context?: ProviderTranslationContext, ): ThreadEvent[] { - const currentTurnId = state.currentTurnId; - if (!currentTurnId) { + const events: ThreadEvent[] = []; + const currentTurnId = resolveProviderTerminalTurn({ + events, + registry: turnState, + state, + threadId: UNSTAMPED_THREAD_ID, + }); + if (currentTurnId === undefined) { return []; } - const events: ThreadEvent[] = []; const openToolCallStatus: ThreadEventItemStatus = stopReason === "end_turn" ? "completed" diff --git a/packages/agent-runtime/src/claude-code/adapter.test.ts b/packages/agent-runtime/src/claude-code/adapter.test.ts index b04bfea81f..9dd3435f1f 100644 --- a/packages/agent-runtime/src/claude-code/adapter.test.ts +++ b/packages/agent-runtime/src/claude-code/adapter.test.ts @@ -2186,6 +2186,99 @@ describe("claude-code provider adapter", () => { ); }); + it("translateEvent settles a zero-work run that never started a turn", () => { + const adapter = createClaudeCodeProviderAdapter(); + + expect( + adapter.translateAcceptedCommand({ + command: { + type: "turn/start", + threadId: "bb-thread-1", + providerThreadId: "claude-session-1", + clientRequestId: "creq_23456789af", + input: [promptTextInput({ text: "/clear" })], + options: fullProviderExecutionContext, + }, + }), + ).toEqual([]); + + // The CLI resolves /clear locally: it emits conversation_reset and then a + // success result, with no model call and so no assistant message to start + // the turn. The result has to settle it, or the thread stays active. + const events = adapter.translateEvent( + { + type: "result", + subtype: "success", + is_error: false, + num_turns: 0, + result: "", + session_id: "claude-session-1", + }, + { threadId: "bb-thread-1" }, + ); + + expect(events.map((event) => event.type)).toEqual([ + "turn/started", + "turn/input/accepted", + "turn/completed", + ]); + expect(events).toContainEqual( + expect.objectContaining({ + type: "turn/completed", + scope: turnScope("turn-1"), + status: "completed", + }), + ); + }); + + it("translateEvent ignores a trailing result once the turn has closed", () => { + const adapter = createClaudeCodeProviderAdapter(); + + adapter.translateAcceptedCommand({ + command: { + type: "turn/start", + threadId: "bb-thread-1", + providerThreadId: "claude-session-1", + clientRequestId: "creq_23456789af", + input: [promptTextInput({ text: "please do this" })], + options: fullProviderExecutionContext, + }, + }); + adapter.translateEvent( + { + type: "assistant", + message: { + id: "msg-1", + role: "assistant", + content: [{ type: "text", text: "Hello world" }], + }, + session_id: "claude-session-1", + }, + { threadId: "bb-thread-1" }, + ); + expect( + adapter.buildCommandPlan({ + type: "thread/stop", + threadId: "bb-thread-1", + providerThreadId: "claude-session-1", + activeTurnId: "turn-1", + }), + ).toEqual({ + kind: "request", + method: "thread/stop", + params: { threadId: "bb-thread-1" }, + }); + + // A stop finishes the open turn before the CLI's result lands, so a result + // with no open turn is routine. It must not open a second, empty turn. + expect( + adapter.translateEvent( + { type: "result", subtype: "success", session_id: "claude-session-1" }, + { threadId: "bb-thread-1" }, + ), + ).toEqual([]); + }); + it("translateEvent completes a pending turn for wrapped Claude synthetic no-response messages", () => { const adapter = createClaudeCodeProviderAdapter(); diff --git a/packages/agent-runtime/src/claude-code/translate-message.ts b/packages/agent-runtime/src/claude-code/translate-message.ts index dced837c3c..83bcd976b0 100644 --- a/packages/agent-runtime/src/claude-code/translate-message.ts +++ b/packages/agent-runtime/src/claude-code/translate-message.ts @@ -17,6 +17,7 @@ import type { ProviderTurnStateRegistry, } from "../shared/turn-state.js"; import { createScopedItemIdFactory } from "../shared/scoped-item-ids.js"; +import { resolveProviderTerminalTurn } from "../shared/provider-terminal-turn.js"; import { UNSTAMPED_THREAD_ID } from "../shared/unstamped-thread-id.js"; import type { ProviderTranslationContext } from "../provider-adapter.js"; import { @@ -908,7 +909,13 @@ export function translateClaudeSdkMessage( }); } const message = parsedMessage.data; - if (state.currentTurnId) { + const turnId = resolveProviderTerminalTurn({ + events, + registry: args.turnState, + state, + threadId, + }); + if (turnId) { const contextWindowUsage = extractClaudeContextWindowUsage({ fallbackModelContextWindow: state.selectedModelContextWindow, latestRequestContextTokens: state.latestRequestContextTokens, @@ -927,7 +934,7 @@ export function translateClaudeSdkMessage( type: "thread/contextWindowUsage/updated", threadId, providerThreadId: "", - scope: turnScope(state.currentTurnId), + scope: turnScope(turnId), contextWindowUsage, }); } @@ -936,7 +943,7 @@ export function translateClaudeSdkMessage( type: "thread/tokenUsage/updated", threadId, providerThreadId: "", - scope: turnScope(state.currentTurnId), + scope: turnScope(turnId), tokenUsage, }); } @@ -968,7 +975,7 @@ export function translateClaudeSdkMessage( httpStatusCode: resultErrorInfo?.httpStatusCode ?? null, }, threadId, - turnId: state.currentTurnId, + turnId, }), ); } @@ -986,7 +993,7 @@ export function translateClaudeSdkMessage( type: "turn/completed", threadId, providerThreadId: "", - scope: turnScope(state.currentTurnId), + scope: turnScope(turnId), status: failed ? "failed" : "completed", ...(state.latestProviderCheckpointId !== undefined ? { diff --git a/packages/agent-runtime/src/pi/adapter.test.ts b/packages/agent-runtime/src/pi/adapter.test.ts index 8cbcafdbba..45899b6a7a 100644 --- a/packages/agent-runtime/src/pi/adapter.test.ts +++ b/packages/agent-runtime/src/pi/adapter.test.ts @@ -733,6 +733,40 @@ describe("pi provider adapter", () => { ); }); + it("settles accepted input when the prompt resolves before agent_start", () => { + const adapter = createPiProviderAdapter(); + adapter.translateAcceptedCommand({ + command: { + type: "turn/start", + clientRequestId: "creq_222222228e", + input: [promptTextInput({ text: "/local-extension-command" })], + options: fullProviderExecutionContext, + providerThreadId: "pi-session-1", + threadId: "bb-t1", + }, + }); + + const events = adapter.translateEvent( + { + jsonrpc: "2.0", + method: "pi/prompt/settled", + params: { threadId: "bb-t1", status: "completed" }, + }, + { threadId: "bb-t1" }, + ); + + expect(events.map((event) => event.type)).toEqual([ + "turn/started", + "turn/input/accepted", + "turn/completed", + ]); + expect(events.at(-1)).toMatchObject({ + type: "turn/completed", + scope: turnScope("turn-1"), + status: "completed", + }); + }); + it("translateEvent keeps turn_start as internal noise while agent_start owns the bb turn", () => { const adapter = createPiProviderAdapter(); adapter.translateEvent(loadFixture("agent-start.json")); diff --git a/packages/agent-runtime/src/pi/adapter.ts b/packages/agent-runtime/src/pi/adapter.ts index 2e9e151bd5..82ae6f7d3e 100644 --- a/packages/agent-runtime/src/pi/adapter.ts +++ b/packages/agent-runtime/src/pi/adapter.ts @@ -53,6 +53,7 @@ import { finishOpenProviderTurn, } from "../shared/turn-state.js"; import { createScopedItemIdFactory } from "../shared/scoped-item-ids.js"; +import { resolveProviderTerminalTurn } from "../shared/provider-terminal-turn.js"; import { buildUnhandledProviderEvents, createUnhandledProviderEvent, @@ -203,6 +204,16 @@ const piEventTypeSchema = z }) .passthrough(); +const piPromptSettledEnvelopeSchema = z.object({ + jsonrpc: z.literal("2.0"), + method: z.literal("pi/prompt/settled"), + params: z.object({ + threadId: z.string().min(1), + status: z.enum(["completed", "failed"]), + error: z.string().optional(), + }), +}); + // Pi events we deliberately drop rather than translate. Without this the // fallback treats them as unknown and emits a `provider/unhandled` event, which // renders as "Unhandled Pi event" in the transcript. @@ -661,6 +672,36 @@ export function createPiProviderAdapter( }); } + const promptSettledEnvelope = + piPromptSettledEnvelopeSchema.safeParse(event); + if (promptSettledEnvelope.success) { + const stateThreadId = + context?.threadId ?? promptSettledEnvelope.data.params.threadId; + const state = turnState.getOrCreate({ threadId: stateThreadId }); + const events: ThreadEvent[] = []; + const turnId = resolveProviderTerminalTurn({ + events, + registry: turnState, + state, + threadId: UNSTAMPED_THREAD_ID, + }); + if (turnId === undefined) { + return events; + } + events.push({ + type: "turn/completed", + threadId: UNSTAMPED_THREAD_ID, + providerThreadId: "", + scope: turnScope(turnId), + status: promptSettledEnvelope.data.params.status, + ...(promptSettledEnvelope.data.params.error !== undefined + ? { error: { message: promptSettledEnvelope.data.params.error } } + : {}), + }); + turnState.finishTurn({ state, threadId: stateThreadId }); + return events; + } + const identityEnvelope = threadIdentityEnvelopeSchema.safeParse(event); if (identityEnvelope.success) { const { threadId = UNSTAMPED_THREAD_ID, providerThreadId } = @@ -829,32 +870,39 @@ export function createPiProviderAdapter( if (!piEvent.success) { return buildUnexpectedEvent(event); } - const currentTurnId = state.currentTurnId; - if (!currentTurnId) { - break; + const currentTurnId = resolveProviderTerminalTurn({ + events, + registry: turnState, + state, + threadId, + }); + if (currentTurnId === undefined) { + return events; } const lastAssistant = findLastAssistantMessage(piEvent.data.messages); if (piEvent.data.willRetry) { - return lastAssistant && isPiAssistantError(lastAssistant) - ? [ - { - type: "provider/error", - threadId, - providerThreadId: "", - scope: turnScope(currentTurnId), - message: "Provider error", - detail: lastAssistant.errorMessage, - willRetry: true, - }, - ] - : []; + if (lastAssistant && isPiAssistantError(lastAssistant)) { + events.push({ + type: "provider/error", + threadId, + providerThreadId: "", + scope: turnScope(currentTurnId), + message: "Provider error", + detail: lastAssistant.errorMessage, + willRetry: true, + }); + } + return events; } if (lastAssistant && isPiAssistantError(lastAssistant)) { resetPiCommandOutputSnapshots(state); - return turnState.buildErrorEvents({ - contextThreadId: context?.threadId, - detail: lastAssistant.errorMessage, - }); + return [ + ...events, + ...turnState.buildErrorEvents({ + contextThreadId: context?.threadId, + detail: lastAssistant.errorMessage, + }), + ]; } if (lastAssistant) { const text = extractAssistantText(lastAssistant); 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 55dde4f78f..73bc8ef0ce 100644 --- a/packages/agent-runtime/src/pi/bridge/__tests__/bridge.test.ts +++ b/packages/agent-runtime/src/pi/bridge/__tests__/bridge.test.ts @@ -888,6 +888,38 @@ describe("pi bridge", () => { } }); + it("reports when a turn/start prompt settles without SDK turn events", async () => { + const bridge = createBridgeJsonRpcTestHarness(handleLine); + const piSession = createControlledPiAgentSession(); + mockCreateAgentSession.mockResolvedValue({ session: piSession }); + + try { + bridge.sendRequest(50, "thread/start", { + cwd: "/tmp/worktree", + threadId: "thread-zero-work", + }); + await bridge.waitForResponse(50); + + bridge.sendRequest(51, "turn/start", { + threadId: "thread-zero-work", + input: [{ type: "text", text: "/local-extension-command" }], + }); + await bridge.waitForResponse(51); + await bridge.flushWork(); + + expect(bridge.messages).toContainEqual({ + jsonrpc: "2.0", + method: "pi/prompt/settled", + params: { + threadId: "thread-zero-work", + status: "completed", + }, + }); + } 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/bridge.ts b/packages/agent-runtime/src/pi/bridge/bridge.ts index 15044c53cd..75d15ed5dc 100644 --- a/packages/agent-runtime/src/pi/bridge/bridge.ts +++ b/packages/agent-runtime/src/pi/bridge/bridge.ts @@ -446,6 +446,31 @@ function createOnSessionDone( }; } +function reportPromptSettled(args: { + error?: unknown; + sessionSerial: number; + threadId: string; +}): void { + if (!getCurrentThreadSession(args)) { + return; + } + const errorMessage = + args.error === undefined + ? undefined + : args.error instanceof Error + ? args.error.message + : String(args.error); + send({ + jsonrpc: "2.0", + method: "pi/prompt/settled", + params: { + threadId: args.threadId, + status: errorMessage === undefined ? "completed" : "failed", + ...(errorMessage !== undefined ? { error: errorMessage } : {}), + }, + }); +} + function reportSessionError( args: CreateSessionCallbackArgs & { error: unknown }, ): void { @@ -768,10 +793,21 @@ async function handleTurnStart( return; } - void threadSession.session.prompt( - text, - images.length > 0 ? images : undefined, - ); + void threadSession.session + .prompt(text, images.length > 0 ? images : undefined) + .then( + () => + reportPromptSettled({ + sessionSerial: threadSession.sessionSerial, + threadId: params.threadId, + }), + (error: unknown) => + reportPromptSettled({ + error, + sessionSerial: threadSession.sessionSerial, + threadId: params.threadId, + }), + ); sendResult(id, { threadId: params.threadId }); } diff --git a/packages/agent-runtime/src/shared/provider-terminal-turn.ts b/packages/agent-runtime/src/shared/provider-terminal-turn.ts new file mode 100644 index 0000000000..97482eb31c --- /dev/null +++ b/packages/agent-runtime/src/shared/provider-terminal-turn.ts @@ -0,0 +1,32 @@ +import type { ThreadEvent } from "@bb/domain"; +import type { AcceptedUserMessageState } from "./accepted-user-messages.js"; +import type { + ProviderTurnState, + ProviderTurnStateRegistry, +} from "./turn-state.js"; + +interface ResolveProviderTerminalTurnArgs< + TState extends ProviderTurnState & AcceptedUserMessageState, +> { + events: ThreadEvent[]; + registry: ProviderTurnStateRegistry; + state: TState; + threadId: string; +} + +/** + * Resolve the turn owned by a provider terminal signal. Accepted input can + * finish before the provider emits an ordinary event that starts the turn. + * The pending-input queue proves that the terminal signal still owns work; + * once a turn starts or a session closes, that queue is drained instead. + */ +export function resolveProviderTerminalTurn< + TState extends ProviderTurnState & AcceptedUserMessageState, +>(args: ResolveProviderTerminalTurnArgs): string | undefined { + return ( + args.state.currentTurnId ?? + (args.state.pendingAcceptedUserMessages.length > 0 + ? args.registry.ensureTurnStarted(args) + : undefined) + ); +} diff --git a/packages/host-daemon-contract/src/commands.ts b/packages/host-daemon-contract/src/commands.ts index ef16deb04e..edda41bb93 100644 --- a/packages/host-daemon-contract/src/commands.ts +++ b/packages/host-daemon-contract/src/commands.ts @@ -36,7 +36,7 @@ import { providerCliStatusResponseSchema, } from "./local.js"; -export const HOST_DAEMON_PROTOCOL_VERSION = 114 as const; +export const HOST_DAEMON_PROTOCOL_VERSION = 115 as const; export { BRANCH_LIST_LIMIT_MAX, diff --git a/packages/host-daemon-contract/test/contract.test.ts b/packages/host-daemon-contract/test/contract.test.ts index 504a91c79b..c204005104 100644 --- a/packages/host-daemon-contract/test/contract.test.ts +++ b/packages/host-daemon-contract/test/contract.test.ts @@ -1056,13 +1056,16 @@ describe("host-daemon local schemas", () => { }); describe("host-daemon command schemas", () => { + // Version 115 settles zero-work provider prompts with a complete synthetic + // turn lifecycle. Older daemons can leave locally handled prompts active + // indefinitely, so enrolled machines must update for reliable completion. // Version 114 lets the daemon report `none` in Pi model reasoning efforts. // A version 113 server accepts that value on the wire but rejects it later // against its Pi provider ladder, so enrolled machines must not run that // mixed version. Version 113 carried the Devin Desktop open target rename // and remains part of the protocol lineage. - it("uses protocol version 114 for Pi thinking-off model capabilities", () => { - expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(114); + it("uses protocol version 115 for zero-work provider turn completion", () => { + expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(115); }); it("binds Plan cancellation to a required turn id and typed result", () => {