From 73fb6542b9e90d328ed6e1e09fee29d629cdb2e8 Mon Sep 17 00:00:00 2001 From: Michael Yong Date: Wed, 19 Aug 2026 23:28:33 -0700 Subject: [PATCH] Fix Claude resumed-input turn correlation --- packages/host-daemon-contract/src/protocol.ts | 8 +- .../test/contract.test.ts | 2 +- .../src/bridge/__tests__/bridge.test.ts | 105 ++++++++++-------- .../provider-claude-code/src/bridge/bridge.ts | 43 +++---- .../src/delta-translation.test.ts | 70 ++++++++++++ .../src/delta-translation.ts | 15 ++- plugins/provider-claude-code/src/schemas.ts | 9 ++ 7 files changed, 173 insertions(+), 79 deletions(-) diff --git a/packages/host-daemon-contract/src/protocol.ts b/packages/host-daemon-contract/src/protocol.ts index 4bc39f147a..c221b0914a 100644 --- a/packages/host-daemon-contract/src/protocol.ts +++ b/packages/host-daemon-contract/src/protocol.ts @@ -1,3 +1,9 @@ +// Version 139 keeps a resumed Claude session's provider-owned task-notification +// result from claiming a newly accepted human input, and delays turn/start +// acceptance until Claude's SDK prompt iterator consumes the input. Older +// daemons can still make a sent message appear to complete immediately while +// its real response continues under a second, unaccepted turn. +// // Version 138 removes the `workspace.discover_repos` command. It existed only // for the first-run onboarding flow's project step, which is deleted; no server // sends it any more. A newer daemon no longer answers it, so an older server @@ -60,7 +66,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 = 138 as const; +export const HOST_DAEMON_PROTOCOL_VERSION = 139 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 dd30f92ac4..51ad6ca7dd 100644 --- a/packages/host-daemon-contract/test/contract.test.ts +++ b/packages/host-daemon-contract/test/contract.test.ts @@ -1134,7 +1134,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(138); + expect(HOST_DAEMON_PROTOCOL_VERSION).toBe(139); expect(HOST_ARTIFACT_MAX_BYTES).toBe(256 * 1024 * 1024); }); diff --git a/plugins/provider-claude-code/src/bridge/__tests__/bridge.test.ts b/plugins/provider-claude-code/src/bridge/__tests__/bridge.test.ts index 50fcab3e2e..a5c5a89dc2 100644 --- a/plugins/provider-claude-code/src/bridge/__tests__/bridge.test.ts +++ b/plugins/provider-claude-code/src/bridge/__tests__/bridge.test.ts @@ -2815,8 +2815,8 @@ describe("bridge", () => { }, }, }); - await bridge.waitForResponse(2); await readNextPrompt(call); + await bridge.waitForResponse(2); expect(queries).toHaveLength(1); expect(query.close).not.toHaveBeenCalled(); @@ -2891,8 +2891,8 @@ describe("bridge", () => { }, }, }); - await bridge.waitForResponse(3); await readNextPrompt(call); + await bridge.waitForResponse(3); expect(queries).toHaveLength(1); expect(query.applyFlagSettings).toHaveBeenLastCalledWith({ @@ -3004,8 +3004,8 @@ describe("bridge", () => { providerOptions: {}, }, }); - await bridge.waitForResponse(2); const deniedPrompt = await readNextPrompt(call); + await bridge.waitForResponse(2); if (!deniedPrompt.uuid) { throw new Error("Expected denied prompt UUID"); } @@ -3034,8 +3034,8 @@ describe("bridge", () => { providerOptions: {}, }, }); - await bridge.waitForResponse(3); const askPrompt = await readNextPrompt(call); + await bridge.waitForResponse(3); if (!askPrompt.uuid) { throw new Error("Expected ask prompt UUID"); } @@ -3088,8 +3088,8 @@ describe("bridge", () => { providerOptions: {}, }, }); - await bridge.waitForResponse(4); const latestPrompt = await readNextPrompt(call); + await bridge.waitForResponse(4); if (!latestPrompt.uuid) { throw new Error("Expected latest prompt UUID"); } @@ -3419,7 +3419,7 @@ describe("bridge", () => { providerOptions: {}, }, }); - await bridge.waitForResponse(2); + await bridge.flushWork(); expect(queries).toHaveLength(2); expect(getLatestQueryOptions()).toMatchObject({ @@ -3428,6 +3428,7 @@ describe("bridge", () => { await expect(readNextPromptText(getLatestQueryCall())).resolves.toBe( inputText, ); + await bridge.waitForResponse(2); bridge.sendRequest(3, "thread/stop", { threadId, @@ -3559,6 +3560,9 @@ describe("bridge", () => { providerOptions: {}, }, }); + await expect(readNextPromptText(getLatestQueryCall())).resolves.toBe( + inputText, + ); await bridge.waitForResponse(2); queries[0]?.emit( @@ -3720,49 +3724,57 @@ describe("bridge", () => { } }); - it("delays turn steer responses until the SDK prompt consumes the input", async () => { - const threadId = "thread-steer-consumed"; - const bridge = createBridgeJsonRpcTestHarness(handleLine); - const queries: ControlledClaudeQuery[] = []; - queryMock.mockImplementation(() => { - const query = createControlledClaudeQuery(); - queries.push(query); - return query; - }); + it.each([ + { method: "turn/start", name: "turn start" }, + { method: "turn/steer", name: "turn steer" }, + ] as const)( + "delays $name responses until the SDK prompt consumes the input", + async (testCase) => { + const threadId = `thread-${testCase.method.replace("/", "-")}-consumed`; + const bridge = createBridgeJsonRpcTestHarness(handleLine); + const queries: ControlledClaudeQuery[] = []; + queryMock.mockImplementation(() => { + const query = createControlledClaudeQuery(); + queries.push(query); + return query; + }); - try { - await startBridgeThread({ bridge, threadId }); + try { + await startBridgeThread({ bridge, threadId }); - bridge.sendRequest(2, "turn/steer", { - threadId, - providerThreadId: threadId, - expectedTurnId: "turn-1", - input: [{ type: "text", text: "Please account for the restart" }], - clientRequestId: "creq_abcdefghjk", - options: { - permissionMode: "accept-edits", - permissionScope: "workspace", - approvalReviewer: "user", - permissionEscalation: "ask", - providerOptions: {}, - }, - }); - await bridge.flushWork(); + bridge.sendRequest(2, testCase.method, { + threadId, + providerThreadId: threadId, + ...(testCase.method === "turn/steer" + ? { expectedTurnId: "turn-1" } + : {}), + input: [{ type: "text", text: "Please account for the restart" }], + clientRequestId: "creq_abcdefghjk", + options: { + permissionMode: "accept-edits", + permissionScope: "workspace", + approvalReviewer: "user", + permissionEscalation: "ask", + providerOptions: {}, + }, + }); + await bridge.flushWork(); - expect(bridge.hasResponse(2)).toBe(false); - await expect(readNextPromptText(getLatestQueryCall())).resolves.toBe( - "Please account for the restart", - ); - await expect(bridge.waitForResponse(2)).resolves.toMatchObject({ - result: { threadId }, - }); + expect(bridge.hasResponse(2)).toBe(false); + await expect(readNextPromptText(getLatestQueryCall())).resolves.toBe( + "Please account for the restart", + ); + await expect(bridge.waitForResponse(2)).resolves.toMatchObject({ + result: { threadId }, + }); - await stopBridgeThread({ bridge, queries, threadId }); - } finally { - queries[0]?.finish(); - bridge.restore(); - } - }); + await stopBridgeThread({ bridge, queries, threadId }); + } finally { + queries[0]?.finish(); + bridge.restore(); + } + }, + ); it.each([ { method: "turn/start", name: "turn start" }, @@ -3861,8 +3873,8 @@ describe("bridge", () => { providerOptions: {}, }, }); - await bridge.waitForResponse(2); const text = await readNextPromptText(getLatestQueryCall()); + await bridge.waitForResponse(2); await stopBridgeThread({ bridge, queries, threadId }); return text; } @@ -4130,6 +4142,7 @@ describe("canonical model context-window hint", () => { input: [{ type: "text", text: "hello", mentions: [] }], options: { ...canonicalOptions, model: "claude-opus-4-7[1m]" }, }); + await readNextPrompt(getLatestQueryCall()); await bridge.waitForResponse(2); // A result with token usage but no `modelUsage`: the only capacity diff --git a/plugins/provider-claude-code/src/bridge/bridge.ts b/plugins/provider-claude-code/src/bridge/bridge.ts index bdbc233e7f..ed894c4711 100644 --- a/plugins/provider-claude-code/src/bridge/bridge.ts +++ b/plugins/provider-claude-code/src/bridge/bridge.ts @@ -563,10 +563,6 @@ function logBridgeError(message: string): void { process.stderr.write(`claude-code bridge: ${message}\n`); } -function ignoreInputConsumption(promise: Promise): void { - void promise.catch(() => {}); -} - function pushPromptInput( threadSession: ThreadSession, input: string, @@ -583,22 +579,6 @@ function pushPromptInput( }); } -function queuePromptInputs( - threadSession: ThreadSession, - inputs: readonly string[], - permissionEscalation: PermissionEscalation | null, -): boolean { - if (!threadSession.session.canPushInput()) { - return false; - } - for (const input of inputs) { - ignoreInputConsumption( - pushPromptInput(threadSession, input, permissionEscalation), - ); - } - return true; -} - async function applyLiveSessionSettings( threadSession: ThreadSession, threadId: string, @@ -2222,15 +2202,22 @@ async function runTurnStart( return; } - if ( - !queuePromptInputs(threadSession, [promptText], params.permissionEscalation) - ) { - sendError(id, -32000, "Claude SDK input stream is closed"); - return; + try { + await pushPromptInput( + threadSession, + promptText, + params.permissionEscalation, + ); + // Like steer, a new turn is accepted only after the SDK prompt iterator + // consumes it. Queueing alone cannot prove which provider-owned segment a + // concurrently drained result belongs to. + emitCanonicalTurnInputAccepted(threadSession, acceptance, params.threadId); + threadSession.permissionEscalation = params.permissionEscalation; + sendResult(id, { threadId: params.threadId }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + sendError(id, -32000, message); } - emitCanonicalTurnInputAccepted(threadSession, acceptance, params.threadId); - threadSession.permissionEscalation = params.permissionEscalation; - sendResult(id, { threadId: params.threadId }); } async function handleTurnStart( diff --git a/plugins/provider-claude-code/src/delta-translation.test.ts b/plugins/provider-claude-code/src/delta-translation.test.ts index 18134a4ae9..22a67a7644 100644 --- a/plugins/provider-claude-code/src/delta-translation.test.ts +++ b/plugins/provider-claude-code/src/delta-translation.test.ts @@ -681,6 +681,76 @@ describe("claude synthetic no-response handling", () => { ); }); + it("does not let a recovered task notification settle pending human input", () => { + const harness = createClaudeDeltaHarness(); + harness.acceptInput("creq_23456789af", "bb-thread-1"); + + // On resume the Claude SDK can drain a provider-owned task notification + // immediately before the queued human prompt. Its zero-work result is a + // different root segment and must not claim the pending bb input. + expect( + harness.translate( + { + type: "result", + subtype: "success", + is_error: false, + num_turns: 0, + result: "", + origin: { kind: "task-notification" }, + session_id: "claude-session-1", + }, + { threadId: "bb-thread-1" }, + ), + ).toEqual([]); + + const assistantEvents = harness.translate( + { + type: "assistant", + message: { + id: "human-response", + role: "assistant", + content: [{ type: "text", text: "I am working on it." }], + }, + session_id: "claude-session-1", + }, + { threadId: "bb-thread-1" }, + ); + + expect(assistantEvents).toContainEqual( + expect.objectContaining({ + type: "turn/input/accepted", + scope: turnScope(TURN_1), + clientRequestId: "creq_23456789af", + }), + ); + expect(assistantEvents).toContainEqual( + expect.objectContaining({ + type: "item/completed", + scope: turnScope(TURN_1), + item: expect.objectContaining({ text: "I am working on it." }), + }), + ); + + expect( + harness.translate( + { + type: "result", + subtype: "success", + is_error: false, + origin: { kind: "human" }, + session_id: "claude-session-1", + }, + { threadId: "bb-thread-1" }, + ), + ).toContainEqual( + expect.objectContaining({ + type: "turn/completed", + scope: turnScope(TURN_1), + status: "completed", + }), + ); + }); + it("ignores a trailing result once the turn has closed", () => { const harness = createClaudeDeltaHarness(); harness.acceptInput("creq_23456789af", "bb-thread-1"); diff --git a/plugins/provider-claude-code/src/delta-translation.ts b/plugins/provider-claude-code/src/delta-translation.ts index b95bdfe127..be2cea41c6 100644 --- a/plugins/provider-claude-code/src/delta-translation.ts +++ b/plugins/provider-claude-code/src/delta-translation.ts @@ -1206,9 +1206,18 @@ export function createClaudeDeltaTranslator() { return unexpectedSdkEventDeltas(event, context); } const message = parsedMessage.data; - // The terminal-turn rule: the result owns the open turn, or claims one - // proven by pending accepted input; on an idle thread it emits nothing. - if (!state.mirror.turnOpen && state.mirror.pendingInputs === 0) { + // The terminal-turn rule: the result owns the open turn, or a human result + // claims one proven by pending accepted input. On resume, Claude can drain + // a recovered task notification immediately before the queued human + // prompt. Its result belongs to a provider-owned root segment and must not + // steal that prompt's pending input. The SDK defines absent origin as + // human, preserving local zero-work commands such as /clear. + const resultCanClaimPendingInput = + message.origin === undefined || message.origin.kind === "human"; + if ( + !state.mirror.turnOpen && + (state.mirror.pendingInputs === 0 || !resultCanClaimPendingInput) + ) { return []; } // Claiming through pending input opens the turn first (clearing the diff --git a/plugins/provider-claude-code/src/schemas.ts b/plugins/provider-claude-code/src/schemas.ts index e98b880788..3ec1e7b470 100644 --- a/plugins/provider-claude-code/src/schemas.ts +++ b/plugins/provider-claude-code/src/schemas.ts @@ -392,6 +392,14 @@ const claudeResultSubtypeSchema = z.enum([ ]); export type ClaudeResultSubtype = z.infer; +const claudeMessageOriginSchema = z + .object({ + // The SDK treats an absent origin as human and can add new non-human + // provenance kinds over time. The translator only needs that distinction. + kind: z.string().min(1), + }) + .passthrough(); + export const claudeResultMessageSchema = z .object({ type: z.literal("result"), @@ -402,6 +410,7 @@ export const claudeResultMessageSchema = z result: z.unknown().optional(), usage: z.unknown().optional(), modelUsage: z.unknown().optional(), + origin: claudeMessageOriginSchema.optional(), }) .passthrough(); export type ClaudeResultMessage = z.infer;