diff --git a/docs/harnesses/claude-code/claude-code-steer.md b/docs/harnesses/claude-code/claude-code-steer.md index 88a118475..d4253ba9f 100644 --- a/docs/harnesses/claude-code/claude-code-steer.md +++ b/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 中模型回复的是插入要求的文本。 diff --git a/packages/adapters/claude-code/src/claude-code-adapter.ts b/packages/adapters/claude-code/src/claude-code-adapter.ts index 233b5b922..f031e46bf 100644 --- a/packages/adapters/claude-code/src/claude-code-adapter.ts +++ b/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 } }; diff --git a/packages/adapters/claude-code/src/sdk-transport.ts b/packages/adapters/claude-code/src/sdk-transport.ts index 78e4c7cdb..badebd932 100644 --- a/packages/adapters/claude-code/src/sdk-transport.ts +++ b/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) { diff --git a/packages/adapters/claude-code/test/claude-code-adapter.test.ts b/packages/adapters/claude-code/test/claude-code-adapter.test.ts index d90e71fc1..51bf16a6a 100644 --- a/packages/adapters/claude-code/test/claude-code-adapter.test.ts +++ b/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 | 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({ diff --git a/packages/adapters/claude-code/test/sdk-transport.test.ts b/packages/adapters/claude-code/test/sdk-transport.test.ts index 22adb10e7..64a86c50e 100644 --- a/packages/adapters/claude-code/test/sdk-transport.test.ts +++ b/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[] = [];