From a84b81080a4127edf40bc448fc8bf2b15503304d Mon Sep 17 00:00:00 2001 From: Andrey Bragin Date: Wed, 5 Aug 2026 18:41:00 +0200 Subject: [PATCH] feat(steering): settle a steered turn at idle, not at the interrupt (#958) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Steering is delivered at priority `now`, which makes the CLI **abort** the running cycle instead of folding the message into it. The aborted cycle ends with an ordinary human-origin `result`, so we answered `session/prompt` with `end_turn` while the steered work was still going — its remaining tool calls, its answer, and its file edits then streamed with no turn in flight. A steered turn's results now only record their outcome, and the turn settles at the SDK's `idle` — the only signal spanning both the interrupted and the steered cycle. The injected message's replay tells them apart: an idle arriving before it means the steered cycle hasn't started, so it is swallowed. `session/cancel` and the next prompt still settle the turn early, with the recorded stop reason. Four steering tests cover the interrupt, an early idle, the next-prompt hand-off, and cancel. --- examples/simple-client.ts | 12 +- src/acp-agent.ts | 124 ++++++++++++++++- src/tests/acp-agent.test.ts | 268 ++++++++++++++++++++++++++++++++++++ 3 files changed, 396 insertions(+), 8 deletions(-) diff --git a/examples/simple-client.ts b/examples/simple-client.ts index 92f1b706..13c92ab1 100644 --- a/examples/simple-client.ts +++ b/examples/simple-client.ts @@ -29,9 +29,11 @@ * !steer `_session/steering` — injected into the turn that is * already running, so it can change course between tool * calls. Answers "injected", or "startedNewTurn" if the turn - * beat us to the finish. Either way the reply arrives as - * `session/update` notifications that can outlast the - * `stopReason` of the turn you steered. See ./steering.ts. + * beat us to the finish. An injected reply streams as + * `session/update` notifications inside that turn, whose + * `stopReason` waits for it. "startedNewTurn" output belongs + * to a detached turn no `session/prompt` of ours tracks. + * See ./steering.ts. * !queue `session/prompt` sent mid-turn anyway. The agent advertises * `promptQueueing`, so it takes the backlog instead of us — * same ordering, different owner. @@ -564,8 +566,8 @@ async function main() { try { const result = await agent.request(STEERING_METHOD, params); // Either way the steered message's own output arrives as `session/update` - // notifications, not in this response — and it may well keep streaming - // after the turn we steered has already answered with its stopReason. + // notifications, not in this response. When injected, it lands inside the + // turn we steered, whose `stopReason` waits for it. log( result.outcome === "injected" ? "steer outcome: injected into the running turn" diff --git a/src/acp-agent.ts b/src/acp-agent.ts index 88a94f56..ff7262ca 100644 --- a/src/acp-agent.ts +++ b/src/acp-agent.ts @@ -361,6 +361,26 @@ type Turn = { * pre-hold behavior (pending wakes are not countable: notifications can * batch into one followup). */ deferredSettle?: PromptResponse; + /** Uuids of `steer()`-injected messages the SDK has not replayed back yet. + * + * A steer is delivered at {@link STEER_PRIORITY} (`now`), so the CLI ABORTS + * the running cycle: it emits its own human-origin `result` — + * indistinguishable from a turn's terminal one — and the steered message runs + * as a SECOND cycle. Settling at that result would answer `session/prompt` + * mid-work, so a steered turn's results only RECORD their outcome + * (`steeredSettle`) and it settles at the SDK's `idle`, the only signal + * spanning both cycles (CLI 2.1.220). + * + * A non-empty set at an idle means the steered cycle hasn't started — the CLI + * replays a message only when it picks it up, always after the interrupted + * cycle's result — so that idle is swallowed. Drained by the replay handler. + * + * Residual: a message the CLI drops unreplayed parks the turn until + * `session/cancel` or the next prompt (both settle it). */ + steeredEchoes?: Set; + /** What a steered turn settles with once its steered work has run: the outcome + * of its latest result, so its usage covers every cycle the turn ran. */ + steeredSettle?: PromptResponse; resolve: (response: PromptResponse) => void; reject: (error: unknown) => void; }; @@ -689,6 +709,12 @@ function isHeldOpen( return turn != null && turn.deferredSettle !== undefined && !turn.settled; } +/** Whether a steer moved this turn's settlement off the next result and onto the + * SDK's `idle` (see Turn.steeredEchoes). Shared by the consumer's settle lanes. */ +function isSteering(turn: Turn | null | undefined): turn is Turn & { steeredEchoes: Set } { + return turn != null && turn.steeredEchoes !== undefined && !turn.settled; +} + /** Disarm the force-cancel backstop (see Session.forceCancelTimer). Every * path that settles the active turn must run this so a timer can never fire * on an already-settled turn — and must leave the field undefined, or the @@ -1884,6 +1910,10 @@ export class ClaudeAcpAgent { * calls). The steered message's own output streams via `session/update`, not * this response. * + * Pre-empting means ABORTING: the interrupted cycle emits a `result` of its + * own and the steered message runs as a second one, so the turn is marked + * (`Turn.steeredEchoes`) to settle at the SDK's `idle` instead of that result. + * * When the session is idle, the opt-in path returns `promptRequired` WITHOUT * calling `prompt()`, pushing SDK input, or mutating `turnQueue`: the content * stays Host-owned so the Host can submit it through a standard @@ -1904,7 +1934,7 @@ export class ClaudeAcpAgent { // which is exactly the window in which steering is meaningful. This check // and the active-path push below stay in one synchronous section so the // turn cannot settle in the gap between deciding to inject and enqueueing. - const turnInFlight = (session.turnQueue ?? []).some((turn) => !turn.settled); + const turnInFlight = (session.turnQueue ?? []).find((turn) => !turn.settled); if (!turnInFlight) { const promptRequest: PromptRequest = { sessionId, @@ -1929,10 +1959,23 @@ export class ClaudeAcpAgent { prompt: params.prompt, }; const userMessage = promptToClaude(promptRequest); - userMessage.uuid = randomUUID(); + const steeredUuid = randomUUID(); + userMessage.uuid = steeredUuid; // Deliver into the running turn rather than queuing behind it as a fresh // prompt would. userMessage.priority = STEER_PRIORITY; + // Mark before the push and in the same synchronous section as the in-flight + // check: the interrupt can have the CLI finalizing the aborted cycle by the + // time the consumer next runs, and an unmarked result would settle the turn + // (see Turn.steeredEchoes). + (turnInFlight.steeredEchoes ??= new Set()).add(steeredUuid); + // A turn already held for background subagents has a recorded outcome the + // steer supersedes: move it into the steer lane so one lane owns settlement. + // The idle handler re-applies the hold through the subagent gate. + if (turnInFlight.deferredSettle !== undefined) { + turnInFlight.steeredSettle = turnInFlight.deferredSettle; + turnInFlight.deferredSettle = undefined; + } session.input.push(userMessage); return { outcome: "injected" }; } @@ -2263,6 +2306,14 @@ export class ClaudeAcpAgent { * calling settleActive directly bypasses the hold and re-opens the * out-of-turn permission deadlock (issue #866) through its lane. */ const settleOrDefer = (outcome: PromptResponse) => { + // No result ends a steered turn: the steer aborted the cycle this result + // may belong to, and the steered one is still to come. Record the outcome + // for the idle lane (see Turn.steeredEchoes); later cycles overwrite it, + // so the last result and its usage win. + if (isSteering(session.activeTurn)) { + session.activeTurn.steeredSettle = outcome; + return; + } if ( session.activeTurn && !session.activeTurn.settled && @@ -2752,7 +2803,31 @@ export class ClaudeAcpAgent { // this lagged idle (no active turn to settle): the idle // still belongs to that settled turn, and skipping the // decrement would leak the debt permanently. + // Deliberately BEFORE the steer lane below: an owed idle + // belongs to an earlier turn, and reading it as the steered + // sequence's turn-over signal would settle a turn whose + // steered cycle is still running. Steered results owe none. session.owedTrailingIdles--; + } else if (isSteering(session.activeTurn)) { + // A steered turn settles here, not at a result: this idle is + // the only signal spanning the interrupted and steered cycles + // (see Turn.steeredEchoes). An idle before the steered echo, + // or before any result was recorded, means the answer is + // still ahead — swallow it, and never let it reach the #825 + // fail below, which would reject a prompt about to answer. + // + // Plain Turn so the fields can be cleared below; isSteering + // narrows steeredEchoes to non-optional. + const steered: Turn = session.activeTurn; + if (steered.steeredEchoes?.size === 0 && steered.steeredSettle !== undefined) { + // Via the subagent gate, not settleActive: a steered turn + // can also have spawned background subagents, which own it + // from here (settles now if none is live, holds otherwise). + steered.deferredSettle = steered.steeredSettle; + steered.steeredEchoes = undefined; + steered.steeredSettle = undefined; + settleDeferredIfDrained(); + } } else if ( !session.cancelled && session.activeTurn && @@ -3201,7 +3276,16 @@ export class ClaudeAcpAgent { // turn's OWN result — a followup result arriving inside the // cancel window still gets its own trailer and must be counted, // or that idle would later false-fail the next prompt. - if (isAutonomousResult || !session.cancelled || !session.activeTurn) { + // A second exclusion: a steered turn's own results (see + // Turn.steeredEchoes). One idle covers the whole interrupted + + // steered sequence and that idle settles the turn, so counting + // either result would leave a debt that swallows it. Autonomous + // results inside a steered turn keep their own trailers. + const owesTrailingIdle = isAutonomousResult || !isSteering(session.activeTurn); + if ( + owesTrailingIdle && + (isAutonomousResult || !session.cancelled || !session.activeTurn) + ) { session.owedTrailingIdles++; } @@ -3654,6 +3738,23 @@ export class ClaudeAcpAgent { // either. Its trailing-idle debt stands and is absorbed // when the drain idle eventually arrives. settleActive(session.activeTurn.deferredSettle); + } else if ( + isSteering(session.activeTurn) && + session.activeTurn.steeredSettle !== undefined + ) { + // Same for a turn with an open steer (see + // Turn.steeredEchoes): the user moving on outranks the + // steered continuation, but its last result's stop reason + // stands. With no result yet it falls through to the + // end_turn hand-off below, owing nothing there. + // + // The steer lane normally pays for that result's trailing + // idle by settling on it; this hand-off settles instead, + // so count it here or it arrives un-owed against the fresh + // turn and false-fails it (issue #825). Harmless if it + // never comes: the debt absorbs one future idle. + session.owedTrailingIdles++; + settleActive(session.activeTurn.steeredSettle); } else { settleActive({ stopReason: "end_turn", usage: sessionUsage(session) }); } @@ -3670,6 +3771,13 @@ export class ClaudeAcpAgent { break; } if ("isReplay" in message && message.isReplay) { + // A steered message's echo matches no turn (steer() creates + // none), but it marks the steered cycle as running — how the idle + // lane tells "the answer is still ahead" from "the turn is over" + // (see Turn.steeredEchoes). + if (isSteering(session.activeTurn)) { + session.activeTurn.steeredEchoes.delete(message.uuid); + } // Unrelated replay (e.g. the echo of an already-settled turn). break; } @@ -4147,6 +4255,16 @@ export class ClaudeAcpAgent { } } + // A steered active turn (see Turn.steeredEchoes) settles "cancelled" in the + // consumer at the interrupt's trailing idle, like any live turn. But the + // steer lane pays for its last result's trailer by settling on it, which + // this cancel pre-empts, so count that trailer here or it arrives un-owed + // against the next prompt and false-fails it (issue #825). Over-counting + // when only one trailer comes absorbs a future idle. + if (isSteering(session.activeTurn) && session.activeTurn.steeredSettle !== undefined) { + session.owedTrailingIdles++; + } + // Arm a backstop before interrupting: if a turn is actively consuming the // query and interrupt() doesn't make the SDK yield (e.g. a wedged TaskOutput // block — issue #680), force the consumer to settle the active turn diff --git a/src/tests/acp-agent.test.ts b/src/tests/acp-agent.test.ts index 1c420244..88420c47 100644 --- a/src/tests/acp-agent.test.ts +++ b/src/tests/acp-agent.test.ts @@ -9834,6 +9834,31 @@ describe("turn steering (_session/steering)", () => { }; } + /** The `result` the CLI emits for the cycle a priority:'now' steer aborted, as + * observed on CLI 2.1.220: `origin.kind` "human" (not autonomous, so it + * reaches the user-turn lifecycle) and `stop_reason` "tool_use" — the abort + * landed on a pending tool call. No test here depends on either: every + * non-error combination maps to the `end_turn` clients observe. */ + function interruptedCycleResult() { + return { ...createResultMessage(), stop_reason: "tool_use", origin: { kind: "human" } }; + } + + /** The SDK's authoritative turn-over signal. */ + function idleMessage() { + return { type: "system", subtype: "session_state_changed", state: "idle" }; + } + + /** Records every `agent_message_chunk` the client observes. */ + function timelineClient(timeline: string[]) { + return { + sessionUpdate: async (notification: any) => { + if (notification.update?.sessionUpdate === "agent_message_chunk") { + timeline.push(notification.update.content?.text); + } + }, + } as unknown as AcpClient; + } + const waitFor = async (cond: () => boolean) => { for (let i = 0; i < 200; i++) { if (cond()) return; @@ -10029,6 +10054,249 @@ describe("turn steering (_session/steering)", () => { expect(JSON.stringify(injected.message.content)).toContain("also handle X"); }); + // A steer does not join the running cycle — it ABORTS it: the CLI interrupts + // the in-flight query as soon as a queued message carries priority 'now'. The + // interrupted cycle ends with an ordinary human-origin `result` and the + // steered message runs as a SECOND cycle. Treating that first result as + // terminal answers session/prompt mid-work: the client is told `end_turn`, + // then keeps receiving the turn's remaining tool calls and its actual answer + // as updates belonging to no turn. + it("keeps the turn open across the interrupt its own steer caused", async () => { + const timeline: string[] = []; + const agent = new ClaudeAcpAgent(timelineClient(timeline), { + log: () => {}, + error: () => {}, + }); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); // turn becomes active + yield createAssistantText("working on it"); + // The steered message is pushed at priority 'now'... + const steered = await iter.next(); + // ...so the CLI aborts the query, ending the interrupted cycle with a + // result of its own. No idle follows it: one comes at the very end, for + // the whole interrupted + steered sequence. + yield interruptedCycleResult(); + // Only now does the steered message run, as a second cycle. Its echo + // matches no queued turn (dropped as an unrelated replay), its output is + // the answer the user is waiting for, and its result has the last word + // on the turn's stop reason. + yield userEcho(steered.value); + yield createAssistantText("STEERED-OK"); + yield createResultMessage(); + yield idleMessage(); + } + return messageGenerator(); + }); + + const turn = agent + .prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "start" }] }) + .then((response) => { + timeline.push(`prompt:${response.stopReason}`); + return response; + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + + await expect( + agent.steer({ sessionId: "test-session", prompt: [{ type: "text", text: "also handle X" }] }), + ).resolves.toEqual({ outcome: "injected" }); + + const response = await turn; + // Drain the rest of the stream so the timeline is complete either way — on + // an early settle the steered output arrives after `response`. + await agent.sessions["test-session"]?.consumer; + + // The steered continuation belongs to the turn the client is waiting on, so + // all of it precedes that turn's response. A `prompt:` entry anywhere but + // last is the bug: updates outlived the stopReason. + expect(timeline).toEqual(["working on it", "STEERED-OK", "prompt:end_turn"]); + // Both cycles ran for this one prompt, so its usage covers both (2 × the + // mock result's 10 in / 5 out) rather than stopping at the interrupt. + expect(response.usage).toEqual({ + inputTokens: 20, + outputTokens: 10, + cachedReadTokens: 0, + cachedWriteTokens: 0, + totalTokens: 30, + }); + // Still exactly one response for one prompt, and nothing left behind. + expect(agent.sessions["test-session"].turnQueue).toHaveLength(0); + }); + + // The other steer ordering: the cycle had already finished when the steer + // landed, so nothing was aborted and its result is a natural one — followed by + // its own idle, before the steered message is picked up. Settling on that idle + // would answer the prompt with the steered work still ahead, so the turn waits + // for the steered message's echo (Turn.steeredEchoes). + it("does not settle a steered turn at an idle that precedes the steered echo", async () => { + const timeline: string[] = []; + const agent = new ClaudeAcpAgent(timelineClient(timeline), { + log: () => {}, + error: () => {}, + }); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield createAssistantText("working on it"); + const steered = await iter.next(); + // The cycle ends and goes idle before the steered message is dequeued. + yield interruptedCycleResult(); + yield idleMessage(); + // Then the steered message runs as its own cycle. + yield userEcho(steered.value); + yield createAssistantText("STEERED-OK"); + yield createResultMessage(); + yield idleMessage(); + } + return messageGenerator(); + }); + + const turn = agent + .prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "start" }] }) + .then((response) => { + timeline.push(`prompt:${response.stopReason}`); + return response; + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + await agent.steer({ + sessionId: "test-session", + prompt: [{ type: "text", text: "also handle X" }], + }); + + await turn; + await agent.sessions["test-session"]?.consumer; + + expect(timeline).toEqual(["working on it", "STEERED-OK", "prompt:end_turn"]); + // The swallowed idle must not leave a phantom debt that would absorb a later + // turn's trailing idle (and blind the issue-#825 detector). + expect(agent.sessions["test-session"].owedTrailingIdles).toBe(0); + }); + + // A steer must not hold a prompt hostage: the next prompt's echo hands the + // steered turn off, reporting — like the subagent hold — the stop reason its + // last result recorded rather than a guessed one. + it("hands a steered turn off to the next prompt with its recorded outcome", async () => { + const timeline: string[] = []; + const agent = new ClaudeAcpAgent(timelineClient(timeline), { + log: () => {}, + error: () => {}, + }); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield createAssistantText("working on it"); + await iter.next(); // the steered message, which this CLI never replays + yield interruptedCycleResult(); + // The user moves on: the second prompt's echo arrives while the steered + // turn is still open. + const second = await iter.next(); + yield userEcho(second.value); + yield createAssistantText("second turn"); + yield createResultMessage(); + yield idleMessage(); + } + return messageGenerator(); + }); + + const first = agent + .prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "start" }] }) + .then((response) => { + timeline.push(`first:${response.stopReason}`); + return response; + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + await agent.steer({ + sessionId: "test-session", + prompt: [{ type: "text", text: "also handle X" }], + }); + + const second = agent + .prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "next" }] }) + .then((response) => { + timeline.push(`second:${response.stopReason}`); + return response; + }); + + const firstResponse = await first; + await second; + await agent.sessions["test-session"]?.consumer; + + // The steered turn ends at the hand-off — before the new turn's output — and + // carries the interrupted cycle's own usage, not the next turn's. + expect(timeline).toEqual(["working on it", "first:end_turn", "second turn", "second:end_turn"]); + expect(firstResponse.usage).toEqual({ + inputTokens: 10, + outputTokens: 5, + cachedReadTokens: 0, + cachedWriteTokens: 0, + totalTokens: 15, + }); + expect(agent.sessions["test-session"].turnQueue).toHaveLength(0); + }); + + // Cancelling a steered turn answers "cancelled" through the normal lane, but + // its last result's trailing idle went uncounted (the steer lane pays for its + // idles by settling on them, which the cancel pre-empted). That idle can lag + // past the next prompt's echo, where an un-owed idle reads as "the SDK went + // idle without a result" and false-fails a healthy prompt (#825). + it("cancels a steered turn without false-failing the next prompt", async () => { + const agent = createMockAgent(); + let releaseAfterCancel = () => {}; + const afterCancel = new Promise((resolve) => (releaseAfterCancel = resolve)); + + injectGeneratorSession(agent, (input) => { + async function* messageGenerator() { + const iter = input[Symbol.asyncIterator](); + const u1 = await iter.next(); + yield userEcho(u1.value); + yield createAssistantText("working on it"); + await iter.next(); // the steered message + yield interruptedCycleResult(); + await afterCancel; + yield idleMessage(); // the interrupt's trailer: settles the turn "cancelled" + const second = await iter.next(); + yield userEcho(second.value); // activates the next turn, clears `cancelled` + yield idleMessage(); // the steered result's lagged trailer — must be absorbed + yield createAssistantText("second turn"); + yield createResultMessage(); + yield idleMessage(); + } + return messageGenerator(); + }); + + const first = agent.prompt({ + sessionId: "test-session", + prompt: [{ type: "text", text: "start" }], + }); + await waitFor(() => !!agent.sessions["test-session"]?.activeTurn); + await agent.steer({ + sessionId: "test-session", + prompt: [{ type: "text", text: "also handle X" }], + }); + // The interrupted cycle's result has been recorded on the turn (not settled). + await waitFor(() => agent.sessions["test-session"]?.activeTurn?.steeredSettle !== undefined); + + await agent.cancel({ sessionId: "test-session" }); + releaseAfterCancel(); + + await expect(first).resolves.toEqual(expect.objectContaining({ stopReason: "cancelled" })); + // The next prompt runs to completion instead of being rejected by the #825 + // detector when the steered turn's lagged idle lands. + await expect( + agent.prompt({ sessionId: "test-session", prompt: [{ type: "text", text: "next" }] }), + ).resolves.toEqual(expect.objectContaining({ stopReason: "end_turn" })); + expect(agent.sessions["test-session"].owedTrailingIdles).toBe(0); + }); + it("always injects at 'now' priority, ignoring any client-supplied priority", async () => { const agent = createMockAgent(); const captured: any[] = [];