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.
2026-08-18 21:51:38 +02:00
parent 25fbc939e0
commit f7e4606452
5 changed files with 244 additions and 2 deletions
@@ -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<typeof vi.fn>;
@@ -15,6 +15,7 @@ interface QueryMock {
  supportedModels: ReturnType<typeof vi.fn>;
  supportedCommands: ReturnType<typeof vi.fn>;
  rewindFiles: ReturnType<typeof vi.fn>;
+  cancelAsyncMessage: ReturnType<typeof vi.fn>;
  [Symbol.asyncIterator]: () => AsyncIterator<Record<string, unknown>, 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<string, unknown>) => {
      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<AgentStreamEvent>,
+  matches: (event: AgentStreamEvent) => boolean,
+): Promise<void> {
+  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<AgentStreamEvent>;
+}> {
+  let query: ScriptedQuery | null = null;
+  queryFactory.mockImplementation(({ prompt }: { prompt: AsyncIterable<unknown> }) => {
+    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[] = [];
@@ -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<SDKUserMessage> | 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<string>();
  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<void> {
+    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<boolean>;
+      }
+    ).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"],
+  };
+}
@@ -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();