From f7e4606452c385b7472ca2085f037b2493e4151f Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Tue, 18 Aug 2026 21:51:38 +0200 Subject: [PATCH] fix(server): discard unread Claude steers on interrupt A steer pushed into the live SDK query input survived query.interrupt(): the SDK dequeued it afterwards and resumed the turn the user had just stopped. Codex clears pending turn input when it aborts a turn; Claude does not, so its adapter now cancels the SDK messages it queued before interrupting. Interrupt-discards is the deliberate ceiling for steer cancellation: Codex exposes no un-steer operation, so a cross-provider take-back cannot exist. A steer the model never read keeps its transcript row. --- docs/glossary.md | 1 + docs/providers.md | 4 +- ...agent.interrupt-restart-regression.test.ts | 105 +++++++++++++++++- .../server/agent/providers/claude/agent.ts | 65 +++++++++++ ...d-during-tool-call-claude.real.e2e.test.ts | 71 ++++++++++++ 5 files changed, 244 insertions(+), 2 deletions(-) diff --git a/docs/glossary.md b/docs/glossary.md index dd0d9fe4d..84d3e4d31 100644 --- a/docs/glossary.md +++ b/docs/glossary.md @@ -43,6 +43,7 @@ Authoritative terminology. UI label wins. Don't invent synonyms; use what's here - **Agent controls** — Provider, model, mode, thinking, and provider-feature controls for an agent or draft agent. Code: `AgentControls` / `DraftAgentControls` (`packages/app/src/composer/agent-controls/index.tsx`). Forbidden: "Agent status bar". - **Composer footer** — Optional area rendered below the composer input but still inside the keyboard-shifted composer layout. Code: `Composer.footer` (`packages/app/src/composer/index.tsx`). - **Composer track** — A contextual lane at the foot of a pane, by the composer input. Specific tracks use the ` track` form: **Queue track**, **Subagents track**, **Tasks track**. The subagents and tasks tracks are pills floating over the transcript, each opening a panel with its rows; the queue track is a lane inside the composer itself. Code: queue track inside `Composer` (`packages/app/src/composer/index.tsx`), pill and panel frame in `packages/app/src/composer/tracks.tsx`, `SubagentsTrack` (`packages/app/src/subagents/track.tsx`), `AgentTaskList` (`packages/app/src/composer/task-list/index.tsx`). +- **Steer** — A follow-up admitted into an agent's already-running turn instead of interrupting it. Interrupting the turn discards steers the provider has not read yet; there is no take-back short of that, and no provider offers one. The stop button and Escape both interrupt. A steer the provider never read keeps its transcript row, matching Codex. UI: "Send and steer". Code: `steerActiveTurn` (`packages/server/src/server/agent/agent-sdk-types.ts`). Don't confuse with **Queue track**, which holds messages client-side until the turn ends and never enters the running turn. Forbidden: "Delivery", "Queued prompt". - **Subagent** — User-facing term for an agent session related to a parent agent session. Use **subagents** in UI copy and docs. Internal daemon/provider plumbing may say "child agent" or `child_session`, especially for provider-managed imports; do not surface "child agent" as a product term. - **Attachment tray** — The selected-attachments row inside the composer input, above the text input. Code: `renderAttachmentTray` (`packages/app/src/composer/index.tsx`). Forbidden: "Attachment bar". - **Fork** — Starting a NEW agent session seeded with a copy of another session's curated chat history. The history rides along as a `chat_history` composer attachment on a draft; the fork does **not** resume or branch the provider-side session, and the source session keeps running untouched. Two targets, one surface, one RPC (`agent.fork_context`, gated on the `agentForkContext` host feature). UI: the fork menu lives in the assistant turn footer and offers "Fork in a new tab" / "Fork in a new workspace". On a **completed** turn it sits beside the copy button and pins a boundary to that turn, so the fork stops where the turn ended. On the **in-flight** turn it sits beside the progress loader and elapsed timer, and omits the boundary entirely — `selectForkContextRows` projects the whole timeline when no boundary is given, so the fork captures everything **up to now**, including the response still streaming. That omission is what makes mid-run forking work; pinning a boundary there would silently drop the live response. Forbidden on provider-subagent (read-only) panes. Code: `useForkAgent` (`packages/app/src/hooks/use-fork-agent.ts`), `selectForkContextRows` (`packages/server/src/server/agent/activity-curator.ts`). Don't confuse with: git fork (unrelated; Paseo has no such concept), **Worktree** (filesystem isolation, not history copying). diff --git a/docs/providers.md b/docs/providers.md index 9ae54d5fb..304070583 100644 --- a/docs/providers.md +++ b/docs/providers.md @@ -81,7 +81,9 @@ OpenCode owns user message IDs. Do not pass Paseo-generated IDs to OpenCode prom `AgentManager` owns the one canonical timeline row for a foreground prompt carrying a Paseo `clientMessageId`. It records that row when `startTurn` accepts, with the wire `messageId` set to the same value. Provider adapters still emit their native user-message echo with the same `clientMessageId` when available; the manager records its provider identity on the internal row without changing or redispatching the wire item. If an adapter emits the echo before `startTurn` resolves, the manager records the provider identity with the row at acceptance. Provider adapters continue to own externally initiated user rows that have no Paseo client identity. Do not perform global transcript text dedupe. -Active-turn steering is an optional `AgentSession.steerActiveTurn` operation. The manager owns admission against its exact foreground turn, canonical user-message creation, echo reconciliation, and falls back to the normal interrupt-and-replace path only when the adapter reports `unavailable`. An adapter error is ambiguous delivery and must surface without an interrupt or retry. Codex calls `turn/steer` with the native expected turn and Paseo client user-message ID. Claude pushes an admitted steer into the exact active SDK query input; isolated control commands remain unavailable. The released OpenCode v1 adapter does not implement this operation. +Active-turn steering is an optional `AgentSession.steerActiveTurn` operation. The manager owns admission against its exact foreground turn, canonical user-message creation, echo reconciliation, and falls back to the normal interrupt-and-replace path only when the adapter reports `unavailable`. An adapter error leaves the steer's fate ambiguous and must surface without an interrupt or retry. Codex calls `turn/steer` with the native expected turn and Paseo client user-message ID. Claude pushes an admitted steer into the exact active SDK query input; isolated control commands remain unavailable. The released OpenCode v1 adapter does not implement this operation. + +A steering adapter also owes its interrupt: stopping a turn must discard the steers the provider has not read yet, or one of them resumes the turn the user just stopped. Codex clears pending input when it aborts a turn; Claude does not, so its adapter cancels the SDK messages it queued before calling `query.interrupt()`. Rewind accepts the canonical wire `messageId` and resolves it to the provider identity before calling the adapter. A submitted prompt cannot be rewound until its provider echo supplies that identity. diff --git a/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts b/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts index 7e785bdda..10f3e99b3 100644 --- a/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.interrupt-restart-regression.test.ts @@ -3,7 +3,7 @@ import { afterEach, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; import { ClaudeAgentClient } from "./agent.js"; import { streamSession } from "../test-utils/session-stream-adapter.js"; -import type { AgentStreamEvent } from "../../agent-sdk-types.js"; +import type { AgentSession, AgentStreamEvent } from "../../agent-sdk-types.js"; interface QueryMock { next: ReturnType; @@ -15,6 +15,7 @@ interface QueryMock { supportedModels: ReturnType; supportedCommands: ReturnType; rewindFiles: ReturnType; + cancelAsyncMessage: ReturnType; [Symbol.asyncIterator]: () => AsyncIterator, void>; } @@ -139,6 +140,7 @@ function createScriptedQuery(params: { supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), supportedCommands: vi.fn(async () => []), rewindFiles: vi.fn(async () => ({ canRewind: true })), + cancelAsyncMessage: vi.fn(async () => true), emit: (message: Record) => { output.push(message); }, @@ -195,6 +197,18 @@ async function collectUntilTerminal( return events; } +/** Pulls the stream forward without closing it, so the caller can keep reading afterwards. */ +async function consumeUntil( + stream: AsyncGenerator, + matches: (event: AgentStreamEvent) => boolean, +): Promise { + while (true) { + const next = await stream.next(); + if (next.done) throw new Error("Stream ended before the expected event"); + if (matches(next.value)) return; + } +} + function collectAssistantText(events: AgentStreamEvent[]): string { return events .flatMap((event) => { @@ -285,6 +299,95 @@ test("interrupt only calls query.interrupt and leaves the query open", async () await session.close(); }); +async function startSteeredTurn(sessionId: string): Promise<{ + session: AgentSession; + query: () => ScriptedQuery | null; + turn: AsyncGenerator; +}> { + let query: ScriptedQuery | null = null; + queryFactory.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { + query = createScriptedQuery({ prompt, sessionId }); + return query; + }); + + const session = await new ClaudeAgentClient({ + logger: createTestLogger(), + queryFactory, + resolveBinary: async () => "/test/claude/bin", + }).createSession({ provider: "claude", cwd: process.cwd() }); + + const turn = streamSession(session, "original running prompt"); + const start = await turn.next(); + if (!start.value || start.value.type !== "turn_started" || !start.value.turnId) { + throw new Error("Expected the original Claude turn to start"); + } + await waitFor(() => query?.prompts.length === 1); + + const steered = await session.steerActiveTurn!("queued steer", { + expectedTurnId: start.value.turnId, + clientMessageId: "steer-client", + }); + expect(steered).toEqual({ status: "accepted" }); + await waitFor(() => query?.prompts.length === 2); + + return { session, query: () => query, turn }; +} + +test("interrupt discards a queued steer so it cannot resume the stopped turn", async () => { + const { session, query, turn } = await startSteeredTurn("queued-steer-discard-session"); + + await session.interrupt(); + await waitFor(() => query()?.interrupt.mock.calls.length === 1); + + expect(query()?.cancelAsyncMessage).toHaveBeenCalledWith(query()?.prompts[1]?.uuid); + expect(await collectUntilTerminal(turn)).toContainEqual( + expect.objectContaining({ type: "turn_canceled" }), + ); + + // Nothing is queued any more, so a second interrupt has no steer left to discard. + query()?.cancelAsyncMessage.mockClear(); + await session.interrupt(); + await waitFor(() => query()?.interrupt.mock.calls.length === 2); + expect(query()?.cancelAsyncMessage).not.toHaveBeenCalled(); + + await session.close(); +}); + +test("interrupt still stops the turn when Claude has already dequeued the steer", async () => { + const { session, query, turn } = await startSteeredTurn("queued-steer-declined-session"); + query()?.cancelAsyncMessage.mockResolvedValue(false); + + await session.interrupt(); + await waitFor(() => query()?.interrupt.mock.calls.length === 1); + + expect(query()?.cancelAsyncMessage).toHaveBeenCalledWith(query()?.prompts[1]?.uuid); + expect(await collectUntilTerminal(turn)).toContainEqual( + expect.objectContaining({ type: "turn_canceled" }), + ); + + await session.close(); +}); + +test("a steer Claude has already read is no longer discardable on interrupt", async () => { + const { session, query, turn } = await startSteeredTurn("queued-steer-completed-session"); + + query()?.emit({ + type: "command_lifecycle", + command_uuid: query()?.prompts[1]?.uuid, + state: "completed", + }); + // Frames are translated in order, so the marker landing proves the lifecycle frame was read. + query()?.emit({ type: "assistant", message: { content: "STEER_READ" } }); + await consumeUntil(turn, (event) => collectAssistantText([event]).includes("STEER_READ")); + + await session.interrupt(); + await waitFor(() => query()?.interrupt.mock.calls.length === 1); + + expect(query()?.cancelAsyncMessage).not.toHaveBeenCalled(); + await collectUntilTerminal(turn); + await session.close(); +}); + test("reuses the existing query after interrupt before starting the next prompt", async () => { const logger = createTestLogger(); const queries: ScriptedQuery[] = []; diff --git a/packages/server/src/server/agent/providers/claude/agent.ts b/packages/server/src/server/agent/providers/claude/agent.ts index 0f4cab9e8..fad5f027e 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -2026,6 +2026,12 @@ class ClaudeAgentSession implements AgentSession { /** The exact SDK query/input pair that owns the current foreground turn. */ private activeForegroundQuery: Query | null = null; private activeForegroundInput: AsyncMessageInput | null = null; + /** + * Steers pushed into the live SDK input that Claude may not have read yet. Interrupting the turn + * has to discard them, or the SDK dequeues one and resumes the turn we just stopped. SDK user + * UUIDs are provider-private; never let them escape the adapter boundary. + */ + private readonly queuedSteerUuids = new Set(); private claudeSessionId: string | null; private persistence: AgentPersistenceHandle | null; private currentMode: PermissionMode; @@ -2289,6 +2295,9 @@ class ClaudeAgentSession implements AgentSession { ) { return { status: "unavailable" }; } + if (message.uuid) { + this.queuedSteerUuids.add(message.uuid); + } input.push(message); return { status: "accepted" }; } @@ -3877,6 +3886,7 @@ class ClaudeAgentSession implements AgentSession { return; } this.pendingInterruptAbort = true; + await this.discardQueuedSteers(queryToInterrupt); try { await this.awaitWithTimeout( queryToInterrupt.interrupt(), @@ -3887,6 +3897,31 @@ class ClaudeAgentSession implements AgentSession { } } + /** + * Interrupt means interrupt: a steer Claude never read dies with the turn instead of resuming it. + * A steer already dequeued cannot be recalled, and does not need to be — the interrupt kills it. + */ + private async discardQueuedSteers(query: Query): Promise { + const uuids = [...this.queuedSteerUuids]; + this.queuedSteerUuids.clear(); + if (uuids.length === 0) return; + // The SDK runtime supports this, but its public Query type has not caught up. Keep the + // compatibility escape hatch inside the Claude adapter. + const cancelAsyncMessage = ( + query as Query & { + cancelAsyncMessage?: (uuid: string) => Promise; + } + ).cancelAsyncMessage; + if (!cancelAsyncMessage) return; + for (const uuid of uuids) { + try { + await cancelAsyncMessage.call(query, uuid); + } catch (error) { + this.logger.warn({ err: error }, "Failed to discard a queued Claude steer"); + } + } + } + private translateMessageToEvents( message: SDKMessage, options?: { @@ -3933,6 +3968,8 @@ class ClaudeAgentSession implements AgentSession { } } + this.forgetReadSteer(message); + switch (message.type) { case "system": this.appendSystemMessageEvents(message, events); @@ -3965,6 +4002,13 @@ class ClaudeAgentSession implements AgentSession { return events; } + /** Once Claude has read a steer there is nothing left to discard on interrupt. */ + private forgetReadSteer(message: unknown): void { + const lifecycle = readClaudeCommandLifecycle(message); + if (!lifecycle || lifecycle.state === "queued") return; + this.queuedSteerUuids.delete(lifecycle.commandUuid); + } + private appendTaskStateEvent(message: SDKMessage, events: AgentStreamEvent[]): void { const item = this.taskState.observe(message); if (item) events.push({ type: "timeline", provider: "claude", item }); @@ -6104,3 +6148,24 @@ function extractClaudeUserText(messageRaw: unknown): string | null { } return null; } + +interface ClaudeCommandLifecycle { + commandUuid: string; + state: "queued" | "started" | "completed"; +} + +/** Runtime-only Claude frames are validated here because the SDK's public union omits them. */ +function readClaudeCommandLifecycle(message: unknown): ClaudeCommandLifecycle | null { + const record = toObjectRecord(message); + if (record?.type !== "command_lifecycle") return null; + if ( + typeof record.command_uuid !== "string" || + !["queued", "started", "completed"].includes(String(record.state)) + ) { + return null; + } + return { + commandUuid: record.command_uuid, + state: record.state as ClaudeCommandLifecycle["state"], + }; +} diff --git a/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts index 0756a802c..6638c425a 100644 --- a/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/send-during-tool-call-claude.real.e2e.test.ts @@ -504,6 +504,77 @@ describe("daemon E2E (real claude) - send message during tool call", () => { } }, 210_000); + test("Stop cancels a queued Claude steer before it can resume the interrupted turn", async () => { + const logger = pino({ level: "silent" }); + const cwd = tmpCwd(); + const daemon = await createTestPaseoDaemon({ + agentClients: { claude: new ClaudeAgentClient({ logger }) }, + logger, + }); + const client = new DaemonClient({ url: `ws://127.0.0.1:${daemon.port}/ws` }); + const collector = createMessageCollector(client); + try { + await within("connect queued-steer Stop client", 15_000, client.connect()); + await within( + "subscribe queued-steer Stop client", + 15_000, + client.fetchAgents({ subscribe: { subscriptionId: "queued-steer-stop" } }), + ); + const agent = await within( + "create queued-steer Stop agent", + 30_000, + client.createAgent({ + cwd, + title: "claude-queued-steer-stop", + ...getRealProviderConfig("claude"), + }), + ); + await within( + "start original Claude sleep turn", + 30_000, + client.sendAgentMessage( + agent.id, + "Use Bash to run exactly: sleep 5 in the foreground. Do not finish until instructed.", + { messageId: generateClientMessageId() }, + ), + ); + await within( + "wait for original Claude tool boundary", + 90_000, + waitForRunningClaudeSleep(client, collector, agent.id, 80_000), + ); + const steerId = generateClientMessageId(); + await within( + "admit queued Claude steer", + 30_000, + client.sendAgentMessage(agent.id, "RESUME_FORBIDDEN", { + messageId: steerId, + activeTurnBehavior: "steer", + }), + ); + const messagesBeforeStop = collector.messages.length; + await within("Stop original Claude turn", 30_000, client.cancelAgent(agent.id)); + await within("wait for stopped Claude turn", 60_000, client.waitForFinish(agent.id, 50_000)); + const afterStop = collector.messages.slice(messagesBeforeStop); + expect( + afterStop.filter( + (message) => + message.type === "agent_stream" && + message.payload.agentId === agent.id && + message.payload.event.type === "turn_started", + ), + ).toHaveLength(0); + // The steer's row stays in the transcript even though Claude never read it, matching Codex. + expect(getAssistantTexts(afterStop, agent.id).join("\n")).not.toContain("RESUME_FORBIDDEN"); + } finally { + const cleanup = await Promise.allSettled([client.close(), daemon.close()]); + rmSync(cwd, { recursive: true, force: true }); + expect( + cleanup.flatMap((result) => (result.status === "rejected" ? [result.reason] : [])), + ).toEqual([]); + } + }, 180_000); + test("sending a message while a tool call is running replaces the turn without error, idle flash, or autonomous fallback", async () => { const logger = pino({ level: "silent" }); const cwd = tmpCwd();