feat(steering): settle a steered turn at idle, not at the interrupt (#958)

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.
This commit is contained in:
Andrey Bragin
2026-08-05 18:41:00 +02:00
committed by GitHub
parent b965dd2191
commit a84b81080a
3 changed files with 396 additions and 8 deletions

View File

@@ -29,9 +29,11 @@
* !steer <text> `_session/steering` — injected into the turn that is * !steer <text> `_session/steering` — injected into the turn that is
* already running, so it can change course between tool * already running, so it can change course between tool
* calls. Answers "injected", or "startedNewTurn" if the turn * calls. Answers "injected", or "startedNewTurn" if the turn
* beat us to the finish. Either way the reply arrives as * beat us to the finish. An injected reply streams as
* `session/update` notifications that can outlast the * `session/update` notifications inside that turn, whose
* `stopReason` of the turn you steered. See ./steering.ts. * `stopReason` waits for it. "startedNewTurn" output belongs
* to a detached turn no `session/prompt` of ours tracks.
* See ./steering.ts.
* !queue <text> `session/prompt` sent mid-turn anyway. The agent advertises * !queue <text> `session/prompt` sent mid-turn anyway. The agent advertises
* `promptQueueing`, so it takes the backlog instead of us — * `promptQueueing`, so it takes the backlog instead of us —
* same ordering, different owner. * same ordering, different owner.
@@ -564,8 +566,8 @@ async function main() {
try { try {
const result = await agent.request<SteeringResponse>(STEERING_METHOD, params); const result = await agent.request<SteeringResponse>(STEERING_METHOD, params);
// Either way the steered message's own output arrives as `session/update` // Either way the steered message's own output arrives as `session/update`
// notifications, not in this response — and it may well keep streaming // notifications, not in this response. When injected, it lands inside the
// after the turn we steered has already answered with its stopReason. // turn we steered, whose `stopReason` waits for it.
log( log(
result.outcome === "injected" result.outcome === "injected"
? "steer outcome: injected into the running turn" ? "steer outcome: injected into the running turn"

View File

@@ -361,6 +361,26 @@ type Turn = {
* pre-hold behavior (pending wakes are not countable: notifications can * pre-hold behavior (pending wakes are not countable: notifications can
* batch into one followup). */ * batch into one followup). */
deferredSettle?: PromptResponse; 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<string>;
/** 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; resolve: (response: PromptResponse) => void;
reject: (error: unknown) => void; reject: (error: unknown) => void;
}; };
@@ -689,6 +709,12 @@ function isHeldOpen(
return turn != null && turn.deferredSettle !== undefined && !turn.settled; 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<string> } {
return turn != null && turn.steeredEchoes !== undefined && !turn.settled;
}
/** Disarm the force-cancel backstop (see Session.forceCancelTimer). Every /** Disarm the force-cancel backstop (see Session.forceCancelTimer). Every
* path that settles the active turn must run this so a timer can never fire * 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 * 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 * calls). The steered message's own output streams via `session/update`, not
* this response. * 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 * When the session is idle, the opt-in path returns `promptRequired` WITHOUT
* calling `prompt()`, pushing SDK input, or mutating `turnQueue`: the content * calling `prompt()`, pushing SDK input, or mutating `turnQueue`: the content
* stays Host-owned so the Host can submit it through a standard * 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 // 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 // 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. // 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) { if (!turnInFlight) {
const promptRequest: PromptRequest = { const promptRequest: PromptRequest = {
sessionId, sessionId,
@@ -1929,10 +1959,23 @@ export class ClaudeAcpAgent {
prompt: params.prompt, prompt: params.prompt,
}; };
const userMessage = promptToClaude(promptRequest); 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 // Deliver into the running turn rather than queuing behind it as a fresh
// prompt would. // prompt would.
userMessage.priority = STEER_PRIORITY; 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); session.input.push(userMessage);
return { outcome: "injected" }; return { outcome: "injected" };
} }
@@ -2263,6 +2306,14 @@ export class ClaudeAcpAgent {
* calling settleActive directly bypasses the hold and re-opens the * calling settleActive directly bypasses the hold and re-opens the
* out-of-turn permission deadlock (issue #866) through its lane. */ * out-of-turn permission deadlock (issue #866) through its lane. */
const settleOrDefer = (outcome: PromptResponse) => { 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 ( if (
session.activeTurn && session.activeTurn &&
!session.activeTurn.settled && !session.activeTurn.settled &&
@@ -2752,7 +2803,31 @@ export class ClaudeAcpAgent {
// this lagged idle (no active turn to settle): the idle // this lagged idle (no active turn to settle): the idle
// still belongs to that settled turn, and skipping the // still belongs to that settled turn, and skipping the
// decrement would leak the debt permanently. // 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--; 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 ( } else if (
!session.cancelled && !session.cancelled &&
session.activeTurn && session.activeTurn &&
@@ -3201,7 +3276,16 @@ export class ClaudeAcpAgent {
// turn's OWN result — a followup result arriving inside the // turn's OWN result — a followup result arriving inside the
// cancel window still gets its own trailer and must be counted, // cancel window still gets its own trailer and must be counted,
// or that idle would later false-fail the next prompt. // 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++; session.owedTrailingIdles++;
} }
@@ -3654,6 +3738,23 @@ export class ClaudeAcpAgent {
// either. Its trailing-idle debt stands and is absorbed // either. Its trailing-idle debt stands and is absorbed
// when the drain idle eventually arrives. // when the drain idle eventually arrives.
settleActive(session.activeTurn.deferredSettle); 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 { } else {
settleActive({ stopReason: "end_turn", usage: sessionUsage(session) }); settleActive({ stopReason: "end_turn", usage: sessionUsage(session) });
} }
@@ -3670,6 +3771,13 @@ export class ClaudeAcpAgent {
break; break;
} }
if ("isReplay" in message && message.isReplay) { 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). // Unrelated replay (e.g. the echo of an already-settled turn).
break; 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 // 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 // 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 // block — issue #680), force the consumer to settle the active turn

View File

@@ -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) => { const waitFor = async (cond: () => boolean) => {
for (let i = 0; i < 200; i++) { for (let i = 0; i < 200; i++) {
if (cond()) return; if (cond()) return;
@@ -10029,6 +10054,249 @@ describe("turn steering (_session/steering)", () => {
expect(JSON.stringify(injected.message.content)).toContain("also handle X"); 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<void>((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 () => { it("always injects at 'now' priority, ignoring any client-supplied priority", async () => {
const agent = createMockAgent(); const agent = createMockAgent();
const captured: any[] = []; const captured: any[] = [];