fix(claude-code): interrupt a written steer's native turn when its Turn is cancelled

docs/harnesses/claude-code/claude-code-steer.md

@@ -24,7 +24,9 @@ Host 在原生接受后自己发当前 Turn 的 `userMessage` Item。Adapter 不
 
 CLI 先发 `command_lifecycle` `queued`，表示 Claude 已接受这条输入；根级工具执行中插入时，工具调用结束后发 `started`，插入进入正在运行的 Turn；续态窗口插入时，`started` 在当前续态输出结束、插入作为下一原生 turn 运行之时发出。Adapter 在 `queued` 时返回 `{ accepted: true }`：此后由 Claude 负责送达，不等 `started`，否则调用方要被整段工具调用挂住（Desktop 提交约 30 秒超时）。输入一旦写入就无法撤回，所以本轮的 `result` 先于 `queued` 到达时不拒绝，仍按接受处理：此时 Turn 不结束，保持 held，插入作为该 Turn 的下一段运行，不再落入另一轮，期间 `turn.start` 返回 `sessionBusy`；`queued` 回执迟迟不到时，等待 5 秒后同样按接受处理，避免串行处理请求的调用方（如 Harness Broker）被无限期占住。只有目标不是当前 Turn、CLI 不支持、当前是自主 Turn、取消已在进行、两个可插入窗口（根级工具执行中与 held Turn 续态）都不成立，或输入写入失败时，才返回失败。写入后会话关闭或出错时抛出回执丢失异常，不返回表示未收下的拒绝结果，Host 不自动重发；Session 故障仍经原有故障路径报告。空文本返回 `invalidRequest`。忙时 `turn.start` 仍是 `sessionBusy`。
 
-插入被接受后、还没并入之前，如果用户停止了这一轮，处理取决于 Turn 是否 held。根级工具执行中、非 held 的活跃 Turn 仍按 SDK 的中断语义，仍在排队的插入会在下一轮执行：公开的 SDK 接口目前没有撤回单条排队消息的方法，Adapter 不处理这种情况，这个窗口只剩当前那次工具调用的剩余时间。held Turn（包括因存在待接管的插入而保持 held 的）则不立即结束，而是走停止后台工作并等待安静的取消流程：插入随后被接管时，它运行所在的原生段会被中断；以终态结束却未被接管时，取消随即收口；迟迟没有结果时沿用既有的取消升级，超时后关闭原生进程。续态窗口插入、尚未被接管时被停止就属于这种情况，窗口只剩插入写入到出现结果之间的短暂时间，且取消已开始后不再接受新的插入。
+用户停止 Turn 时，只要还有已写入、尚未被原生接管的插入，无论 Turn 是否 held，Adapter 都走停止后台工作并等待安静的取消流程。SDK 中断不保证丢弃排队输入，因此当前 Root 的 `result` 不会提前结束这个 Turn：插入随后发出 `started` 时，它运行所在的原生段也会被中断；以终态结束却未被接管时，取消进入安静期；迟迟没有结果时沿用既有的取消升级，超时后关闭原生进程。取消期间 `turn.start` 返回 `sessionBusy`，且不再接受新的插入；所有待接管插入已结算、后台任务已确认停止、Root 段已结束，并经过 continuation 安静期（默认 2 秒）后，Turn 才结束为 cancelled。没有待接管插入、也没有后台或续态工作的普通停止路径不变。
+
+插入的 `command_lifecycle started` 可以是 Root result 后的首个 idle 帧。Transport 在发出中断前确保该段的 accumulator 已创建并记录取消，使后续 `aborted_tools` / `aborted_streaming` 原生终态按取消处理；其他失败或无法证明中断的终态仍按失败处理。
 
 ## 版本
 
@@ -36,6 +38,8 @@ CLI 先发 `command_lifecycle` `queued`，表示 Claude 已接受这条输入；
 
 ## 验证
 
+自动化回归：`packages/adapters/claude-code/test/claude-code-adapter.test.ts` 的 `preserves cancellation when a native steer starts before its idle accumulator exists` 使用真实 `ClaudeSdkTransport` 与 `FakeQuery`，按 Root 中断结果 → 插入 `started`（首个 idle 帧）→ 插入中断结果的顺序送入 SDK 消息，断言最终为 cancelled、等待期间拒绝新 Turn、结束后可启动新 Turn，且没有自主 Turn。这是合成 SDK 流测试，不代表真实 CLI 或 Desktop 实机验收。
+
 实机：模型 `haiku`，权限 `bypassPermissions`，一轮里调用 Bash 执行 `sleep 4; echo tool-done`，工具开始后以 `priority: "next"` 插入一句。流里只有一次 `result`（`subtype: "success"`，`terminal_reason: "completed"`）。`started` 出现在工具结果之后、下一次模型请求之前，模型回复提到了插入的句子。因此实时结果处理不改：纯文本 user 回显本来就被忽略，第一个 `result` 仍是本轮结束。
 
 2026-10-21 复核：Linux aarch64，Claude Code 2.1.305，Adapter 自身的 `ClaudeSdkTransport` 驱动真实 CLI。一轮要求依次执行三次 `sleep 6`，第一个工具开始约 1.5 秒后调用 `steer()`。`steer()` 在 `queued` 时立即 resolve；本轮只执行了一次 `sleep`，`runTurn` 以 `succeeded` 结束，transcript 中模型回复的是插入要求的文本。

packages/adapters/claude-code/src/claude-code-adapter.ts

@@ -1602,10 +1602,10 @@ class ClaudeHarnessSession implements HarnessSession {
     if (
       transport &&
       (taskIds.length > 0 ||
-        (active.held &&
-          (active.rootSegmentActive ||
-            this.#occupancy.awaitingContinuation ||
-            active.steersAwaitingNative.size > 0)))
+        // A written steer survives the interrupt and runs as Claude's next native turn, so the
+        // Turn stays open to interrupt that turn too instead of leaking it into the next one.
+        active.steersAwaitingNative.size > 0 ||
+        (active.held && (active.rootSegmentActive || this.#occupancy.awaitingContinuation)))
     ) {
       await this.#cancelBackgroundWork(active, transport, taskIds);
       return { ok: true, value: { cancellationRequested: true } };

packages/adapters/claude-code/src/sdk-transport.ts

@@ -999,7 +999,11 @@ export class ClaudeSdkTransport implements ClaudeTurnTransport {
     if (this.#active) return this.abort();
     const activeQuery = this.#query;
     if (!activeQuery) throw new Error("Claude SDK transport is not running");
-    this.#idleAccumulator?.requestCancel();
+    // A steer takeover can interrupt on its first idle frame, before consume creates the accumulator.
+    const idle = (this.#idleAccumulator ??= new ClaudeNativeTurnAccumulator(
+      this.#provider ? { provider: this.#provider } : {},
+    ));
+    idle.requestCancel();
     try {
       await this.#stopRequest(activeQuery.interrupt(), INTERRUPT_TIMEOUT_MESSAGE);
     } catch (error) {

packages/adapters/claude-code/test/claude-code-adapter.test.ts

@@ -4,7 +4,7 @@ import { mkdir, mkdtemp, open, rm, writeFile } from "node:fs/promises";
 import { tmpdir } from "node:os";
 
 import { afterEach, describe, expect, it, vi } from "vitest";
-import type { Query, SDKMessage } from "@anthropic-ai/claude-agent-sdk";
+import type { Query, SDKMessage, SDKUserMessage } from "@anthropic-ai/claude-agent-sdk";
 import {
   harnessPermissionModeIdSchema,
   harnessThinkingOptionIdSchema,
@@ -1352,6 +1352,208 @@ describe("Claude Code HarnessAdapter", () => {
     await adapter.close();
   });
 
+  it("keeps a cancelled Turn open to interrupt the native turn of its written steer", async () => {
+    const { adapter, transports } = fixture({ continuationQuiescenceMs: 10 });
+    const session = await openSession(adapter);
+    const outputs: HarnessOutput[] = [];
+    const consuming = (async () => {
+      for await (const output of session.outputs) outputs.push(output);
+    })();
+    const completed = (turnId: string) =>
+      outputs.some(
+        (output) =>
+          output.kind === "event" &&
+          output.event.type === "turn.completed" &&
+          output.event.turnId === turnId,
+      );
+    await session.execute(textTurn("turn-1"));
+    const transport = transports[0];
+    if (!transport) throw new Error("Fake Claude transport was not created");
+    const steered = session.execute({
+      type: "turn.steer",
+      turnId: hostTurnIdSchema.parse("turn-1"),
+      input: [{ type: "text", text: "steer then stop" }],
+    });
+    await vi.waitFor(() => expect(transport.steers).toHaveLength(1));
+    const messageId = transport.steers[0]?.messageId ?? "";
+    transport.recordSteer(messageId);
+    await expect(steered).resolves.toEqual({ ok: true, value: { accepted: true } });
+
+    await expect(
+      session.execute({ type: "turn.cancel", turnId: hostTurnIdSchema.parse("turn-1") }),
+    ).resolves.toEqual({ ok: true, value: { cancellationRequested: true } });
+    // Claude Code 2.1.316 ends the interrupted Root with its result, then still runs the
+    // written steer as its next native turn: the interrupt does not drop queued input.
+    transport.finish({ status: "cancelled", reason: "aborted_tools" });
+    await Promise.resolve();
+    expect(completed("turn-1")).toBe(false);
+    // A new Turn must not take over the stream that still owes the steer's native turn.
+    await expect(session.execute(textTurn("turn-2"))).resolves.toMatchObject({
+      ok: false,
+      error: { code: "sessionBusy" },
+    });
+
+    transport.startSteer(messageId);
+    await vi.waitFor(() => expect(transport.abortContinuation).toHaveBeenCalledTimes(2));
+    transport.event({ type: "segment.started" });
+    transport.delta("Stale steer answer.", "stale-steer-answer");
+    transport.finish({ status: "cancelled", reason: "aborted_streaming" });
+    await vi.waitFor(() => expect(completed("turn-1")).toBe(true));
+    expect(
+      outputs.find((output) => output.kind === "event" && output.event.type === "turn.completed"),
+    ).toMatchObject({ event: { turnId: "turn-1", outcome: { status: "cancelled" } } });
+
+    await expect(session.execute(textTurn("turn-2"))).resolves.toEqual({
+      ok: true,
+      value: { turnId: "turn-2" },
+    });
+    transport.delta("Fresh answer.", "fresh-answer");
+    transport.finish({ status: "succeeded" });
+    await vi.waitFor(() => expect(completed("turn-2")).toBe(true));
+    const turnTwoText = outputs
+      .flatMap((output) =>
+        output.kind === "event" &&
+        output.event.type === "item.completed" &&
+        output.event.turnId === "turn-2" &&
+        output.event.snapshot.item.type === "agentMessage"
+          ? [output.event.snapshot.item.text]
+          : [],
+      )
+      .join("");
+    expect(turnTwoText).toBe("Fresh answer.");
+    expect(
+      outputs.some(
+        (output) => output.kind === "event" && output.event.type === "turn.autonomous.started",
+      ),
+    ).toBe(false);
+    await session.close();
+    await consuming;
+    await adapter.close();
+  });
+
+  it("preserves cancellation when a native steer starts before its idle accumulator exists", async () => {
+    const { adapter, dependencies } = fixture({ continuationQuiescenceMs: 10 });
+    const fakeQuery = new FakeQuery();
+    let transport: ClaudeSdkTransport | undefined;
+    let writtenInput: AsyncIterator<SDKUserMessage> | undefined;
+    dependencies.createTransport = (input) => {
+      transport = new ClaudeSdkTransport({
+        ...input,
+        command: process.execPath,
+        closeTimeoutMs: 50,
+        queryFactory: ({ prompt }) => {
+          if (typeof prompt === "string") throw new Error("Expected streaming input");
+          writtenInput = prompt[Symbol.asyncIterator]();
+          return fakeQuery as unknown as Query;
+        },
+      });
+      return transport;
+    };
+    const session = await openSession(adapter);
+    const outputs: HarnessOutput[] = [];
+    const consuming = (async () => {
+      for await (const output of session.outputs) outputs.push(output);
+    })();
+    const completed = (turnId: string) =>
+      outputs.find(
+        (output) =>
+          output.kind === "event" &&
+          output.event.type === "turn.completed" &&
+          output.event.turnId === turnId,
+      );
+    try {
+      await session.execute(textTurn("turn-1"));
+      if (!writtenInput) throw new Error("SDK streaming input was not created");
+      await writtenInput.next();
+      fakeQuery.push({
+        type: "system",
+        subtype: "init",
+        capabilities: ["msg_lifecycle_v1"],
+      } as unknown as SDKMessage);
+      fakeQuery.push({
+        type: "assistant",
+        parent_tool_use_id: null,
+        message: {
+          role: "assistant",
+          content: [
+            { type: "tool_use", id: "toolu_root", name: "Bash", input: { command: "sleep 1" } },
+          ],
+        },
+      } as unknown as SDKMessage);
+      await vi.waitFor(() => expect(transport?.canSteerActiveTurn()).toBe(true));
+      const steered = session.execute({
+        type: "turn.steer",
+        turnId: hostTurnIdSchema.parse("turn-1"),
+        input: [{ type: "text", text: "steer then stop" }],
+      });
+      const steer = await writtenInput.next();
+      const messageId = steer.value?.uuid;
+      expect(messageId).toBeTruthy();
+      fakeQuery.push({
+        type: "command_lifecycle",
+        command_uuid: messageId,
+        state: "queued",
+      } as unknown as SDKMessage);
+      await expect(steered).resolves.toEqual({ ok: true, value: { accepted: true } });
+      await session.execute({ type: "turn.cancel", turnId: hostTurnIdSchema.parse("turn-1") });
+      expect(fakeQuery.interrupt).toHaveBeenCalledOnce();
+      fakeQuery.push({
+        type: "result",
+        subtype: "error_during_execution",
+        is_error: true,
+        terminal_reason: "aborted_tools",
+      } as unknown as SDKMessage);
+      await vi.waitFor(() => expect(transport?.canSteerActiveTurn()).toBe(true));
+      expect(completed("turn-1")).toBeUndefined();
+      await expect(session.execute(textTurn("turn-2"))).resolves.toMatchObject({
+        ok: false,
+        error: { code: "sessionBusy" },
+      });
+
+      // This receipt is the first idle frame: its takeover callback interrupts before
+      // the consume loop would normally create the continuation's accumulator.
+      fakeQuery.push({
+        type: "command_lifecycle",
+        command_uuid: messageId,
+        state: "started",
+      } as unknown as SDKMessage);
+      await vi.waitFor(() => expect(fakeQuery.interrupt).toHaveBeenCalledTimes(2));
+      fakeQuery.push({ type: "system", subtype: "init" } as unknown as SDKMessage);
+      fakeQuery.push({
+        type: "result",
+        subtype: "error_during_execution",
+        is_error: true,
+        terminal_reason: "aborted_streaming",
+      } as unknown as SDKMessage);
+      await vi.waitFor(() =>
+        expect(completed("turn-1")).toMatchObject({
+          event: { outcome: { status: "cancelled" } },
+        }),
+      );
+      await expect(session.execute(textTurn("turn-2"))).resolves.toMatchObject({ ok: true });
+      fakeQuery.push({
+        type: "result",
+        subtype: "success",
+        is_error: false,
+        terminal_reason: "completed",
+      } as unknown as SDKMessage);
+      await vi.waitFor(() =>
+        expect(completed("turn-2")).toMatchObject({
+          event: { outcome: { status: "succeeded" } },
+        }),
+      );
+      expect(
+        outputs.some(
+          (output) => output.kind === "event" && output.event.type === "turn.autonomous.started",
+        ),
+      ).toBe(false);
+    } finally {
+      await session.close();
+      await consuming;
+      await adapter.close();
+    }
+  });
+
   it("omits Auto when no runtime Model explicitly supports it", async () => {
     const { adapter, dependencies } = fixture();
     vi.mocked(dependencies.createInspector).mockReturnValueOnce({

packages/adapters/claude-code/test/sdk-transport.test.ts

@@ -2803,6 +2803,39 @@ describe("ClaudeSdkTransport confirmed background stops", () => {
     }
   });
 
+  it.each(["aborted_tools", "aborted_streaming", "completed"])(
+    "records cancellation before the first idle frame without trusting an unproven stop (%s)",
+    async (terminalReason) => {
+      const { transport, fakeQuery } = fixture();
+      const terminals: ClaudeTransportTurnResult[] = [];
+      transport.setIdleTurnHandler({
+        onEvent: () => undefined,
+        onTerminal: (result) => terminals.push(result),
+      });
+      await transport.start();
+      transport.setIdleLive(true);
+      try {
+        await transport.abortContinuation();
+        expect(fakeQuery.interrupt).toHaveBeenCalledOnce();
+        fakeQuery.push({
+          type: "result",
+          subtype: terminalReason === "completed" ? "success" : "error_during_execution",
+          is_error: terminalReason !== "completed",
+          terminal_reason: terminalReason,
+        } as unknown as SDKMessage);
+        await vi.waitFor(() =>
+          expect(terminals).toEqual([
+            terminalReason === "completed"
+              ? { status: "failed", kind: "cancellationUnproven" }
+              : { status: "cancelled", reason: terminalReason },
+          ]),
+        );
+      } finally {
+        await transport.close();
+      }
+    },
+  );
+
   it("interrupts a Root continuation that runs after the requested Turn", async () => {
     const value = fixture();
     const idleEvents: ClaudeTurnEvent[] = [];