Fix Claude resumed-input turn correlation

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 —

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);
   });
 

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

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 {
-  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(

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");

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