From c46c03158e90967e058aa27418a76d6b58374701 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Sat, 21 Mar 2026 19:47:56 +0700 Subject: [PATCH] Simplify Claude agent session flow --- .../claude-agent.integration.test.ts | 414 +++ ...agent.interrupt-restart-regression.test.ts | 2470 +++-------------- .../server/agent/providers/claude-agent.ts | 1654 ++--------- .../providers/claude/sidechain-tracker.ts | 324 +++ 4 files changed, 1383 insertions(+), 3479 deletions(-) create mode 100644 packages/server/src/server/agent/providers/claude-agent.integration.test.ts create mode 100644 packages/server/src/server/agent/providers/claude/sidechain-tracker.ts diff --git a/packages/server/src/server/agent/providers/claude-agent.integration.test.ts b/packages/server/src/server/agent/providers/claude-agent.integration.test.ts new file mode 100644 index 000000000..d2f02c4d5 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude-agent.integration.test.ts @@ -0,0 +1,414 @@ +import { describe, expect, test, beforeAll } from "vitest"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import path from "node:path"; +import pino from "pino"; + +import type { + AgentSession, + AgentStreamEvent, + ToolCallTimelineItem, +} from "../agent-sdk-types.js"; +import { isCommandAvailable } from "../provider-launch-config.js"; +import { ClaudeAgentClient } from "./claude-agent.js"; + +const logger = pino({ level: "silent" }); +const client = new ClaudeAgentClient({ logger }); + +function tmpCwd(prefix: string): string { + return mkdtempSync(path.join(tmpdir(), prefix)); +} + +function compactText(value: string): string { + return value.replace(/\s+/g, "").toLowerCase(); +} + +function isTerminalEvent(event: AgentStreamEvent): boolean { + return ( + event.type === "turn_completed" || + event.type === "turn_failed" || + event.type === "turn_canceled" + ); +} + +async function nextStreamEvent( + stream: AsyncGenerator, + timeoutMs: number, + label: string +): Promise> { + let timer: ReturnType | null = null; + try { + return await Promise.race([ + stream.next(), + new Promise((_, reject) => { + timer = setTimeout(() => { + reject(new Error(`Timed out waiting for ${label}`)); + }, timeoutMs); + }), + ]); + } finally { + if (timer) { + clearTimeout(timer); + } + } +} + +async function collectUntilTerminal( + stream: AsyncGenerator, + options?: { + timeoutMs?: number; + onEvent?: (event: AgentStreamEvent) => Promise | void; + } +): Promise { + const events: AgentStreamEvent[] = []; + while (true) { + const next = await nextStreamEvent( + stream, + options?.timeoutMs ?? 45_000, + "stream event" + ); + if (next.done || !next.value) { + return events; + } + const event = next.value; + events.push(event); + await options?.onEvent?.(event); + if (isTerminalEvent(event)) { + return events; + } + } +} + +async function collectUntil( + stream: AsyncGenerator, + predicate: (event: AgentStreamEvent) => boolean, + timeoutMs = 45_000 +): Promise { + const events: AgentStreamEvent[] = []; + while (true) { + const next = await nextStreamEvent(stream, timeoutMs, "matching stream event"); + if (next.done || !next.value) { + throw new Error("Stream ended before the expected event arrived"); + } + const event = next.value; + events.push(event); + if (predicate(event) || isTerminalEvent(event)) { + return events; + } + } +} + +function getAssistantText(events: AgentStreamEvent[]): string { + return events + .flatMap((event) => { + if (event.type !== "timeline" || event.item.type !== "assistant_message") { + return []; + } + return [event.item.text]; + }) + .join("\n"); +} + +function getToolCalls(events: AgentStreamEvent[]): ToolCallTimelineItem[] { + return events.flatMap((event) => { + if (event.type !== "timeline" || event.item.type !== "tool_call") { + return []; + } + return [event.item]; + }); +} + +function getLatestCompletedBashCall( + events: AgentStreamEvent[] +): ToolCallTimelineItem | undefined { + return [...getToolCalls(events)] + .reverse() + .find( + (item) => + item.status === "completed" && + item.name.toLowerCase() === "bash" + ); +} + +function getInternalQuery(session: AgentSession): unknown { + return (session as AgentSession & { query?: unknown | null }).query ?? null; +} + +async function createSession(params?: { + cwdPrefix?: string; + modeId?: string; + title?: string; +}): Promise<{ cwd: string; session: AgentSession }> { + const cwd = tmpCwd(params?.cwdPrefix ?? "claude-agent-integration-"); + const session = await client.createSession({ + provider: "claude", + cwd, + title: params?.title ?? "ClaudeAgentSession integration", + modeId: params?.modeId ?? "acceptEdits", + model: "haiku", + }); + return { cwd, session }; +} + +async function cleanupSession(handle: { + cwd: string; + session: AgentSession; +}): Promise { + await handle.session.close().catch(() => undefined); + rmSync(handle.cwd, { recursive: true, force: true }); +} + +describe("ClaudeAgentSession integration", () => { + beforeAll(() => { + expect(isCommandAvailable("claude")).toBe(true); + }); + + test( + "streams a basic response turn end-to-end", + async () => { + const handle = await createSession({ + cwdPrefix: "claude-agent-basic-response-", + }); + + try { + const events = await collectUntilTerminal( + handle.session.stream("Respond with exactly: HELLO_WORLD") + ); + + expect(events[0]).toMatchObject({ + type: "turn_started", + provider: "claude", + }); + expect( + events.some( + (event) => + event.type === "timeline" && + event.item.type === "assistant_message" && + compactText(event.item.text).includes("hello_world") + ) + ).toBe(true); + expect(events.at(-1)).toMatchObject({ + type: "turn_completed", + provider: "claude", + }); + } finally { + await cleanupSession(handle); + } + }, + 60_000 + ); + + test( + "runs a real Bash tool call and completes it", + async () => { + const handle = await createSession({ + cwdPrefix: "claude-agent-basic-tool-", + }); + + try { + const events = await collectUntilTerminal( + handle.session.stream( + [ + "Use the Bash tool.", + "Run exactly: echo TOOL_TEST_OUTPUT", + "After the command completes, reply with exactly: TOOL_DONE", + ].join(" ") + ) + ); + + const bashCalls = getToolCalls(events).filter( + (item) => item.name.toLowerCase() === "bash" + ); + const completedBashCall = getLatestCompletedBashCall(events); + + expect(bashCalls.length).toBeGreaterThan(0); + expect(completedBashCall).toBeDefined(); + expect(completedBashCall?.detail.type).toBe("shell"); + expect( + completedBashCall?.detail.type === "shell" && + completedBashCall.detail.output?.includes("TOOL_TEST_OUTPUT") + ).toBe(true); + expect(compactText(getAssistantText(events))).toContain("tool_done"); + expect(events.at(-1)).toMatchObject({ + type: "turn_completed", + provider: "claude", + }); + } finally { + await cleanupSession(handle); + } + }, + 60_000 + ); + + test( + "interrupts a running Bash turn and continues on the same query", + async () => { + const handle = await createSession({ + cwdPrefix: "claude-agent-interrupt-continue-", + }); + + try { + const firstStream = handle.session.stream( + [ + "Use the Bash tool.", + "Run exactly: sleep 10", + "Do not use a background task.", + "Do not do anything after starting the command.", + ].join(" ") + ); + + const initialEvents = await collectUntil( + firstStream, + (event) => + event.type === "timeline" && + event.item.type === "tool_call" && + event.item.name.toLowerCase() === "bash", + 45_000 + ); + const firstQuery = getInternalQuery(handle.session); + + expect(firstQuery).toBeTruthy(); + + await handle.session.interrupt(); + + const canceledEvents = await collectUntilTerminal(firstStream, { + timeoutMs: 20_000, + }); + const allFirstTurnEvents = [...initialEvents, ...canceledEvents]; + + expect( + allFirstTurnEvents.some( + (event) => + event.type === "turn_canceled" && event.provider === "claude" + ) + ).toBe(true); + + const followUpEvents = await collectUntilTerminal( + handle.session.stream("Respond with exactly: AFTER_INTERRUPT_OK") + ); + const secondQuery = getInternalQuery(handle.session); + + expect(secondQuery).toBe(firstQuery); + expect(compactText(getAssistantText(followUpEvents))).toContain( + "after_interrupt_ok" + ); + expect(followUpEvents.at(-1)).toMatchObject({ + type: "turn_completed", + provider: "claude", + }); + } finally { + await cleanupSession(handle); + } + }, + 60_000 + ); + + test( + "creates an autonomous live turn when a background task completes", + async () => { + const handle = await createSession({ + cwdPrefix: "claude-agent-autonomous-", + }); + const autonomousWakeToken = `AUTONOMOUS_WAKE_${Date.now().toString(36)}`; + + try { + const liveEventsStream = handle.session.streamLiveEvents(); + const foregroundEvents = await collectUntilTerminal( + handle.session.stream( + [ + "Use the Task tool to start a background sub-agent.", + "In that task, run the Bash command exactly: sleep 3 && echo BACKGROUND_DONE", + "Do not wait for task completion.", + "Reply immediately with exactly: SPAWNED", + `When the background task completes later, reply with exactly: ${autonomousWakeToken}`, + ].join(" ") + ), + { timeoutMs: 45_000 } + ); + + expect(compactText(getAssistantText(foregroundEvents))).toContain("spawned"); + + const liveEvents = await collectUntilTerminal(liveEventsStream, { + timeoutMs: 45_000, + }); + + expect( + liveEvents.some( + (event) => event.type === "turn_started" && event.provider === "claude" + ) + ).toBe(true); + expect( + compactText(getAssistantText(liveEvents)) + ).toContain(autonomousWakeToken.toLowerCase()); + expect(liveEvents.at(-1)).toMatchObject({ + type: "turn_completed", + provider: "claude", + }); + } finally { + await cleanupSession(handle); + } + }, + 60_000 + ); + + test( + "surfaces permission requests and resumes after approval", + async () => { + const handle = await createSession({ + cwdPrefix: "claude-agent-permission-", + modeId: "default", + }); + const permissionFile = path.join(handle.cwd, "permission.txt"); + + try { + const events = await collectUntilTerminal( + handle.session.stream( + [ + "Use the Bash tool to run exactly: printf 'PERM_TEST' > permission.txt", + "If approval is required, wait for approval.", + "After the command succeeds, reply with exactly: PERM_DONE", + ].join(" ") + ), + { + timeoutMs: 45_000, + onEvent: async (event) => { + if (event.type !== "permission_requested") { + return; + } + await handle.session.respondToPermission(event.request.id, { + behavior: "allow", + }); + }, + } + ); + + const permissionRequest = events.find( + (event): event is Extract => + event.type === "permission_requested" + ); + const permissionResolved = events.find( + (event): event is Extract => + event.type === "permission_resolved" + ); + const completedBashCall = getLatestCompletedBashCall(events); + + expect(permissionRequest?.request.kind).toBe("tool"); + expect(permissionResolved).toMatchObject({ + type: "permission_resolved", + provider: "claude", + resolution: { behavior: "allow" }, + }); + expect(completedBashCall).toBeDefined(); + expect(readFileSync(permissionFile, "utf8")).toBe("PERM_TEST"); + expect(compactText(getAssistantText(events))).toContain("perm_done"); + expect(events.at(-1)).toMatchObject({ + type: "turn_completed", + provider: "claude", + }); + } finally { + await cleanupSession(handle); + } + }, + 60_000 + ); +}); 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 ffa1d0cb1..24835f73a 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 @@ -1,28 +1,8 @@ -import { afterEach, beforeEach, describe, expect, test, vi } from "vitest"; +import { afterEach, describe, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../test-utils/test-logger.js"; import { ClaudeAgentClient } from "./claude-agent.js"; -import type { - AgentPersistenceHandle, - AgentStreamEvent, -} from "../agent-sdk-types.js"; - -type Deferred = { - promise: Promise; - resolve: (value: T) => void; - reject: (error: unknown) => void; -}; - -const sdkMocks = vi.hoisted(() => ({ - query: vi.fn(), - firstQuery: null as QueryMock | null, - secondQuery: null as QueryMock | null, - releaseOldAssistant: null as (() => void) | null, -})); - -vi.mock("@anthropic-ai/claude-agent-sdk", () => ({ - query: sdkMocks.query, -})); +import type { AgentStreamEvent } from "../agent-sdk-types.js"; type QueryMock = { next: ReturnType; @@ -33,16 +13,77 @@ type QueryMock = { supportedModels: ReturnType; supportedCommands: ReturnType; rewindFiles: ReturnType; + [Symbol.asyncIterator]: () => AsyncIterator, void>; }; -function deferred(): Deferred { - let resolve!: (value: T) => void; - let reject!: (error: unknown) => void; - const promise = new Promise((res, rej) => { - resolve = res; - reject = rej; - }); - return { promise, resolve, reject }; +type PromptRecord = { + text: string; + uuid: string | null; +}; + +type AsyncQueue = { + push: (value: T) => void; + next: () => Promise>; + end: () => void; +}; + +type ScriptedQuery = QueryMock & { + emit: (message: Record) => void; + end: () => void; + prompts: PromptRecord[]; +}; + +type PromptHandler = (input: { + prompt: Record; + promptRecord: PromptRecord; + query: ScriptedQuery; +}) => void | Promise; + +const sdkMocks = vi.hoisted(() => ({ + query: vi.fn(), +})); + +vi.mock("@anthropic-ai/claude-agent-sdk", () => ({ + query: sdkMocks.query, +})); + +function createAsyncQueue(): AsyncQueue { + const items: T[] = []; + const resolvers: Array<(value: IteratorResult) => void> = []; + let ended = false; + + return { + push(value) { + if (ended) { + return; + } + const resolve = resolvers.shift(); + if (resolve) { + resolve({ value, done: false }); + return; + } + items.push(value); + }, + async next() { + const value = items.shift(); + if (value !== undefined) { + return { value, done: false }; + } + if (ended) { + return { value: undefined, done: true }; + } + return await new Promise>((resolve) => { + resolvers.push(resolve); + }); + }, + end() { + ended = true; + while (resolvers.length > 0) { + const resolve = resolvers.shift(); + resolve?.({ value: undefined, done: true }); + } + }, + }; } function buildUsage() { @@ -53,148 +94,91 @@ function buildUsage() { }; } -function createPromptUuidReader(prompt: AsyncIterable) { - const iterator = prompt[Symbol.asyncIterator](); - let cached: Promise | null = null; - return async () => { - if (!cached) { - cached = iterator.next().then((next) => { - if (next.done) { - return null; - } - const value = next.value as { uuid?: unknown } | undefined; - return typeof value?.uuid === "string" ? value.uuid : null; +function buildSuccessResult(sessionId: string) { + return { + type: "result", + subtype: "success", + usage: buildUsage(), + total_cost_usd: 0, + session_id: sessionId, + }; +} + +function extractPromptText(message: Record): string { + const content = (message.message as { content?: unknown } | undefined)?.content; + if (typeof content === "string") { + return content; + } + if (!Array.isArray(content)) { + return ""; + } + return content + .flatMap((block) => { + if (!block || typeof block !== "object") { + return []; + } + const text = (block as { text?: unknown }).text; + return typeof text === "string" ? [text] : []; + }) + .join(""); +} + +function createScriptedQuery(params: { + prompt: AsyncIterable; + sessionId: string; + handlePrompt?: PromptHandler; +}): ScriptedQuery { + const output = createAsyncQueue>(); + const prompts: PromptRecord[] = []; + + const scriptedQuery = { + next: vi.fn(() => output.next()), + interrupt: vi.fn(async () => undefined), + return: vi.fn(async () => { + output.end(); + }), + setPermissionMode: vi.fn(async () => undefined), + setModel: vi.fn(async () => undefined), + supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), + supportedCommands: vi.fn(async () => []), + rewindFiles: vi.fn(async () => ({ canRewind: true })), + emit: (message: Record) => { + output.push(message); + }, + end: () => { + output.end(); + }, + prompts, + [Symbol.asyncIterator]() { + return this; + }, + } satisfies ScriptedQuery; + + scriptedQuery.emit({ + type: "system", + subtype: "init", + session_id: params.sessionId, + permissionMode: "default", + model: "opus", + }); + + void (async () => { + for await (const prompt of params.prompt) { + const promptMessage = prompt as Record; + const promptRecord = { + text: extractPromptText(promptMessage), + uuid: typeof promptMessage.uuid === "string" ? promptMessage.uuid : null, + }; + prompts.push(promptRecord); + await params.handlePrompt?.({ + prompt: promptMessage, + promptRecord, + query: scriptedQuery, }); } - return cached; - }; -} + })(); -function buildFirstQueryMock( - allowOldAssistant: Promise -): QueryMock { - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "interrupt-regression-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - await allowOldAssistant; - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "OLD_TURN_RESPONSE", - }, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => { - throw new Error("simulated interrupt failure"); - }), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - }; -} - -function buildSecondQueryMock(prompt: AsyncIterable): QueryMock { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "interrupt-regression-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "second prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "interrupt-regression-session", - isReplay: true, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "NEW_TURN_RESPONSE", - }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - }; + return scriptedQuery; } async function collectUntilTerminal( @@ -216,81 +200,48 @@ async function collectUntilTerminal( function collectAssistantText(events: AgentStreamEvent[]): string { return events - .filter( - (event): event is Extract => - event.type === "timeline" && event.item.type === "assistant_message" - ) - .map((event) => event.item.text) + .flatMap((event) => { + if (event.type !== "timeline" || event.item.type !== "assistant_message") { + return []; + } + return [event.item.text]; + }) .join(""); } -function collectUserText(events: AgentStreamEvent[]): string { - return events - .filter( - (event): event is Extract => - event.type === "timeline" && event.item.type === "user_message" - ) - .map((event) => event.item.text) - .join(""); +async function waitFor( + predicate: () => boolean, + options?: { timeoutMs?: number; intervalMs?: number } +): Promise { + const timeoutMs = options?.timeoutMs ?? 2_000; + const intervalMs = options?.intervalMs ?? 5; + const startedAt = Date.now(); + while (!predicate()) { + if (Date.now() - startedAt > timeoutMs) { + throw new Error("Timed out waiting for condition"); + } + await new Promise((resolve) => setTimeout(resolve, intervalMs)); + } } -function createTimedIteratorReader(params: { iterator: AsyncIterator }) { - const { iterator } = params; - let pendingNext: Promise> | null = null; +afterEach(() => { + sdkMocks.query.mockReset(); +}); - return { - async nextWithTimeout(timeoutMs: number): Promise> { - if (!pendingNext) { - pendingNext = iterator.next(); - } - const timeout = new Promise((resolve) => { - setTimeout(() => resolve(null), timeoutMs); - }); - const outcome = await Promise.race([ - pendingNext.then((result) => ({ kind: "result" as const, result })), - timeout.then(() => ({ kind: "timeout" as const })), - ]); - if (outcome.kind === "timeout") { - throw new Error("Timed out waiting for live event"); - } - pendingNext = null; - return outcome.result; - }, - }; -} - -describe("ClaudeAgentSession interrupt restart regression", () => { - beforeEach(() => { - const allowOldAssistant = deferred(); - let queryCreateCount = 0; - - sdkMocks.query.mockImplementation( - ({ prompt }: { prompt: AsyncIterable }) => { - queryCreateCount += 1; - if (queryCreateCount === 1) { - const mock = buildFirstQueryMock(allowOldAssistant.promise); - sdkMocks.firstQuery = mock; - return mock; - } - const mock = buildSecondQueryMock(prompt); - if (queryCreateCount === 2) { - sdkMocks.secondQuery = mock; - } - return mock; - } - ); - sdkMocks.releaseOldAssistant = () => allowOldAssistant.resolve(); - }); - - afterEach(() => { - sdkMocks.query.mockReset(); - sdkMocks.firstQuery = null; - sdkMocks.secondQuery = null; - sdkMocks.releaseOldAssistant = null; - }); - - test("starts a fresh query after interrupt failure to avoid stale old-turn response", async () => { +describe("ClaudeAgentSession interrupt regression", () => { + test("interrupt only calls query.interrupt and leaves the query open", async () => { const logger = createTestLogger(); + const queries: ScriptedQuery[] = []; + + sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { + const scriptedQuery = createScriptedQuery({ + prompt, + sessionId: "interrupt-keep-query-session", + }); + queries.push(scriptedQuery); + return scriptedQuery; + }); + const client = new ClaudeAgentClient({ logger }); const session = await client.createSession({ provider: "claude", @@ -299,249 +250,144 @@ describe("ClaudeAgentSession interrupt restart regression", () => { const firstTurn = session.stream("first prompt"); await firstTurn.next(); + await waitFor(() => queries[0]?.prompts.length === 1); - const secondTurnPromise = collectUntilTerminal(session.stream("second prompt")); - for (let attempt = 0; attempt < 40; attempt += 1) { - if (sdkMocks.secondQuery) { - break; - } - await new Promise((resolve) => setTimeout(resolve, 5)); - } - sdkMocks.releaseOldAssistant?.(); + await session.interrupt(); + await waitFor(() => queries[0]?.interrupt.mock.calls.length === 1); - const secondTurnEvents = await secondTurnPromise; - const secondAssistantText = collectAssistantText(secondTurnEvents); + expect(sdkMocks.query).toHaveBeenCalledTimes(1); + expect(queries[0]?.return).not.toHaveBeenCalled(); - expect(sdkMocks.firstQuery).toBeTruthy(); - expect(sdkMocks.secondQuery).toBeTruthy(); - expect(sdkMocks.firstQuery).not.toBe(sdkMocks.secondQuery); - expect(sdkMocks.firstQuery?.interrupt).toHaveBeenCalledTimes(1); - expect(sdkMocks.secondQuery?.next).toHaveBeenCalled(); - expect(secondAssistantText).toContain("NEW_TURN_RESPONSE"); - expect(secondAssistantText).not.toContain("OLD_TURN_RESPONSE"); + const firstTurnEvents = await collectUntilTerminal(firstTurn); + expect( + firstTurnEvents.find((event) => event.type === "turn_canceled") + ).toMatchObject({ + type: "turn_canceled", + provider: "claude", + reason: "Interrupted", + }); + + await session.close(); + }); + + test("pushes the next prompt into the existing query instead of rebuilding it", async () => { + const logger = createTestLogger(); + const queries: ScriptedQuery[] = []; + + sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { + const scriptedQuery = createScriptedQuery({ + prompt, + sessionId: "interrupt-reuse-query-session", + async handlePrompt({ promptRecord, query }) { + if (promptRecord.text !== "second prompt") { + return; + } + query.emit({ + type: "assistant", + message: { content: "SECOND_PROMPT_RESPONSE" }, + session_id: "interrupt-reuse-query-session", + }); + query.emit(buildSuccessResult("interrupt-reuse-query-session")); + }, + }); + queries.push(scriptedQuery); + return scriptedQuery; + }); + + const client = new ClaudeAgentClient({ logger }); + const session = await client.createSession({ + provider: "claude", + cwd: process.cwd(), + }); + + const firstTurn = session.stream("first prompt"); + await firstTurn.next(); + await waitFor(() => queries[0]?.prompts.length === 1); + + const secondTurnEvents = await collectUntilTerminal(session.stream("second prompt")); + + expect(sdkMocks.query).toHaveBeenCalledTimes(1); + expect(queries[0]?.prompts.map((prompt) => prompt.text)).toEqual([ + "first prompt", + "second prompt", + ]); + expect(queries[0]?.interrupt).toHaveBeenCalledTimes(1); + expect(queries[0]?.return).not.toHaveBeenCalled(); + expect(collectAssistantText(secondTurnEvents)).toContain("SECOND_PROMPT_RESPONSE"); await firstTurn.return?.(); await session.close(); }); - test("restarts after interrupt scaffold query drain without surfacing placeholder transcript noise", async () => { + test("recovers when the query pump sees a single interrupt abort before the next prompt", async () => { const logger = createTestLogger(); - let queryCreateCount = 0; + const output = createAsyncQueue>(); + const prompts: PromptRecord[] = []; + let throwAbortOnNext = false; sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - queryCreateCount += 1; - - if (queryCreateCount === 1) { - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "interrupt-scaffold-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "user", - message: { - role: "user", - content: [{ type: "text", text: "[Request interrupted by user]" }], - }, - parent_tool_use_id: null, - uuid: "interrupt-scaffold-1", - session_id: "interrupt-scaffold-session", - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "Got it. Sorry for the mess.", - }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "No response requested.", - }, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - } - - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { + const scriptedQuery = { next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "interrupt-scaffold-session", - permissionMode: "default", - model: "opus", - }, - }; + if (throwAbortOnNext) { + throwAbortOnNext = false; + throw new Error("Request was aborted."); } - if (step === 1) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "fresh prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "interrupt-scaffold-session", - isReplay: true, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "FRESH_RESPONSE_AFTER_INTERRUPT", - }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; + return output.next(); + }), + interrupt: vi.fn(async () => { + throwAbortOnNext = true; + }), + return: vi.fn(async () => { + output.end(); }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), setPermissionMode: vi.fn(async () => undefined), setModel: vi.fn(async () => undefined), supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), supportedCommands: vi.fn(async () => []), rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); + emit: (message: Record) => { + output.push(message); + }, + end: () => { + output.end(); + }, + prompts, + [Symbol.asyncIterator]() { + return this; + }, + } satisfies ScriptedQuery; - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); + scriptedQuery.emit({ + type: "system", + subtype: "init", + session_id: "interrupt-abort-recovery-session", + permissionMode: "default", + model: "opus", + }); - const events = await collectUntilTerminal(session.stream("fresh prompt")); - const assistantText = collectAssistantText(events); - const userText = collectUserText(events); - await session.close(); + void (async () => { + for await (const promptMessage of prompt) { + const record = promptMessage as Record; + const promptRecord = { + text: extractPromptText(record), + uuid: typeof record.uuid === "string" ? record.uuid : null, + }; + prompts.push(promptRecord); - expect(queryCreateCount).toBeGreaterThanOrEqual(2); - expect(assistantText).toContain("FRESH_RESPONSE_AFTER_INTERRUPT"); - expect(assistantText).not.toContain("Got it. Sorry for the mess."); - expect(assistantText).not.toContain("No response requested."); - expect(userText).not.toContain("[Request interrupted by user]"); - expect( - events.some( - (event) => - event.type === "turn_failed" && - event.error.includes("Claude stream ended before terminal result") - ) - ).toBe(false); - expect(events.some((event) => event.type === "turn_completed")).toBe(true); - }); + if (promptRecord.text !== "second prompt") { + continue; + } - test("ignores stale interrupted query completion after the replacement run starts", async () => { - const logger = createTestLogger(); - const releaseOldDone = deferred(); - let queryCreateCount = 0; + output.push({ + type: "assistant", + message: { content: "SECOND_PROMPT_RESPONSE" }, + session_id: "interrupt-abort-recovery-session", + }); + output.push(buildSuccessResult("interrupt-abort-recovery-session")); + } + })(); - sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - queryCreateCount += 1; - if (queryCreateCount === 1) { - let step = 0; - const mock = { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "interrupt-stale-done-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - await releaseOldDone.promise; - step += 1; - return { done: true, value: undefined }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - sdkMocks.firstQuery = mock; - return mock; - } - - const mock = buildSecondQueryMock(prompt); - if (queryCreateCount === 2) { - sdkMocks.secondQuery = mock; - } - return mock; + return scriptedQuery; }); const client = new ClaudeAgentClient({ logger }); @@ -552,145 +398,42 @@ describe("ClaudeAgentSession interrupt restart regression", () => { const firstTurn = session.stream("first prompt"); await firstTurn.next(); + await session.interrupt(); + await collectUntilTerminal(firstTurn); - const secondTurnPromise = collectUntilTerminal(session.stream("second prompt")); - for (let attempt = 0; attempt < 40; attempt += 1) { - if (sdkMocks.secondQuery) { - break; - } - await new Promise((resolve) => setTimeout(resolve, 5)); - } - releaseOldDone.resolve(undefined); + const secondTurnEvents = await collectUntilTerminal(session.stream("second prompt")); - const secondTurnEvents = await secondTurnPromise; - const secondAssistantText = collectAssistantText(secondTurnEvents); - - expect(sdkMocks.firstQuery?.interrupt).toHaveBeenCalledTimes(1); - expect(sdkMocks.secondQuery?.next).toHaveBeenCalled(); - expect(secondAssistantText).toContain("NEW_TURN_RESPONSE"); - expect( - secondTurnEvents.some( - (event) => - event.type === "turn_failed" && - event.error.includes("Claude stream ended before terminal result") - ) - ).toBe(false); + expect(sdkMocks.query).toHaveBeenCalledTimes(1); + expect(prompts.map((prompt) => prompt.text)).toEqual(["first prompt", "second prompt"]); + expect(collectAssistantText(secondTurnEvents)).toContain("SECOND_PROMPT_RESPONSE"); expect(secondTurnEvents.some((event) => event.type === "turn_completed")).toBe(true); - await firstTurn.return?.(); await session.close(); }); +}); - test("ignores stale task-notification assistant/result events queued before the current prompt", async () => { +describe("ClaudeAgentSession autonomous turns", () => { + test("creates an autonomous live turn when assistant output arrives without a foreground run", async () => { const logger = createTestLogger(); + let queryRef: ScriptedQuery | null = null; sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "task-notification-session", - permissionMode: "default", - model: "opus", - }, - }; + queryRef = createScriptedQuery({ + prompt, + sessionId: "autonomous-live-session", + async handlePrompt({ promptRecord, query }) { + if (promptRecord.text !== "seed prompt") { + return; } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "task_notification", - task_id: "task-123", - status: "completed", - output_file: "/tmp/task-123.txt", - summary: "Codex agent is done", - session_id: "task-notification-session", - uuid: "task-note-1", - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "STALE_TASK_NOTIFICATION_RESPONSE", - }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - if (step === 4) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "current prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "task-notification-session", - isReplay: true, - }, - }; - } - if (step === 5) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "CURRENT_PROMPT_RESPONSE", - }, - }, - }; - } - if (step === 6) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; + query.emit({ + type: "assistant", + message: { content: "SEED_RESPONSE" }, + session_id: "autonomous-live-session", + }); + query.emit(buildSuccessResult("autonomous-live-session")); + }, + }); + return queryRef; }); const client = new ClaudeAgentClient({ logger }); @@ -699,1422 +442,69 @@ describe("ClaudeAgentSession interrupt restart regression", () => { cwd: process.cwd(), }); - const events = await collectUntilTerminal(session.stream("current prompt")); - const assistantText = collectAssistantText(events); - - expect(assistantText).toContain("CURRENT_PROMPT_RESPONSE"); - expect(assistantText).not.toContain("STALE_TASK_NOTIFICATION_RESPONSE"); - - await session.close(); - }); - - test("ignores stale task-notification message_start bursts before prompt replay", async () => { - const logger = createTestLogger(); - - sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "task-notification-message-start-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "task_notification", - task_id: "task-msg-start-1", - status: "completed", - output_file: "/tmp/task-msg-start-1.txt", - summary: "Background task finished", - session_id: "task-notification-message-start-session", - uuid: "task-msg-start-note-1", - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - parent_tool_use_id: null, - event: { - type: "message_start", - message: { - id: "stale-msg-start-1", - role: "assistant", - model: "opus", - usage: { input_tokens: 1, output_tokens: 0 }, - }, - }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "STALE_MESSAGE_START_RESPONSE", - }, - }, - }; - } - if (step === 4) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - if (step === 5) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "current prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "task-notification-message-start-session", - isReplay: true, - }, - }; - } - if (step === 6) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "CURRENT_AFTER_MESSAGE_START", - }, - }, - }; - } - if (step === 7) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); - - const events = await collectUntilTerminal(session.stream("current prompt")); - const assistantText = collectAssistantText(events); - - expect(assistantText).toContain("CURRENT_AFTER_MESSAGE_START"); - expect(assistantText).not.toContain("STALE_MESSAGE_START_RESPONSE"); - - await session.close(); - }); - - test("ignores stale user-shaped task-notification message_start bursts before prompt replay", async () => { - const logger = createTestLogger(); - - sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "task-notification-user-message-start-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "user", - message: { - role: "user", - content: - "\ntask-msg-start-user-1\n", - }, - parent_tool_use_id: null, - uuid: "task-msg-start-user-1", - session_id: "task-notification-user-message-start-session", - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - parent_tool_use_id: null, - event: { - type: "message_start", - message: { - id: "stale-user-msg-start-1", - role: "assistant", - model: "opus", - usage: { input_tokens: 1, output_tokens: 0 }, - }, - }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "STALE_USER_TASK_NOTIFICATION_RESPONSE", - }, - }, - }; - } - if (step === 4) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - if (step === 5) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "current prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "task-notification-user-message-start-session", - isReplay: true, - }, - }; - } - if (step === 6) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "CURRENT_AFTER_USER_MESSAGE_START", - }, - }, - }; - } - if (step === 7) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); - - const events = await collectUntilTerminal(session.stream("current prompt")); - const assistantText = collectAssistantText(events); - - expect(assistantText).toContain("CURRENT_AFTER_USER_MESSAGE_START"); - expect(assistantText).not.toContain("STALE_USER_TASK_NOTIFICATION_RESPONSE"); - - await session.close(); - }); - - test("does not terminate the current prompt on a stale pre-prompt result event", async () => { - const logger = createTestLogger(); - - sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "stale-result-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - if (step === 2) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "current prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "stale-result-session", - isReplay: true, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "FRESH_AFTER_STALE_RESULT", - }, - }, - }; - } - if (step === 4) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); - - const events = await collectUntilTerminal(session.stream("current prompt")); - const assistantText = collectAssistantText(events); - - expect(assistantText).toContain("FRESH_AFTER_STALE_RESULT"); - - await session.close(); - }); - - test("does not create an orphan autonomous run from pre-replay task_started metadata", async () => { - const logger = createTestLogger(); - const keepQueryAlive = deferred(); - - sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "task-started-fallback-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - id: "tool-call-msg", - content: [ - { - type: "tool_use", - id: "toolu_1", - name: "Agent", - input: { description: "verify", prompt: "sub-task" }, - }, - ], - }, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "task_started", - task_id: "task-1", - tool_use_id: "toolu_1", - description: "verify", - task_type: "local_agent", - session_id: "task-started-fallback-session", - uuid: "task-started-1", - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - event: { - type: "message_delta", - delta: { stop_reason: "tool_use", stop_sequence: null }, - usage: buildUsage(), - }, - session_id: "task-started-fallback-session", - parent_tool_use_id: null, - uuid: "msg-delta-tool-use", - }, - }; - } - if (step === 4) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "current prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "task-started-fallback-session", - isReplay: true, - }, - }; - } - if (step === 5) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "FOREGROUND_DONE", - }, - }, - }; - } - if (step === 6) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - if (step === 7) { - await keepQueryAlive.promise; - return { done: true, value: undefined }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); - - const events = await collectUntilTerminal(session.stream("current prompt")); - const assistantText = collectAssistantText(events); - - expect(assistantText).toContain("FOREGROUND_DONE"); - expect( - (session as unknown as { turnState?: string }).turnState ?? null - ).toBe("idle"); - expect( - ( - session as unknown as { - runTracker?: { listActiveRuns: (owner?: "foreground" | "autonomous") => unknown[] }; - } - ).runTracker?.listActiveRuns("autonomous") ?? [] - ).toHaveLength(0); - - keepQueryAlive.resolve(undefined); - await session.close(); - }); - - test("ignores unmatched resumed-session errors without starting an autonomous run", async () => { - const logger = createTestLogger(); - const keepQueryAlive = deferred(); - - sdkMocks.query.mockImplementation(() => { - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "error_during_execution", - session_id: "new-session-after-missing-conversation", - errors: [ - "No conversation found with session ID: persisted-stale-session", - ], - num_turns: 0, - duration_ms: 0, - duration_api_ms: 0, - is_error: true, - stop_reason: null, - total_cost_usd: 0, - usage: buildUsage(), - }, - }; - } - await keepQueryAlive.promise; - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const handle: AgentPersistenceHandle = { - provider: "claude", - sessionId: "persisted-stale-session", - nativeHandle: "persisted-stale-session", - metadata: { - provider: "claude", - cwd: process.cwd(), - }, - }; - - const session = await client.resumeSession(handle, { cwd: process.cwd() }); - await new Promise((resolve) => setTimeout(resolve, 25)); - - const activeRuns = ( - session as unknown as { - runTracker: { listActiveRuns: (owner?: "foreground" | "autonomous") => unknown[] }; - } - ).runTracker.listActiveRuns(); - - expect(activeRuns).toHaveLength(0); - - keepQueryAlive.resolve(undefined); - await session.close(); - }); - - test("stops retrying live query pump when resumed Claude session no longer exists", async () => { - const logger = createTestLogger(); - let queryCreateCount = 0; - - sdkMocks.query.mockImplementation(() => { - queryCreateCount += 1; - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "error_during_execution", - session_id: "new-session-after-missing-conversation", - errors: [ - "No conversation found with session ID: persisted-stale-session", - ], - num_turns: 0, - duration_ms: 0, - duration_api_ms: 0, - is_error: true, - stop_reason: null, - total_cost_usd: 0, - usage: buildUsage(), - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const handle: AgentPersistenceHandle = { - provider: "claude", - sessionId: "persisted-stale-session", - nativeHandle: "persisted-stale-session", - metadata: { - provider: "claude", - cwd: process.cwd(), - }, - }; - - const session = await client.resumeSession(handle, { cwd: process.cwd() }); - const liveIterator = ( - session as unknown as { - streamLiveEvents: () => AsyncGenerator; - } - ).streamLiveEvents(); - const liveNext = liveIterator.next(); - - await new Promise((resolve) => setTimeout(resolve, 25)); - - await new Promise((resolve) => setTimeout(resolve, 650)); - - expect(queryCreateCount).toBe(1); - expect(session.describePersistence()).toBeNull(); - - await session.close(); - await liveNext; - }); - - test("does not emit live autonomous turn events for local_agent task_started during a foreground run", async () => { - const logger = createTestLogger(); - const keepQueryAlive = deferred(); - - sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "task-started-live-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - id: "tool-call-msg", - content: [ - { - type: "tool_use", - id: "toolu_live_1", - name: "Agent", - input: { description: "verify", prompt: "sub-task" }, - }, - ], - }, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "task_started", - task_id: "task-live-1", - tool_use_id: "toolu_live_1", - description: "verify", - task_type: "local_agent", - session_id: "task-started-live-session", - uuid: "task-started-live-1", - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - event: { - type: "content_block_start", - index: 2, - content_block: { - type: "tool_use", - id: "toolu_live_2", - name: "Agent", - input: {}, - caller: { type: "direct" }, - }, - }, - session_id: "task-started-live-session", - parent_tool_use_id: null, - uuid: "content-block-start-live-tool-use", - }, - }; - } - if (step === 4) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "current prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "task-started-live-session", - isReplay: true, - }, - }; - } - if (step === 5) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - content: "FOREGROUND_DONE", - }, - }, - }; - } - if (step === 6) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - if (step === 7) { - await keepQueryAlive.promise; - return { done: true, value: undefined }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); - - const foregroundEvents = await collectUntilTerminal(session.stream("current prompt")); - const liveIterator = ( - session as unknown as { - streamLiveEvents: () => AsyncGenerator; - } - ).streamLiveEvents(); - const timedReader = createTimedIteratorReader({ iterator: liveIterator }); - const liveEvents: AgentStreamEvent[] = []; - - for (let attempt = 0; attempt < 3; attempt += 1) { - try { - const next = await timedReader.nextWithTimeout(25); - if (next.done) { - break; - } - liveEvents.push(next.value); - } catch { - break; - } - } - - expect(collectAssistantText(foregroundEvents)).toContain("FOREGROUND_DONE"); - expect(liveEvents.some((event) => event.type === "turn_started")).toBe(false); - expect(liveEvents.some((event) => event.type === "turn_completed")).toBe(false); - - keepQueryAlive.resolve(undefined); - await session.close(); - }); - - test("does not let task_notification reservations steal a foreground terminal result", async () => { - const logger = createTestLogger(); - const keepQueryAlive = deferred(); - - sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "task-notification-foreground-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "task_notification", - task_id: "task-foreground-1", - tool_use_id: "toolu_foreground_1", - status: "completed", - output_file: "/tmp/task-foreground-1.txt", - summary: "Check Phase 1", - session_id: "task-notification-foreground-session", - uuid: "task-note-foreground-1", - }, - }; - } - if (step === 2) { - step += 1; - const promptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "verify prompt" }, - parent_tool_use_id: null, - uuid: promptUuid, - session_id: "task-notification-foreground-session", - isReplay: true, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - event: { - type: "message_start", - message: { - id: "foreground-tool-msg", - role: "assistant", - model: "opus", - usage: { input_tokens: 1, output_tokens: 0 }, - }, - }, - session_id: "task-notification-foreground-session", - parent_tool_use_id: null, - uuid: "foreground-message-start", - }, - }; - } - if (step === 4) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - event: { - type: "content_block_start", - index: 0, - content_block: { - type: "tool_use", - id: "toolu_foreground_2", - name: "Agent", - input: {}, - caller: { type: "direct" }, - }, - }, - session_id: "task-notification-foreground-session", - parent_tool_use_id: null, - uuid: "foreground-tool-use-start", - }, - }; - } - if (step === 5) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { - id: "foreground-final-msg", - content: "FOREGROUND_RESULT_STAYS_ATTACHED", - }, - session_id: "task-notification-foreground-session", - }, - }; - } - if (step === 6) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - event: { - type: "message_delta", - delta: { stop_reason: "end_turn", stop_sequence: null }, - usage: { input_tokens: 1, output_tokens: 1 }, - }, - session_id: "task-notification-foreground-session", - parent_tool_use_id: null, - uuid: "foreground-message-delta", - }, - }; - } - if (step === 7) { - step += 1; - return { - done: false, - value: { - type: "stream_event", - event: { type: "message_stop" }, - session_id: "task-notification-foreground-session", - parent_tool_use_id: null, - uuid: "foreground-message-stop", - }, - }; - } - if (step === 8) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - stop_reason: "end_turn", - session_id: "task-notification-foreground-session", - }, - }; - } - if (step === 9) { - await keepQueryAlive.promise; - return { done: true, value: undefined }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - }); - - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); - - const foregroundEvents = await collectUntilTerminal(session.stream("verify prompt")); - const liveIterator = ( - session as unknown as { - streamLiveEvents: () => AsyncGenerator; - } - ).streamLiveEvents(); - const timedReader = createTimedIteratorReader({ iterator: liveIterator }); - const liveEvents: AgentStreamEvent[] = []; - - for (let attempt = 0; attempt < 3; attempt += 1) { - try { - const next = await timedReader.nextWithTimeout(25); - if (next.done) { - break; - } - liveEvents.push(next.value); - } catch { - break; - } - } - - expect(collectAssistantText(foregroundEvents)).toContain( - "FOREGROUND_RESULT_STAYS_ATTACHED" - ); - expect(foregroundEvents.some((event) => event.type === "turn_completed")).toBe(true); - expect(liveEvents.some((event) => event.type === "turn_started")).toBe(false); - expect(liveEvents.some((event) => event.type === "turn_completed")).toBe(false); - - keepQueryAlive.resolve(undefined); - await session.close(); - }); - - test("emits autonomous live events from SDK stream when Claude wakes itself", async () => { - const logger = createTestLogger(); - let queryCreateCount = 0; - let localPromptUuid: string | null = null; - - sdkMocks.query.mockImplementation( - ({ prompt }: { prompt: AsyncIterable }) => { - queryCreateCount += 1; - if (queryCreateCount === 1) { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "live-autonomous-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - localPromptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "seed prompt" }, - parent_tool_use_id: null, - uuid: localPromptUuid, - session_id: "live-autonomous-session", - isReplay: true, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { content: "SEED_DONE" }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - } - - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "user", - message: { - role: "user", - content: - "\nbg-1\ncompleted\n", - }, - parent_tool_use_id: null, - uuid: "task-note-user-1", - session_id: "live-autonomous-session", - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { content: "AUTONOMOUS_WAKE_RESPONSE" }, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - } - ); - - const client = new ClaudeAgentClient({ logger }); - const session = await client.createSession({ - provider: "claude", - cwd: process.cwd(), - }); - await collectUntilTerminal(session.stream("seed prompt")); - expect(localPromptUuid).toBeTruthy(); - expect(session.describePersistence()?.sessionId).toBe("live-autonomous-session"); - for (let attempt = 0; attempt < 80; attempt += 1) { - const activeTurnPromise = ( - session as unknown as { activeTurnPromise?: Promise | null } - ).activeTurnPromise; - if (!activeTurnPromise) { - break; - } - await new Promise((resolve) => setTimeout(resolve, 25)); - } - expect( - (session as unknown as { activeTurnPromise?: Promise | null }) - .activeTurnPromise ?? null - ).toBeNull(); - const liveIterator = ( - session as unknown as { - streamLiveEvents: () => AsyncGenerator; - } - ).streamLiveEvents(); - const timedReader = createTimedIteratorReader({ iterator: liveIterator }); - const liveEvents: AgentStreamEvent[] = []; + const liveIterator = session.streamLiveEvents(); + queryRef?.emit({ + type: "assistant", + message: { content: "AUTONOMOUS_WAKE_RESPONSE" }, + session_id: "autonomous-live-session", + }); + queryRef?.emit(buildSuccessResult("autonomous-live-session")); - for (let attempt = 0; attempt < 8; attempt += 1) { - const next = await timedReader.nextWithTimeout(5_000); - if (next.done) { - break; - } - liveEvents.push(next.value); - if (next.value.type === "turn_completed") { - break; - } - } + const started = await liveIterator.next(); + const timeline = await liveIterator.next(); + const completed = await liveIterator.next(); - expect(liveEvents.some((event) => event.type === "turn_started")).toBe(true); - expect( - liveEvents.some( - (event) => - event.type === "timeline" && - event.item.type === "assistant_message" && - event.item.text.includes("AUTONOMOUS_WAKE_RESPONSE") - ) - ).toBe(true); - expect(liveEvents.some((event) => event.type === "turn_completed")).toBe(true); + expect(started.value).toMatchObject({ type: "turn_started", provider: "claude" }); + expect(timeline.value).toMatchObject({ + type: "timeline", + provider: "claude", + item: { + type: "assistant_message", + text: "AUTONOMOUS_WAKE_RESPONSE", + }, + }); + expect(completed.value).toMatchObject({ + type: "turn_completed", + provider: "claude", + }); await liveIterator.return?.(); await session.close(); }); - test("releases local-turn suppression when task notifications arrive as user payloads", async () => { + test("auto-completes an open autonomous turn when a foreground prompt starts", async () => { const logger = createTestLogger(); - let queryCreateCount = 0; - let localPromptUuid: string | null = null; + let queryRef: ScriptedQuery | null = null; - sdkMocks.query.mockImplementation( - ({ prompt }: { prompt: AsyncIterable }) => { - queryCreateCount += 1; - if (queryCreateCount === 1) { - const readPromptUuid = createPromptUuidReader(prompt); - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "system", - subtype: "init", - session_id: "live-task-user-session", - permissionMode: "default", - model: "opus", - }, - }; - } - if (step === 1) { - step += 1; - localPromptUuid = (await readPromptUuid()) ?? "missing-prompt-uuid"; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "seed prompt" }, - parent_tool_use_id: null, - uuid: localPromptUuid, - session_id: "live-task-user-session", - isReplay: true, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { content: "SEED_DONE" }, - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - } + sdkMocks.query.mockImplementation(({ prompt }: { prompt: AsyncIterable }) => { + queryRef = createScriptedQuery({ + prompt, + sessionId: "autonomous-handoff-session", + async handlePrompt({ promptRecord, query }) { + if (promptRecord.text === "seed prompt") { + query.emit({ + type: "assistant", + message: { content: "SEED_RESPONSE" }, + session_id: "autonomous-handoff-session", + }); + query.emit(buildSuccessResult("autonomous-handoff-session")); + return; + } - let step = 0; - return { - next: vi.fn(async () => { - if (step === 0) { - step += 1; - return { - done: false, - value: { - type: "user", - message: { role: "user", content: "seed prompt" }, - parent_tool_use_id: null, - uuid: localPromptUuid ?? "missing-prompt-uuid", - session_id: "live-task-user-session", - isReplay: true, - }, - }; - } - if (step === 1) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { content: "SHOULD_STAY_SUPPRESSED" }, - }, - }; - } - if (step === 2) { - step += 1; - return { - done: false, - value: { - type: "user", - message: { - role: "user", - content: - "\nbg-1\ncompleted\n", - }, - parent_tool_use_id: null, - uuid: "task-note-user-1", - session_id: "live-task-user-session", - }, - }; - } - if (step === 3) { - step += 1; - return { - done: false, - value: { - type: "assistant", - message: { content: "AUTONOMOUS_AFTER_TASK_NOTIFICATION" }, - }, - }; - } - if (step === 4) { - step += 1; - return { - done: false, - value: { - type: "result", - subtype: "success", - usage: buildUsage(), - total_cost_usd: 0, - }, - }; - } - return { done: true, value: undefined }; - }), - interrupt: vi.fn(async () => undefined), - return: vi.fn(async () => undefined), - setPermissionMode: vi.fn(async () => undefined), - setModel: vi.fn(async () => undefined), - supportedModels: vi.fn(async () => [{ value: "opus", displayName: "Opus" }]), - supportedCommands: vi.fn(async () => []), - rewindFiles: vi.fn(async () => ({ canRewind: true })), - } satisfies QueryMock; - } - ); + if (promptRecord.text === "foreground prompt") { + query.emit({ + type: "assistant", + message: { content: "FOREGROUND_RESPONSE" }, + session_id: "autonomous-handoff-session", + }); + query.emit(buildSuccessResult("autonomous-handoff-session")); + } + }, + }); + return queryRef; + }); const client = new ClaudeAgentClient({ logger }); const session = await client.createSession({ @@ -2123,75 +513,47 @@ describe("ClaudeAgentSession interrupt restart regression", () => { }); await collectUntilTerminal(session.stream("seed prompt")); - expect(localPromptUuid).toBeTruthy(); - expect(session.describePersistence()?.sessionId).toBe("live-task-user-session"); - for (let attempt = 0; attempt < 80; attempt += 1) { - const activeTurnPromise = ( - session as unknown as { activeTurnPromise?: Promise | null } - ).activeTurnPromise; - if (!activeTurnPromise) { - break; - } - await new Promise((resolve) => setTimeout(resolve, 25)); - } + + const liveIterator = session.streamLiveEvents(); + queryRef?.emit({ + type: "assistant", + message: { content: "BACKGROUND_ONLY_RESPONSE" }, + session_id: "autonomous-handoff-session", + }); + + const autonomousStart = await liveIterator.next(); + const autonomousTimeline = await liveIterator.next(); + const foregroundEvents = await collectUntilTerminal(session.stream("foreground prompt")); + const autonomousComplete = await liveIterator.next(); + + expect(autonomousStart.value).toMatchObject({ + type: "turn_started", + provider: "claude", + }); + expect(autonomousTimeline.value).toMatchObject({ + type: "timeline", + provider: "claude", + item: { + type: "assistant_message", + text: "BACKGROUND_ONLY_RESPONSE", + }, + }); + expect(autonomousComplete.value).toMatchObject({ + type: "turn_completed", + provider: "claude", + }); + expect(foregroundEvents.some((event) => event.type === "turn_completed")).toBe(true); + expect(collectAssistantText(foregroundEvents)).toContain("FOREGROUND_RESPONSE"); expect( - (session as unknown as { activeTurnPromise?: Promise | null }) - .activeTurnPromise ?? null - ).toBeNull(); - - const liveIterator = ( - session as unknown as { - streamLiveEvents: () => AsyncGenerator; - } - ).streamLiveEvents(); - const timedReader = createTimedIteratorReader({ iterator: liveIterator }); - const liveEvents: AgentStreamEvent[] = []; - - for (let attempt = 0; attempt < 12; attempt += 1) { - const next = await timedReader.nextWithTimeout(5_000); - if (next.done) { - break; - } - liveEvents.push(next.value); - if (next.value.type === "turn_completed") { - break; - } - } - - expect( - liveEvents.some( - (event) => - event.type === "timeline" && - event.item.type === "user_message" && - event.item.text.includes("") + [autonomousStart.value, autonomousTimeline.value, autonomousComplete.value].some( + (event) => event?.type === "turn_canceled" ) ).toBe(false); - expect( - liveEvents.some( - (event) => - event.type === "timeline" && - event.item.type === "tool_call" && - event.item.name === "task_notification" && - event.item.status === "completed" - ) - ).toBe(true); - expect(liveEvents.some((event) => event.type === "turn_started")).toBe(true); - expect( - liveEvents.some( - (event) => - event.type === "timeline" && - event.item.type === "assistant_message" && - event.item.text.includes("SHOULD_STAY_SUPPRESSED") - ) - ).toBe(false); - expect( - liveEvents.some( - (event) => - event.type === "timeline" && - event.item.type === "assistant_message" && - event.item.text.includes("AUTONOMOUS_AFTER_TASK_NOTIFICATION") - ) - ).toBe(true); + expect(sdkMocks.query).toHaveBeenCalledTimes(1); + expect(queryRef?.prompts.map((prompt) => prompt.text)).toEqual([ + "seed prompt", + "foreground prompt", + ]); await liveIterator.return?.(); await session.close(); diff --git a/packages/server/src/server/agent/providers/claude-agent.ts b/packages/server/src/server/agent/providers/claude-agent.ts index e773804c3..d192579d6 100644 --- a/packages/server/src/server/agent/providers/claude-agent.ts +++ b/packages/server/src/server/agent/providers/claude-agent.ts @@ -30,7 +30,6 @@ import { } from "./claude/tool-call-mapper.js"; import { coerceTaskNotificationHistoryRecordToSystemMessage, - isTaskNotificationUserContent, mapTaskNotificationSystemRecordToToolCall, mapTaskNotificationUserContentToToolCall, } from "./claude/task-notification-tool-call.js"; @@ -41,7 +40,7 @@ import { type ClaudeModelFamily, } from "./claude/model-catalog.js"; import { parsePartialJsonObject } from "./claude/partial-json.js"; -import { buildToolCallDisplayModel } from "../../../shared/tool-call-display.js"; +import { ClaudeSidechainTracker } from "./claude/sidechain-tracker.js"; import type { AgentCapabilityFlags, @@ -75,24 +74,6 @@ import { } from "../provider-launch-config.js"; import { getOrchestratorModeInstructions } from "../orchestrator-instructions.js"; -/* - * Routing invariant: - * While a foreground Claude turn is active, identifier-less assistant/stream/result - * events must stay attached to that foreground run unless we have explicit evidence - * that a distinct autonomous run has started. - * - * We previously allowed task_notification metadata to reserve autonomous ownership, - * then consumed that reservation on the next unbound chunk. In practice Claude often - * emits task_notification records and foreground tool-use stream chunks interleaved - * within the same turn. That let a foreground turn's terminal result get misrouted - * into an autonomous side run, which stranded the agent in "running" because the - * foreground stream never received turn_completed. - * - * The rule below is intentionally conservative: foreground turns get first claim on - * same-turn traffic, and autonomous wake reservations are only consumed once no - * foreground turn is active. This keeps task_notification advisory instead of letting - * it steal ownership from the active user turn. - */ const fsPromises = promises; const CLAUDE_SETTING_SOURCES: NonNullable = [ "user", @@ -101,56 +82,20 @@ const CLAUDE_SETTING_SOURCES: NonNullable = [ type TurnState = "idle" | "foreground" | "autonomous"; -type RunOwner = "foreground" | "autonomous"; - -type RunLifecycleState = - | "queued" - | "awaiting_response" - | "streaming" - | "finalizing" - | "completed" - | "interrupted" - | "error"; - -type SessionLifecycleState = "running" | "permission" | "error" | "idle"; - -type RoutingReason = - | "foreground" - | "task_id" - | "task_id_new" - | "parent_message_id" - | "message_id" - | "unbound_autonomous" - | "reserved_autonomous" - | "metadata" - | "ignored_unmatched"; - type EventIdentifiers = { taskId: string | null; parentMessageId: string | null; messageId: string | null; }; -type RunRoute = { - run: RunRecord | null; - reason: RoutingReason; - dispatchWithoutRun?: boolean; -}; - -type RunRecord = { - id: string; - owner: RunOwner; - queue: Pushable | null; - state: RunLifecycleState; - promptReplaySeen: boolean; - taskIds: Set; - parentMessageIds: Set; - messageIds: Set; -}; - type ForegroundTurnState = { - runId: string; + id: string; queue: Pushable; + hasVisibleActivity: boolean; +}; + +type AutonomousTurnState = { + id: string; }; type NormalizeClaudeRuntimeModelIdOptions = { @@ -646,14 +591,6 @@ function collectClaudeTextContentParts(content: unknown): string[] { return parts; } -function isClaudeInterruptScaffoldContent(content: unknown): boolean { - const parts = collectClaudeTextContentParts(content); - return ( - parts.length > 0 && - parts.every((part) => isClaudeInterruptPlaceholderText(part)) - ); -} - function isClaudeTranscriptNoiseContent(content: unknown): boolean { const parts = collectClaudeTextContentParts(content); return ( @@ -723,30 +660,6 @@ type ToolUseCacheEntry = { files?: { path: string; kind: string }[]; input?: AgentMetadata | null; }; - -type SubAgentActionEntry = { - index: number; - toolName: string; - summary?: string; -}; - -type SubAgentActivityState = { - subAgentType?: string; - description?: string; - actions: SubAgentActionEntry[]; - actionKeys: string[]; - nextActionIndex: number; - actionIndexByKey: Map; -}; - -type SubAgentActionCandidate = { - key: string; - toolName: string; - input: unknown; -}; - -const MAX_SUB_AGENT_LOG_ENTRIES = 200; -const MAX_SUB_AGENT_SUMMARY_CHARS = 160; function isMetadata(value: unknown): value is AgentMetadata { return typeof value === "object" && value !== null; } @@ -903,205 +816,6 @@ function resolvePermissionKind( return "tool"; } -const ACTIVE_RUN_STATES = new Set([ - "queued", - "awaiting_response", - "streaming", - "finalizing", -]); - -class RunTracker { - private readonly runs = new Map(); - private readonly runByTaskId = new Map(); - private readonly runByParentMessageId = new Map(); - private readonly runByMessageId = new Map(); - - createRun(input: { - id: string; - owner: RunOwner; - queue: Pushable | null; - promptReplaySeen?: boolean; - }): RunRecord { - const run: RunRecord = { - id: input.id, - owner: input.owner, - queue: input.queue, - state: "queued", - promptReplaySeen: input.promptReplaySeen ?? true, - taskIds: new Set(), - parentMessageIds: new Set(), - messageIds: new Set(), - }; - this.runs.set(run.id, run); - return run; - } - - getRun(runId: string): RunRecord | null { - return this.runs.get(runId) ?? null; - } - - getForegroundRun(): RunRecord | null { - for (const run of this.runs.values()) { - if (run.owner === "foreground" && this.isActive(run.state)) { - return run; - } - } - return null; - } - - listActiveRuns(owner?: RunOwner): RunRecord[] { - const runs: RunRecord[] = []; - for (const run of this.runs.values()) { - if (!this.isActive(run.state)) { - continue; - } - if (owner && run.owner !== owner) { - continue; - } - runs.push(run); - } - return runs; - } - - hasActiveRuns(owner?: RunOwner): boolean { - for (const run of this.runs.values()) { - if (!this.isActive(run.state)) { - continue; - } - if (owner && run.owner !== owner) { - continue; - } - return true; - } - return false; - } - - getLatestActiveRun(owner?: RunOwner): RunRecord | null { - let latest: RunRecord | null = null; - for (const run of this.runs.values()) { - if (!this.isActive(run.state)) { - continue; - } - if (owner && run.owner !== owner) { - continue; - } - latest = run; - } - return latest; - } - - isRunActive(run: RunRecord | null): boolean { - if (!run) { - return false; - } - return this.isActive(run.state); - } - - resolveByIdentifiers(identifiers: EventIdentifiers): RunRoute { - if (identifiers.taskId) { - const run = this.resolveMappedRun(this.runByTaskId, identifiers.taskId); - if (run) { - return { run, reason: "task_id" }; - } - } - if (identifiers.parentMessageId) { - const run = this.resolveMappedRun( - this.runByParentMessageId, - identifiers.parentMessageId - ); - if (run) { - return { run, reason: "parent_message_id" }; - } - } - if (identifiers.messageId) { - const run = this.resolveMappedRun( - this.runByMessageId, - identifiers.messageId - ); - if (run) { - return { run, reason: "message_id" }; - } - } - return { run: null, reason: "metadata" }; - } - - bindIdentifiers(run: RunRecord, identifiers: EventIdentifiers): void { - if (identifiers.taskId) { - run.taskIds.add(identifiers.taskId); - this.runByTaskId.set(identifiers.taskId, run.id); - } - if (identifiers.parentMessageId) { - run.parentMessageIds.add(identifiers.parentMessageId); - this.runByParentMessageId.set(identifiers.parentMessageId, run.id); - } - if (identifiers.messageId) { - run.messageIds.add(identifiers.messageId); - this.runByMessageId.set(identifiers.messageId, run.id); - } - } - - transition(run: RunRecord, nextState: RunLifecycleState): void { - run.state = nextState; - } - - complete(run: RunRecord, terminalState: "completed" | "interrupted" | "error"): void { - run.state = terminalState; - this.clearRunIndex(run); - } - - deriveLifecycle(pendingPermissionCount: number): SessionLifecycleState { - for (const run of this.runs.values()) { - if (this.isActive(run.state)) { - return "running"; - } - } - if (pendingPermissionCount > 0) { - return "permission"; - } - for (const run of this.runs.values()) { - if (run.state === "error") { - return "error"; - } - } - return "idle"; - } - - private resolveMappedRun( - mapping: Map, - identifier: string - ): RunRecord | null { - const runId = mapping.get(identifier); - if (!runId) { - return null; - } - const run = this.runs.get(runId); - if (!run || !this.isActive(run.state)) { - mapping.delete(identifier); - return null; - } - return run; - } - - private clearRunIndex(run: RunRecord): void { - for (const taskId of run.taskIds) { - this.runByTaskId.delete(taskId); - } - for (const parentMessageId of run.parentMessageIds) { - this.runByParentMessageId.delete(parentMessageId); - } - for (const messageId of run.messageIds) { - this.runByMessageId.delete(messageId); - } - run.taskIds.clear(); - run.parentMessageIds.clear(); - run.messageIds.clear(); - } - - private isActive(state: RunLifecycleState): boolean { - return ACTIVE_RUN_STATES.has(state); - } -} - type TimelineFragment = { kind: "assistant" | "reasoning"; text: string; @@ -1415,28 +1129,6 @@ class TimelineAssembler { } } -function isMetadataOnlySdkMessage(message: SDKMessage): boolean { - if (message.type === "system") { - return true; - } - if ( - message.type === "assistant" && - isClaudeTranscriptNoiseContent(message.message?.content) - ) { - return true; - } - if (message.type !== "user") { - return false; - } - if (isSyntheticUserEntry(message)) { - return true; - } - if (isClaudeInterruptScaffoldContent(message.message?.content)) { - return true; - } - return isTaskNotificationUserContent(message.message?.content); -} - function isSyntheticUserEntry(entry: unknown): boolean { if (!entry || typeof entry !== "object") { return false; @@ -1587,35 +1279,31 @@ class ClaudeAgentSession implements AgentSession { private toolUseInputBuffers = new Map(); private pendingPermissions = new Map(); private activeForegroundTurn: ForegroundTurnState | null = null; - private activeForegroundPrompt: SDKUserMessage | null = null; + private autonomousTurn: AutonomousTurnState | null = null; private liveEventQueue = new Pushable(); - private readonly runTracker = new RunTracker(); private readonly timelineAssembler = new TimelineAssembler(); + private readonly sidechainTracker = new ClaudeSidechainTracker({ + getToolInput: (toolUseId) => this.toolUseCache.get(toolUseId)?.input ?? null, + }); private persistedHistory: AgentTimelineItem[] = []; private historyPending = false; private historyOffsetSessionId: string | null = null; private historyReadOffsetBytes = 0; private historyLineFragment = ""; private turnState: TurnState = "idle"; - private preReplayMetadataSeen = false; - private preReplayDoneRecoveryUsed = false; - private pendingAutonomousWakeReservations = 0; - private nextRunOrdinal = 1; + private nextTurnOrdinal = 1; private cancelCurrentTurn: (() => void) | null = null; - private pendingInterruptPromise: Promise | null = null; private activeTurnPromise: Promise | null = null; private cachedRuntimeInfo: AgentRuntimeInfo | null = null; private lastOptionsModel: string | null = null; private selectableModelIds: Set | null = buildClaudeSelectableModelIds(); private selectableModelFamilyAliases: Map | null = buildClaudeModelFamilyAliases(); - private activeSidechains = new Map(); private compacting = false; private queryPumpPromise: Promise | null = null; private queryRestartNeeded = false; + private pendingInterruptAbort = false; private userMessageIds: string[] = []; - private readonly localUserMessageIds = new Set(); - private suppressLocalReplayActivity = false; private recentStderr = ""; private closed = false; @@ -1723,9 +1411,6 @@ class ClaudeAgentSession implements AgentSession { if (this.cancelCurrentTurn) { this.cancelCurrentTurn(); } - this.suppressLocalReplayActivity = false; - this.pendingAutonomousWakeReservations = 0; - this.preReplayDoneRecoveryUsed = false; const slashCommand = this.resolveSlashCommandInvocation(prompt); if (slashCommand?.commandName === REWIND_COMMAND_NAME) { @@ -1733,31 +1418,21 @@ class ClaudeAgentSession implements AgentSession { return; } - await this.awaitPendingInterruptPromise(); - if ( - this.turnState === "autonomous" && - this.runTracker.hasActiveRuns("autonomous") - ) { - await this.transitionAutonomousToForeground(); + if (this.autonomousTurn) { + this.completeAutonomousTurn(); } const sdkMessage = this.toSdkUserMessage(prompt); const queue = new Pushable(); - const run = this.createRun("foreground", queue); - this.runTracker.bindIdentifiers(run, { - taskId: null, - parentMessageId: null, - messageId: typeof sdkMessage.uuid === "string" ? sdkMessage.uuid : null, - }); const foregroundTurn: ForegroundTurnState = { - runId: run.id, + id: this.createTurnId("foreground"), queue, + hasVisibleActivity: false, }; this.activeForegroundTurn = foregroundTurn; - this.activeForegroundPrompt = sdkMessage; - this.preReplayMetadataSeen = false; this.transitionTurnState("foreground", "foreground stream started"); this.clearRecentStderr(); + queue.push({ type: "turn_started", provider: "claude" }); let finishedNaturally = false; let cancelIssued = false; @@ -1770,20 +1445,16 @@ class ClaudeAgentSession implements AgentSession { return; } cancelIssued = true; - if (this.activeForegroundTurn?.runId === run.id) { - this.activeForegroundTurn = null; - this.activeForegroundPrompt = null; - } if (this.cancelCurrentTurn === requestCancel) { this.cancelCurrentTurn = null; } this.rejectAllPendingPermissions(new Error("Permission request aborted")); - this.cancelRun(run, { + this.finishForegroundTurn({ type: "turn_canceled", provider: "claude", reason: "Interrupted", }); - this.pendingInterruptPromise = this.interruptActiveTurn().catch((error) => { + void this.interruptActiveTurn().catch((error) => { this.logger.warn({ err: error }, "Failed to interrupt during cancel"); }); }; @@ -1797,9 +1468,10 @@ class ClaudeAgentSession implements AgentSession { this.startQueryPump(); this.input.push(sdkMessage); } catch (error) { - this.failRun( - run, - error instanceof Error ? error.message : "Claude stream failed" + this.finishForegroundTurn( + this.buildTurnFailedEvent( + error instanceof Error ? error.message : "Claude stream failed" + ) ); finishedNaturally = true; } @@ -1843,16 +1515,8 @@ class ClaudeAgentSession implements AgentSession { return; } - const autonomousRuns = this.runTracker.listActiveRuns("autonomous"); - if (autonomousRuns.length > 0) { - this.flushPendingToolCalls(); - for (const run of autonomousRuns) { - this.emitRunEvent(run, { - type: "turn_canceled", - provider: "claude", - reason: "Interrupted", - }); - } + if (this.autonomousTurn) { + this.cancelAutonomousTurn("Interrupted"); } await this.interruptActiveTurn(); @@ -2026,13 +1690,12 @@ class ClaudeAgentSession implements AgentSession { this.cancelCurrentTurn?.(); this.activeForegroundTurn?.queue.end(); this.activeForegroundTurn = null; - this.activeForegroundPrompt = null; + this.autonomousTurn = null; this.cancelCurrentTurn = null; this.turnState = "idle"; - this.suppressLocalReplayActivity = false; - this.pendingAutonomousWakeReservations = 0; this.liveEventQueue.end(); this.activeTurnPromise = null; + this.sidechainTracker.clear(); this.input?.end(); await this.awaitWithTimeout(this.query?.interrupt?.(), "close query interrupt"); await this.awaitWithTimeout(this.query?.return?.(), "close query return"); @@ -2483,7 +2146,6 @@ class ClaudeAgentSession implements AgentSession { const messageId = randomUUID(); this.rememberUserMessageId(messageId); - this.localUserMessageIds.add(messageId); return { type: "user", @@ -2497,29 +2159,6 @@ class ClaudeAgentSession implements AgentSession { }; } - private async awaitPendingInterruptPromise(): Promise { - if (!this.pendingInterruptPromise) { - return; - } - await this.pendingInterruptPromise; - this.pendingInterruptPromise = null; - } - - private createRun( - owner: RunOwner, - queue: Pushable | null - ): RunRecord { - const runId = `${owner}-run-${this.nextRunOrdinal++}`; - const run = this.runTracker.createRun({ - id: runId, - owner, - queue, - promptReplaySeen: owner === "autonomous", - }); - this.logger.debug({ runId, owner, state: run.state }, "Created Claude run"); - return run; - } - private transitionTurnState(next: TurnState, reason: string): void { if (this.turnState === next) { return; @@ -2531,22 +2170,18 @@ class ClaudeAgentSession implements AgentSession { this.turnState = next; } - private transitionTurnStateFromActiveRuns(reason: string): void { - if (this.runTracker.hasActiveRuns("foreground")) { + private syncTurnState(reason: string): void { + if (this.activeForegroundTurn) { this.transitionTurnState("foreground", reason); return; } - if (this.runTracker.hasActiveRuns("autonomous")) { + if (this.autonomousTurn) { this.transitionTurnState("autonomous", reason); return; } this.transitionTurnState("idle", reason); } - private failRun(run: RunRecord, errorMessage: string): void { - this.emitRunEvent(run, this.buildTurnFailedEvent(errorMessage)); - } - private buildTurnFailedEvent( errorMessage: string ): Extract { @@ -2577,362 +2212,129 @@ class ClaudeAgentSession implements AgentSession { } private getRecentStderrDiagnostic(): string | undefined { - const text = this.recentStderr.trim(); - return text.length > 0 ? text : undefined; + return this.recentStderr.trim() || undefined; } - private cancelRun( - run: RunRecord, - event: Extract - ): void { - this.flushPendingToolCalls(); - this.emitRunEvent(run, event); + private createTurnId(owner: "foreground" | "autonomous"): string { + return `${owner}-turn-${this.nextTurnOrdinal++}`; } - private emitRunEvent(run: RunRecord, event: AgentStreamEvent): void { - if ( - event.type === "turn_started" || + private isTerminalTurnEvent(event: AgentStreamEvent): boolean { + return ( event.type === "turn_completed" || event.type === "turn_failed" || event.type === "turn_canceled" - ) { - this.logger.trace( - { - runId: run.id, - owner: run.owner, - runState: run.state, - eventType: event.type, - routedTo: run.owner === "foreground" && run.queue ? "foreground_queue" : "live_queue", - }, - "Claude run event emitted" - ); - } - if (run.owner === "foreground" && run.queue) { - run.queue.push(event); - if ( - event.type === "turn_completed" || - event.type === "turn_failed" || - event.type === "turn_canceled" - ) { - run.queue.end(); - } - } else { - this.liveEventQueue.push(event); - } - this.handleRunTerminalEvent(run, event); + ); } - private handleRunTerminalEvent(run: RunRecord, event: AgentStreamEvent): void { - if (event.type === "turn_completed") { - this.runTracker.complete(run, "completed"); - } else if (event.type === "turn_failed") { - this.runTracker.complete(run, "error"); - } else if (event.type === "turn_canceled") { - this.runTracker.complete(run, "interrupted"); - } else { + private shouldRecoverInterruptedQueryAbort( + error: unknown, + consecutiveRecoveries: number + ): boolean { + if (consecutiveRecoveries >= 3) { + return false; + } + const message = + typeof error === "string" + ? error + : error instanceof Error + ? `${error.message}\n${error.stack ?? ""}` + : JSON.stringify(error); + return message.toLowerCase().includes("request was aborted"); + } + + private finishForegroundTurn(event: Extract< + AgentStreamEvent, + { type: "turn_completed" | "turn_failed" | "turn_canceled" } + >): void { + if (event.type === "turn_failed" || event.type === "turn_canceled") { + this.flushPendingToolCalls(); + } + this.dispatchForegroundEvents([event]); + } + + private dispatchForegroundEvents(events: AgentStreamEvent[]): void { + const foregroundTurn = this.activeForegroundTurn; + if (!foregroundTurn) { + this.dispatchLiveEvents(events); return; } - if (this.activeForegroundTurn?.runId === run.id) { + let terminalSeen = false; + for (const event of events) { + foregroundTurn.queue.push(event); + terminalSeen ||= this.isTerminalTurnEvent(event); + } + + if (!terminalSeen) { + return; + } + + foregroundTurn.queue.end(); + if (this.activeForegroundTurn === foregroundTurn) { this.activeForegroundTurn = null; - this.activeForegroundPrompt = null; - this.preReplayMetadataSeen = false; - this.preReplayDoneRecoveryUsed = false; } - this.logger.trace( - { - runId: run.id, - owner: run.owner, - eventType: event.type, - runState: run.state, - hasActiveForegroundTurn: Boolean(this.activeForegroundTurn), - }, - "Claude run terminal event handled" - ); - this.transitionTurnStateFromActiveRuns(`run ${run.id} terminal`); + this.syncTurnState("foreground turn terminal"); } - private async transitionAutonomousToForeground(): Promise { - const autonomousRuns = this.runTracker.listActiveRuns("autonomous"); - if (autonomousRuns.length === 0) { - this.transitionTurnStateFromActiveRuns("no autonomous runs to transition"); + private dispatchLiveEvents(events: AgentStreamEvent[]): void { + let terminalSeen = false; + for (const event of events) { + this.liveEventQueue.push(event); + terminalSeen ||= this.isTerminalTurnEvent(event); + } + + if (terminalSeen && this.autonomousTurn) { + this.autonomousTurn = null; + this.syncTurnState("autonomous turn terminal"); + } + } + + private startAutonomousTurn(): void { + if (this.autonomousTurn) { return; } - - this.logger.debug( - { runIds: autonomousRuns.map((run) => run.id) }, - "Transitioning autonomous runs to foreground ownership" - ); - this.flushPendingToolCalls(); - for (const run of autonomousRuns) { - this.emitRunEvent(run, { - type: "turn_canceled", - provider: "claude", - reason: "Interrupted by foreground prompt", - }); - } - this.pendingInterruptPromise = this.interruptActiveTurn().catch((error) => { - this.logger.warn( - { err: error }, - "Failed to interrupt autonomous run during foreground transition" - ); - }); - await this.awaitPendingInterruptPromise(); - this.transitionTurnStateFromActiveRuns("autonomous interrupted for foreground"); - } - - private routeMessage( - normalized: { - message: SDKMessage; - identifiers: EventIdentifiers; - metadataOnly: boolean; - } - ): RunRoute { - if (normalized.metadataOnly) { - if ( - (normalized.message.type === "user" && - isTaskNotificationUserContent(normalized.message.message?.content)) || - (normalized.message.type === "system" && - normalized.message.subtype === "task_notification") - ) { - this.reserveAutonomousWake("task_notification"); - } - this.notePreReplayMetadata(normalized.message); - return { run: null, reason: "metadata" }; - } - - const hasIdentifiers = Boolean( - normalized.identifiers.taskId || - normalized.identifiers.parentMessageId || - normalized.identifiers.messageId - ); - - const byIdentifiers = this.runTracker.resolveByIdentifiers(normalized.identifiers); - if (byIdentifiers.run) { - return byIdentifiers; - } - - const foregroundRun = this.activeForegroundTurn - ? this.runTracker.getRun(this.activeForegroundTurn.runId) - : null; - - // A previously unseen task_id during foreground ownership is deterministic - // evidence of a distinct autonomous wake/run, not foreground response text. - if ( - this.turnState === "foreground" && - foregroundRun && - normalized.identifiers.taskId - ) { - const incomingTaskId = normalized.identifiers.taskId; - // Foreground must claim its first task_id; otherwise early foreground - // result events can be misrouted to autonomous fallback runs. - if (foregroundRun.taskIds.size === 0) { - if (foregroundRun.state !== "finalizing") { - return { run: foregroundRun, reason: "foreground" }; - } - } else if (foregroundRun.taskIds.has(incomingTaskId)) { - return { run: foregroundRun, reason: "foreground" }; - } - - const autonomousRun = this.createRun("autonomous", null); - this.emitRunEvent(autonomousRun, { type: "turn_started", provider: "claude" }); - return { run: autonomousRun, reason: "task_id_new" }; - } - - if ( - this.turnState === "foreground" && - foregroundRun && - this.shouldPreferForegroundRun({ - run: foregroundRun, - message: normalized.message, - }) - ) { - return { run: foregroundRun, reason: "foreground" }; - } - - if ( - this.pendingAutonomousWakeReservations > 0 && - !normalized.identifiers.taskId && - !normalized.identifiers.parentMessageId && - !normalized.identifiers.messageId - ) { - const reservedAutonomousRun = this.claimOrCreateAutonomousRun( - "reservation_unbound" - ); - return { - run: reservedAutonomousRun, - reason: "reserved_autonomous", - }; - } - - if (!hasIdentifiers) { - const activeAutonomousRun = this.runTracker.getLatestActiveRun("autonomous"); - if (activeAutonomousRun) { - return { run: activeAutonomousRun, reason: "unbound_autonomous" }; - } - if ( - !foregroundRun && - (normalized.message.type === "assistant" || - normalized.message.type === "stream_event" || - normalized.message.type === "result" || - normalized.message.type === "tool_progress") - ) { - const autonomousRun = this.claimOrCreateAutonomousRun("unbound_implicit"); - return { run: autonomousRun, reason: "unbound_autonomous" }; - } - } - - if (this.pendingAutonomousWakeReservations > 0) { - const reservedAutonomousRun = this.claimOrCreateAutonomousRun( - "reservation_fallback" - ); - return { run: reservedAutonomousRun, reason: "reserved_autonomous" }; - } - - this.logger.debug( - { - messageType: normalized.message.type, - hasIdentifiers, - taskId: normalized.identifiers.taskId, - parentMessageId: normalized.identifiers.parentMessageId, - messageId: normalized.identifiers.messageId, - turnState: this.turnState, - pendingAutonomousWakeReservations: this.pendingAutonomousWakeReservations, - }, - "Ignoring unmatched Claude SDK message without explicit run start signal" - ); - return { - run: null, - reason: "ignored_unmatched", - dispatchWithoutRun: false, + this.autonomousTurn = { + id: this.createTurnId("autonomous"), }; + this.liveEventQueue.push({ type: "turn_started", provider: "claude" }); + this.syncTurnState("autonomous turn started"); } - private shouldPreferForegroundRun(input: { - run: RunRecord; - message: SDKMessage; - }): boolean { - const { run, message } = input; - if ( - run.state === "completed" || - run.state === "interrupted" || - run.state === "error" - ) { - return false; - } - - // Before prompt replay is observed, prefer foreground by default so the - // first turn cannot be stranded in autonomous fallback. If metadata churn - // was observed pre-replay, stay conservative and wait for replay. - if (!run.promptReplaySeen) { - if (this.isToolUseBoundaryStreamEvent(input.message)) { - return true; - } - // Keep pre-replay result events with the foreground run so stale result - // bursts cannot consume autonomous wake reservations. - if (message.type === "result") { - return true; - } - if ( - message.type === "assistant" || - message.type === "stream_event" || - message.type === "tool_progress" - ) { - return !this.preReplayMetadataSeen; - } - return true; - } - - if ( - run.state === "finalizing" && - (message.type === "assistant" || message.type === "stream_event") - ) { - return false; - } - - return true; - } - - private isToolUseBoundaryStreamEvent(message: SDKMessage): boolean { - if (message.type !== "stream_event") { - return false; - } - const event = (message as unknown as { event?: Record }).event; - if (!event || event.type !== "message_delta") { - return false; - } - const delta = - "delta" in event && event.delta && typeof event.delta === "object" - ? (event.delta as Record) - : null; - return delta?.stop_reason === "tool_use"; - } - - private notePreReplayMetadata(message: SDKMessage): void { - if (this.turnState !== "foreground") { + private completeAutonomousTurn(): void { + if (!this.autonomousTurn) { return; } - const foregroundRun = this.activeForegroundTurn - ? this.runTracker.getRun(this.activeForegroundTurn.runId) - : null; - if (!foregroundRun || foregroundRun.promptReplaySeen) { - return; - } - // Most system metadata (init/hook callbacks/etc.) can precede the first prompt - // replay for a legitimate foreground run. Treating all of it as churn strands - // one-shot helper runs. task_notification is the exception: it represents - // background agent activity and should suppress pre-replay foreground routing. - if ( - message.type === "system" && - message.subtype !== "task_notification" - ) { - return; - } - this.preReplayMetadataSeen = true; + this.autonomousTurn = null; + this.liveEventQueue.push({ type: "turn_completed", provider: "claude" }); + this.syncTurnState("autonomous turn completed"); } - private reserveAutonomousWake(reason: string): void { - this.pendingAutonomousWakeReservations += 1; - this.logger.debug( - { - reason, - pendingAutonomousWakeReservations: this.pendingAutonomousWakeReservations, - }, - "Reserved autonomous wake" - ); + private cancelAutonomousTurn(reason: string): void { + if (!this.autonomousTurn) { + return; + } + this.flushPendingToolCalls(); + this.autonomousTurn = null; + this.liveEventQueue.push({ + type: "turn_canceled", + provider: "claude", + reason, + }); + this.syncTurnState("autonomous turn canceled"); } - private claimOrCreateAutonomousRun(reason: string): RunRecord { - const existing = this.runTracker.getLatestActiveRun("autonomous"); - if (existing) { - if (this.pendingAutonomousWakeReservations > 0) { - this.pendingAutonomousWakeReservations -= 1; - } - this.logger.debug( - { - reason, - runId: existing.id, - pendingAutonomousWakeReservations: this.pendingAutonomousWakeReservations, - }, - "Claimed autonomous wake reservation on existing run" - ); - return existing; + private failActiveTurns(errorMessage: string): void { + const failure = this.buildTurnFailedEvent(errorMessage); + if (this.activeForegroundTurn) { + this.flushPendingToolCalls(); + this.dispatchForegroundEvents([failure]); + return; } - - const run = this.createRun("autonomous", null); - this.emitRunEvent(run, { type: "turn_started", provider: "claude" }); - if (this.pendingAutonomousWakeReservations > 0) { - this.pendingAutonomousWakeReservations -= 1; + if (this.autonomousTurn) { + this.flushPendingToolCalls(); + this.dispatchLiveEvents([failure]); } - this.logger.debug( - { - reason, - runId: run.id, - pendingAutonomousWakeReservations: this.pendingAutonomousWakeReservations, - }, - "Claimed autonomous wake reservation with new run" - ); - return run; } private startQueryPump(): void { @@ -2953,224 +2355,138 @@ class ClaudeAgentSession implements AgentSession { } private async runQueryPump(): Promise { - while (!this.closed) { - if (!this.claudeSessionId && !this.activeForegroundTurn && !this.query) { - await this.waitForLiveHistoryPoll(); - continue; - } + let activeQuery: Query; + try { + activeQuery = await this.ensureQuery(); + } catch (error) { + this.logger.trace({ err: error }, "Failed to initialize Claude query pump"); + this.failActiveTurns(error instanceof Error ? error.message : "Claude stream failed"); + return; + } - let q: Query; - try { - q = await this.ensureQuery(); - } catch (error) { - this.logger.trace({ err: error }, "Failed to initialize Claude query pump"); - await this.waitForLiveHistoryPoll(); - continue; - } - - let next: IteratorResult; - try { - next = await q.next(); - this.logger.trace( - { claudeSessionId: this.claudeSessionId, next }, - "Claude query pump raw next()" - ); - } catch (error) { - if (this.query !== q) { - this.logger.trace( - { err: error, staleQuery: true }, - "Ignoring Claude query pump next() failure from replaced query" - ); - await this.awaitWithTimeout( - q.return?.(), - "query pump return after stale failure" - ); - continue; - } - this.logger.trace({ err: error }, "Claude query pump next() failed"); - for (const run of this.runTracker.listActiveRuns()) { - this.failRun( - run, - error instanceof Error ? error.message : "Claude stream failed" - ); - } - this.input?.end(); - await this.awaitWithTimeout(q.return?.(), "query pump return after failure"); - if (this.query === q) { - this.query = null; - this.input = null; - } - await this.waitForLiveHistoryPoll(); - continue; - } - - if (next.done) { - if (this.query !== q) { - this.logger.trace( - { - claudeSessionId: this.claudeSessionId, - activeRunCount: this.runTracker.listActiveRuns().length, - staleQuery: true, - }, - "Ignoring replaced Claude query pump completion" - ); - await this.awaitWithTimeout( - q.return?.(), - "query pump return on stale done" - ); - continue; - } - this.logger.trace( - { - claudeSessionId: this.claudeSessionId, - activeRunCount: this.runTracker.listActiveRuns().length, - }, - "Claude query pump next() returned done" - ); - this.input?.end(); - await this.awaitWithTimeout(q.return?.(), "query pump return on done"); - if (this.query === q) { - this.query = null; - this.input = null; - } - const activeRuns = this.runTracker.listActiveRuns(); - if (this.shouldRecoverFromPreReplayQueryDone(activeRuns)) { - this.logger.warn( - { - claudeSessionId: this.claudeSessionId, - activeRunCount: activeRuns.length, - preReplayMetadataSeen: this.preReplayMetadataSeen, - }, - "Claude query ended before prompt replay after interrupt scaffold; retrying fresh query" - ); - try { - await this.ensureQuery(); - if (this.input && this.activeForegroundPrompt) { - this.input.push(this.activeForegroundPrompt); + let consecutiveInterruptAbortRecoveries = 0; + try { + while (!this.closed && this.query === activeQuery) { + try { + for await (const message of activeQuery) { + consecutiveInterruptAbortRecoveries = 0; + if (await this.handleMissingResumedConversation(message, activeQuery)) { + return; } - } catch (error) { - for (const run of activeRuns) { - this.failRun( - run, - error instanceof Error ? error.message : "Claude stream failed" - ); - } - await this.waitForLiveHistoryPoll(); + this.routeSdkMessageFromPump(message); } - continue; - } - if (activeRuns.length > 0) { - for (const run of activeRuns) { - this.failRun(run, "Claude stream ended before terminal result"); + if (!this.closed && this.query === activeQuery) { + this.failActiveTurns("Claude stream ended before terminal result"); } + return; + } catch (error) { + if ( + !this.closed && + this.query === activeQuery && + this.shouldRecoverInterruptedQueryAbort( + error, + consecutiveInterruptAbortRecoveries + ) + ) { + consecutiveInterruptAbortRecoveries += 1; + this.logger.debug( + { recoveries: consecutiveInterruptAbortRecoveries }, + "Recovering Claude query pump after interrupt abort" + ); + continue; + } + if (!this.closed && this.query === activeQuery) { + this.failActiveTurns(error instanceof Error ? error.message : "Claude stream failed"); + } + return; } - await this.waitForLiveHistoryPoll(); - continue; } - - const sdkMessage = next.value; - if (!sdkMessage) { - continue; - } - - if (this.query !== q) { - this.logger.trace( - { - claudeSessionId: this.claudeSessionId, - messageType: sdkMessage.type, - staleQuery: true, - }, - "Ignoring Claude SDK message from replaced query" - ); - await this.awaitWithTimeout( - q.return?.(), - "query pump return on stale message" - ); - continue; - } - - if (await this.handleMissingResumedConversation(sdkMessage, q)) { - continue; - } - - try { - this.routeSdkMessageFromPump(sdkMessage); - } catch (error) { - this.logger.trace({ err: error }, "Failed to route Claude SDK message from query pump"); + } finally { + if (this.query === activeQuery) { + this.query = null; + this.input = null; } } } private routeSdkMessageFromPump(message: SDKMessage): void { - if (this.shouldSuppressLocalReplayActivity(message)) { + const routeToForeground = Boolean(this.activeForegroundTurn); + const assistantishMessage = + message.type === "assistant" || + message.type === "stream_event" || + message.type === "tool_progress"; + + if (!routeToForeground && assistantishMessage) { + this.startAutonomousTurn(); + } + if (!routeToForeground && !this.autonomousTurn && message.type === "result") { return; } + const turnId = this.activeForegroundTurn?.id ?? this.autonomousTurn?.id ?? null; const identifiers = readEventIdentifiers(message); - const metadataOnly = isMetadataOnlySdkMessage(message); - const route = this.routeMessage({ - message, - identifiers, - metadataOnly, - }); - const suppressTerminalEvents = this.shouldSuppressReplayResultTerminal({ - run: route.run, - message, - }); + this.logger.trace( { claudeSessionId: this.claudeSessionId, messageType: message.type, - routeReason: route.reason, - runId: route.run?.id ?? null, - runOwner: route.run?.owner ?? null, - suppressTerminalEvents, - metadataOnly, + routedTo: routeToForeground ? "foreground_queue" : "live_queue", + turnId, }, "Claude query pump routed SDK message" ); - if (route.run) { - this.transitionTurnStateFromActiveRuns(`routed via ${route.reason}`); - this.runTracker.bindIdentifiers(route.run, identifiers); - if (!suppressTerminalEvents) { - this.updateRunLifecycleForMessage(route.run, message, identifiers); - } - } const messageEvents = this.translateMessageToEvents(message, { suppressAssistantText: true, suppressReasoning: true, - suppressTerminalEvents, }); - const assistantTimelineItems = this.timelineAssembler.consume({ - message, - runId: route.run?.id ?? null, - messageIdHint: identifiers.messageId, - }); - const assistantTimelineEvents: AgentStreamEvent[] = assistantTimelineItems.map( - (item) => ({ + const assistantTimelineEvents = this.timelineAssembler + .consume({ + message, + runId: turnId, + messageIdHint: identifiers.messageId, + }) + .map((item) => ({ type: "timeline", item, provider: "claude", - }) - ); + }) satisfies AgentStreamEvent); const events = [...messageEvents, ...assistantTimelineEvents]; if (events.length === 0) { return; } - if (!route.run) { - if (route.dispatchWithoutRun === false) { - return; - } - this.dispatchMetadataEvents(events); + if ( + this.pendingInterruptAbort && + message.type === "result" && + events.some( + (event) => event.type === "turn_completed" || event.type === "turn_failed" + ) && + (!this.activeForegroundTurn || !this.activeForegroundTurn.hasVisibleActivity) + ) { + this.pendingInterruptAbort = false; + this.logger.debug("Suppressing stale Claude interrupt terminal result"); return; } - - for (const event of events) { - this.emitRunEvent(route.run, event); + if ( + this.activeForegroundTurn && + events.some( + (event) => + event.type === "timeline" || + event.type === "permission_requested" || + event.type === "permission_resolved" + ) + ) { + this.activeForegroundTurn.hasVisibleActivity = true; + this.pendingInterruptAbort = false; } + + if (routeToForeground) { + this.dispatchForegroundEvents(events); + return; + } + this.dispatchLiveEvents(events); } private async handleMissingResumedConversation( @@ -3190,10 +2506,7 @@ class ClaudeAgentSession implements AgentSession { "Claude resumed session no longer exists; invalidating persisted session" ); - for (const run of this.runTracker.listActiveRuns()) { - this.failRun(run, staleResumeError); - } - this.transitionTurnStateFromActiveRuns("missing resumed conversation"); + this.failActiveTurns(staleResumeError); this.input?.end(); await this.awaitWithTimeout( query.return?.(), @@ -3212,520 +2525,41 @@ class ClaudeAgentSession implements AgentSession { this.historyLineFragment = ""; this.cachedRuntimeInfo = null; this.queryRestartNeeded = false; + this.autonomousTurn = null; + this.activeForegroundTurn = null; + this.syncTurnState("missing resumed conversation"); return true; } - private shouldSuppressReplayResultTerminal(input: { - run: RunRecord | null; - message: SDKMessage; - }): boolean { - const { run, message } = input; - if (!run || run.owner !== "foreground" || message.type !== "result") { - return false; - } - if (run.promptReplaySeen) { - return false; - } - if (run.state === "streaming" || run.state === "finalizing") { - return false; - } - - const resultSubtype = - "subtype" in message && typeof message.subtype === "string" - ? message.subtype - : null; - - // Pre-replay success results are stale in practice (leftover from an - // earlier query segment) and must not end the current foreground run. - if (resultSubtype === "success") { - this.logger.trace( - { - runId: run.id, - runOwner: run.owner, - runState: run.state, - promptReplaySeen: run.promptReplaySeen, - resultSubtype, - }, - "Suppressing pre-replay foreground success result terminal event" - ); - return true; - } - - // For non-success results, keep the metadata-churn guard to avoid - // suppressing legitimate hard failures. - return this.preReplayMetadataSeen; - } - - private dispatchMetadataEvents(events: AgentStreamEvent[]): void { - for (const event of events) { - this.pushEvent(event); - } - } - - private shouldRecoverFromPreReplayQueryDone(activeRuns: RunRecord[]): boolean { - if ( - this.preReplayDoneRecoveryUsed || - !this.preReplayMetadataSeen || - !this.activeForegroundTurn || - activeRuns.length !== 1 - ) { - return false; - } - - const foregroundRun = this.runTracker.getRun(this.activeForegroundTurn.runId); - if (!foregroundRun || activeRuns[0]?.id !== foregroundRun.id) { - return false; - } - if (foregroundRun.owner !== "foreground" || foregroundRun.promptReplaySeen) { - return false; - } - - this.preReplayDoneRecoveryUsed = true; - return true; - } - - private updateRunLifecycleForMessage( - run: RunRecord, - message: SDKMessage, - identifiers: EventIdentifiers - ): void { - const previousState = run.state; - if ( - message.type === "user" && - identifiers.messageId && - run.messageIds.has(identifiers.messageId) - ) { - run.promptReplaySeen = true; - this.preReplayMetadataSeen = false; - this.preReplayDoneRecoveryUsed = false; - } - - if (run.state === "queued") { - this.runTracker.transition(run, "awaiting_response"); - } - - if ( - message.type === "assistant" || - message.type === "stream_event" || - message.type === "tool_progress" - ) { - this.runTracker.transition(run, "streaming"); - return; - } - - if (message.type === "result") { - this.runTracker.transition(run, "finalizing"); - } else { - return; - } - - if (run.state !== previousState) { - this.logger.trace( - { - runId: run.id, - owner: run.owner, - messageType: message.type, - previousState, - nextState: run.state, - taskId: identifiers.taskId, - parentMessageId: identifiers.parentMessageId, - messageId: identifiers.messageId, - }, - "Updated Claude run lifecycle from SDK message" - ); - } - } - - private shouldSuppressLocalReplayActivity(message: SDKMessage): boolean { - const replayUuid = this.getLocalReplayUserMessageUuid(message); - if (replayUuid) { - // Don't suppress the echo of the current foreground turn's prompt — - // updateRunLifecycleForMessage needs it to set promptReplaySeen. - if (this.activeForegroundTurn) { - const foregroundRun = this.runTracker.getRun( - this.activeForegroundTurn.runId - ); - if (foregroundRun?.messageIds.has(replayUuid)) { - this.suppressLocalReplayActivity = false; - return false; - } - } - this.suppressLocalReplayActivity = true; - this.logger.debug( - { uuid: replayUuid }, - "Suppressing local replay user message from live pump" - ); - return true; - } - - if (!this.suppressLocalReplayActivity) { - return false; - } - - // Suppress only replay scaffolding. Do not suppress autonomous - // assistant/result events; otherwise task-notification replies can be dropped. - if (replayUuid) { - return true; - } - - if (message.type === "system") { - return true; - } - - const identifiers = readEventIdentifiers(message); - const hasIdentifiers = Boolean( - identifiers.taskId || identifiers.parentMessageId || identifiers.messageId - ); - - if (message.type !== "user" && !hasIdentifiers) { - if (this.pendingAutonomousWakeReservations > 0) { - this.suppressLocalReplayActivity = false; - return false; - } - return true; - } - - if (message.type === "user") { - this.suppressLocalReplayActivity = false; - return false; - } - - this.suppressLocalReplayActivity = false; - return false; - } - - private getLocalReplayUserMessageUuid(message: SDKMessage): string | null { - if (message.type !== "user") { - return null; - } - const uuid = readTrimmedString( - (message as unknown as { uuid?: unknown }).uuid - ); - if (!uuid) { - return null; - } - return this.localUserMessageIds.has(uuid) ? uuid : null; - } - private async interruptActiveTurn(): Promise { const queryToInterrupt = this.query; if (!queryToInterrupt || typeof queryToInterrupt.interrupt !== "function") { this.logger.trace("interruptActiveTurn: no query to interrupt"); return; } + this.pendingInterruptAbort = true; try { - this.logger.trace("interruptActiveTurn: calling query.interrupt()..."); await this.awaitWithTimeout( queryToInterrupt.interrupt(), "interruptActiveTurn query.interrupt()" ); - this.input?.end(); - this.logger.trace("interruptActiveTurn: calling query.return()..."); - await this.awaitWithTimeout( - queryToInterrupt.return?.(), - "interruptActiveTurn query.return()" - ); - this.query = null; - this.input = null; - this.queryRestartNeeded = false; } catch (error) { this.logger.warn({ err: error }, "Failed to interrupt active turn"); - this.input?.end(); - if (this.query === queryToInterrupt) { - this.query = null; - this.input = null; - } - // Try to force-close the iterator to unblock the pump's q.next() call. - this.awaitWithTimeout( - queryToInterrupt.return?.(), - "interruptActiveTurn force return after failure" - ).catch(() => {}); - // Disown the current pump and start a fresh one immediately so - // autonomous wakes are not lost while waiting for the next user turn. - this.queryPumpPromise = null; - this.startQueryPump(); } } - private handleSidechainMessage( - message: SDKMessage, - parentToolUseId: string - ): AgentStreamEvent[] { - const state = - this.activeSidechains.get(parentToolUseId) ?? - ({ - actions: [], - actionKeys: [], - nextActionIndex: 1, - actionIndexByKey: new Map(), - } satisfies SubAgentActivityState); - this.activeSidechains.set(parentToolUseId, state); - - const contextUpdated = this.updateSubAgentContextFromTaskInput( - state, - parentToolUseId - ); - const actionCandidates = this.extractSubAgentActionCandidates(message); - let actionUpdated = false; - for (const action of actionCandidates) { - if (this.appendSubAgentAction(state, action)) { - actionUpdated = true; - } - } - - if (!contextUpdated && !actionUpdated) { - return []; - } - - const toolCall = mapClaudeRunningToolCall({ - name: "Task", - callId: parentToolUseId, - input: null, - output: null, - }); - if (!toolCall) { - return []; - } - - const detail: Extract["detail"] = { - type: "sub_agent", - ...(state.subAgentType ? { subAgentType: state.subAgentType } : {}), - ...(state.description ? { description: state.description } : {}), - log: state.actions - .map((action) => - action.summary - ? `[${action.toolName}] ${action.summary}` - : `[${action.toolName}]` - ) - .join("\n"), - actions: state.actions.map((action) => ({ - index: action.index, - toolName: action.toolName, - ...(action.summary ? { summary: action.summary } : {}), - })), - }; - - return [ - { - type: "timeline", - item: { - ...toolCall, - detail, - }, - provider: "claude", - }, - ]; - } - - private updateSubAgentContextFromTaskInput( - state: SubAgentActivityState, - parentToolUseId: string - ): boolean { - const taskInput = this.toolUseCache.get(parentToolUseId)?.input; - const nextSubAgentType = this.normalizeSubAgentText(taskInput?.subagent_type); - const nextDescription = this.normalizeSubAgentText(taskInput?.description); - - let changed = false; - if (nextSubAgentType && nextSubAgentType !== state.subAgentType) { - state.subAgentType = nextSubAgentType; - changed = true; - } - if (nextDescription && nextDescription !== state.description) { - state.description = nextDescription; - changed = true; - } - return changed; - } - - private normalizeSubAgentText(value: unknown): string | undefined { - const normalized = readTrimmedString(value)?.replace(/\s+/g, " "); - if (!normalized) { - return undefined; - } - if (normalized.length <= MAX_SUB_AGENT_SUMMARY_CHARS) { - return normalized; - } - return `${normalized.slice(0, MAX_SUB_AGENT_SUMMARY_CHARS)}...`; - } - - private extractSubAgentActionCandidates( - message: SDKMessage - ): SubAgentActionCandidate[] { - if (message.type === "assistant") { - const content = message.message?.content; - if (!Array.isArray(content)) { - return []; - } - const actions: SubAgentActionCandidate[] = []; - for (const block of content) { - if ( - !isClaudeContentChunk(block) || - !( - block.type === "tool_use" || - block.type === "mcp_tool_use" || - block.type === "server_tool_use" - ) || - typeof block.name !== "string" - ) { - continue; - } - const key = - readTrimmedString(block.id) ?? - `assistant:${block.name}:${actions.length}`; - actions.push({ - key, - toolName: block.name, - input: block.input ?? null, - }); - } - return actions; - } - - if (message.type === "stream_event") { - const event = message.event; - if (event.type !== "content_block_start") { - return []; - } - const block = isClaudeContentChunk(event.content_block) - ? event.content_block - : null; - if ( - !block || - !( - block.type === "tool_use" || - block.type === "mcp_tool_use" || - block.type === "server_tool_use" - ) || - typeof block.name !== "string" - ) { - return []; - } - const key = - readTrimmedString(block.id) ?? - `stream:${block.name}:${typeof event.index === "number" ? event.index : 0}`; - return [ - { - key, - toolName: block.name, - input: block.input ?? null, - }, - ]; - } - - if (message.type === "tool_progress") { - const toolName = readTrimmedString(message.tool_name); - if (!toolName) { - return []; - } - const key = - readTrimmedString(message.tool_use_id) ?? `progress:${toolName}`; - return [{ key, toolName, input: null }]; - } - - return []; - } - - private appendSubAgentAction( - state: SubAgentActivityState, - candidate: SubAgentActionCandidate - ): boolean { - const normalizedToolName = readTrimmedString(candidate.toolName); - if (!normalizedToolName) { - return false; - } - - const summary = this.deriveSubAgentActionSummary( - normalizedToolName, - candidate.input - ); - const existingIndex = state.actionIndexByKey.get(candidate.key); - - if (existingIndex !== undefined) { - const existing = state.actions[existingIndex]; - if (!existing) { - return false; - } - const nextSummary = existing.summary ?? summary; - const unchanged = - existing.toolName === normalizedToolName && - existing.summary === nextSummary; - if (unchanged) { - return false; - } - state.actions[existingIndex] = { - ...existing, - toolName: normalizedToolName, - ...(nextSummary ? { summary: nextSummary } : {}), - }; - return true; - } - - const nextEntry: SubAgentActionEntry = { - index: state.nextActionIndex, - toolName: normalizedToolName, - ...(summary ? { summary } : {}), - }; - state.nextActionIndex += 1; - state.actions.push(nextEntry); - state.actionKeys.push(candidate.key); - this.trimSubAgentTail(state); - this.rebuildSubAgentActionIndex(state); - return true; - } - - private trimSubAgentTail(state: SubAgentActivityState): void { - while (state.actions.length > MAX_SUB_AGENT_LOG_ENTRIES) { - state.actions.shift(); - state.actionKeys.shift(); - } - } - - private rebuildSubAgentActionIndex(state: SubAgentActivityState): void { - state.actionIndexByKey.clear(); - for (let index = 0; index < state.actionKeys.length; index += 1) { - const key = state.actionKeys[index]; - if (key) { - state.actionIndexByKey.set(key, index); - } - } - } - - private deriveSubAgentActionSummary( - toolName: string, - input: unknown - ): string | undefined { - const runningToolCall = mapClaudeRunningToolCall({ - name: toolName, - callId: `sub-agent-summary-${toolName}`, - input, - output: null, - }); - if (!runningToolCall) { - return undefined; - } - const display = buildToolCallDisplayModel({ - name: runningToolCall.name, - status: runningToolCall.status, - error: runningToolCall.error, - detail: runningToolCall.detail, - metadata: runningToolCall.metadata, - }); - return this.normalizeSubAgentText(display.summary); - } - private translateMessageToEvents( message: SDKMessage, options?: { suppressAssistantText?: boolean; suppressReasoning?: boolean; - suppressTerminalEvents?: boolean; } ): AgentStreamEvent[] { const parentToolUseId = "parent_tool_use_id" in message ? (message as { parent_tool_use_id: string | null }).parent_tool_use_id : null; if (parentToolUseId) { - return this.handleSidechainMessage(message, parentToolUseId); + return this.sidechainTracker.handleMessage(message, parentToolUseId); } const events: AgentStreamEvent[] = []; @@ -3768,8 +2602,7 @@ class ClaudeAgentSession implements AgentSession { item: { type: "compaction", status: "completed", - trigger: - compactMetadata?.trigger === "manual" ? "manual" : "auto", + trigger: compactMetadata?.trigger === "manual" ? "manual" : "auto", preTokens: compactMetadata?.preTokens, }, provider: "claude", @@ -3814,7 +2647,6 @@ class ClaudeAgentSession implements AgentSession { break; } if (typeof content === "string" && content.length > 0) { - // String content from user messages (e.g., local command output) if (!isClaudeTranscriptNoiseText(content)) { events.push({ type: "timeline", @@ -3827,9 +2659,6 @@ class ClaudeAgentSession implements AgentSession { }); } } else if (Array.isArray(content)) { - // User SDK entries with array content (e.g. interrupt messages, replayed JSONL entries) - // must declare textMessageType so mapBlocksToTimeline emits user_message, not assistant_message. - // Without this, user text during interrupts appears as assistant output in paseo logs and the app. const timelineItems = this.mapBlocksToTimeline(content, { textMessageType: "user_message", }); @@ -3868,9 +2697,6 @@ class ClaudeAgentSession implements AgentSession { break; } case "result": { - if (options?.suppressTerminalEvents) { - break; - } const usage = this.convertUsage(message); if (message.subtype === "success") { events.push({ type: "turn_completed", provider: "claude", usage }); @@ -3949,7 +2775,6 @@ class ClaudeAgentSession implements AgentSession { let threadStartedSessionId: string | null = null; if (existingSessionId === null) { - // First time setting session ID (empty → filled) - this is expected this.claudeSessionId = newSessionId; threadStartedSessionId = newSessionId; this.logger.debug( @@ -3957,14 +2782,11 @@ class ClaudeAgentSession implements AgentSession { "Claude session ID set for the first time" ); } else if (existingSessionId === newSessionId) { - // Same session ID - no-op, but log for visibility this.logger.debug( { sessionId: newSessionId }, "Claude session ID unchanged (same value)" ); } else { - // CRITICAL: Session ID is being overwritten with a different value - // This should NEVER happen and indicates a serious bug throw new Error( `CRITICAL: Claude session ID overwrite detected! ` + `Existing: ${existingSessionId}, New: ${newSessionId}. ` + @@ -3974,7 +2796,6 @@ class ClaudeAgentSession implements AgentSession { this.availableModes = DEFAULT_MODES; this.currentMode = message.permissionMode; this.persistence = null; - // Capture actual model from SDK init message (not just the configured model) if (message.model) { const normalizedModel = normalizeClaudeRuntimeModelId({ runtimeModelId: message.model, @@ -3988,7 +2809,6 @@ class ClaudeAgentSession implements AgentSession { "Captured model from SDK init" ); this.lastOptionsModel = normalizedModel; - // Invalidate cached runtime info so it picks up the new model this.cachedRuntimeInfo = null; } return threadStartedSessionId; @@ -4122,7 +2942,7 @@ class ClaudeAgentSession implements AgentSession { } } this.toolUseCache.clear(); - this.activeSidechains.clear(); + this.sidechainTracker.clear(); } private pushToolCall( @@ -4142,17 +2962,8 @@ class ClaudeAgentSession implements AgentSession { private pushEvent(event: AgentStreamEvent) { const foregroundTurn = this.activeForegroundTurn; if (foregroundTurn) { - const run = this.runTracker.getRun(foregroundTurn.runId); - if ( - run && - run.owner === "foreground" && - run.queue === foregroundTurn.queue && - this.runTracker.isRunActive(run) - ) { - foregroundTurn.queue.push(event); - return; - } - this.activeForegroundTurn = null; + foregroundTurn.queue.push(event); + return; } this.liveEventQueue.push(event); } @@ -4175,13 +2986,6 @@ class ClaudeAgentSession implements AgentSession { } } - private async waitForLiveHistoryPoll(): Promise { - await new Promise((resolve) => setTimeout(resolve, 250)); - if (this.claudeSessionId) { - this.loadPersistedHistory(this.claudeSessionId, { dispatchLive: true }); - } - } - private loadPersistedHistory( sessionId: string, options?: { dispatchLive?: boolean } @@ -4506,7 +3310,7 @@ class ClaudeAgentSession implements AgentSession { if (typeof block.tool_use_id === "string") { this.toolUseCache.delete(block.tool_use_id); - this.activeSidechains.delete(block.tool_use_id); + this.sidechainTracker.delete(block.tool_use_id); } } diff --git a/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts b/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts new file mode 100644 index 000000000..0cca7e753 --- /dev/null +++ b/packages/server/src/server/agent/providers/claude/sidechain-tracker.ts @@ -0,0 +1,324 @@ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; + +import { mapClaudeRunningToolCall } from "./tool-call-mapper.js"; +import { buildToolCallDisplayModel } from "../../../../shared/tool-call-display.js"; + +import type { + AgentMetadata, + AgentStreamEvent, + AgentTimelineItem, +} from "../../agent-sdk-types.js"; + +type ClaudeContentChunk = { type: string; [key: string]: unknown }; + +type SubAgentActionEntry = { + index: number; + toolName: string; + summary?: string; +}; + +type SubAgentActivityState = { + subAgentType?: string; + description?: string; + actions: SubAgentActionEntry[]; + actionKeys: string[]; + nextActionIndex: number; + actionIndexByKey: Map; +}; + +type SubAgentActionCandidate = { + key: string; + toolName: string; + input: unknown; +}; + +const MAX_SUB_AGENT_LOG_ENTRIES = 200; +const MAX_SUB_AGENT_SUMMARY_CHARS = 160; + +function readTrimmedString(value: unknown): string | undefined { + if (typeof value !== "string") { + return undefined; + } + const trimmed = value.trim(); + return trimmed.length > 0 ? trimmed : undefined; +} + +function isClaudeContentChunk(value: unknown): value is ClaudeContentChunk { + return Boolean( + value && + typeof value === "object" && + typeof (value as { type?: unknown }).type === "string" + ); +} + +export class ClaudeSidechainTracker { + private readonly activeSidechains = new Map(); + private readonly getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; + + constructor(input: { + getToolInput: (toolUseId: string) => AgentMetadata | null | undefined; + }) { + this.getToolInput = input.getToolInput; + } + + handleMessage(message: SDKMessage, parentToolUseId: string): AgentStreamEvent[] { + const state = + this.activeSidechains.get(parentToolUseId) ?? + ({ + actions: [], + actionKeys: [], + nextActionIndex: 1, + actionIndexByKey: new Map(), + } satisfies SubAgentActivityState); + this.activeSidechains.set(parentToolUseId, state); + + const contextUpdated = this.updateSubAgentContextFromTaskInput( + state, + parentToolUseId + ); + const actionCandidates = this.extractSubAgentActionCandidates(message); + let actionUpdated = false; + for (const action of actionCandidates) { + if (this.appendSubAgentAction(state, action)) { + actionUpdated = true; + } + } + + if (!contextUpdated && !actionUpdated) { + return []; + } + + const toolCall = mapClaudeRunningToolCall({ + name: "Task", + callId: parentToolUseId, + input: null, + output: null, + }); + if (!toolCall) { + return []; + } + + const detail: Extract["detail"] = { + type: "sub_agent", + ...(state.subAgentType ? { subAgentType: state.subAgentType } : {}), + ...(state.description ? { description: state.description } : {}), + log: state.actions + .map((action) => + action.summary ? `[${action.toolName}] ${action.summary}` : `[${action.toolName}]` + ) + .join("\n"), + actions: state.actions.map((action) => ({ + index: action.index, + toolName: action.toolName, + ...(action.summary ? { summary: action.summary } : {}), + })), + }; + + return [ + { + type: "timeline", + item: { + ...toolCall, + detail, + }, + provider: "claude", + }, + ]; + } + + delete(toolUseId: string): void { + this.activeSidechains.delete(toolUseId); + } + + clear(): void { + this.activeSidechains.clear(); + } + + private updateSubAgentContextFromTaskInput( + state: SubAgentActivityState, + parentToolUseId: string + ): boolean { + const taskInput = this.getToolInput(parentToolUseId); + const nextSubAgentType = this.normalizeSubAgentText(taskInput?.subagent_type); + const nextDescription = this.normalizeSubAgentText(taskInput?.description); + + let changed = false; + if (nextSubAgentType && nextSubAgentType !== state.subAgentType) { + state.subAgentType = nextSubAgentType; + changed = true; + } + if (nextDescription && nextDescription !== state.description) { + state.description = nextDescription; + changed = true; + } + return changed; + } + + private normalizeSubAgentText(value: unknown): string | undefined { + const normalized = readTrimmedString(value)?.replace(/\s+/g, " "); + if (!normalized) { + return undefined; + } + if (normalized.length <= MAX_SUB_AGENT_SUMMARY_CHARS) { + return normalized; + } + return `${normalized.slice(0, MAX_SUB_AGENT_SUMMARY_CHARS)}...`; + } + + private extractSubAgentActionCandidates(message: SDKMessage): SubAgentActionCandidate[] { + if (message.type === "assistant") { + const content = message.message?.content; + if (!Array.isArray(content)) { + return []; + } + const actions: SubAgentActionCandidate[] = []; + for (const block of content) { + if ( + !isClaudeContentChunk(block) || + !( + block.type === "tool_use" || + block.type === "mcp_tool_use" || + block.type === "server_tool_use" + ) || + typeof block.name !== "string" + ) { + continue; + } + const key = readTrimmedString(block.id) ?? `assistant:${block.name}:${actions.length}`; + actions.push({ + key, + toolName: block.name, + input: block.input ?? null, + }); + } + return actions; + } + + if (message.type === "stream_event") { + const event = message.event; + if (event.type !== "content_block_start") { + return []; + } + const block = isClaudeContentChunk(event.content_block) ? event.content_block : null; + if ( + !block || + !( + block.type === "tool_use" || + block.type === "mcp_tool_use" || + block.type === "server_tool_use" + ) || + typeof block.name !== "string" + ) { + return []; + } + const key = + readTrimmedString(block.id) ?? + `stream:${block.name}:${typeof event.index === "number" ? event.index : 0}`; + return [ + { + key, + toolName: block.name, + input: block.input ?? null, + }, + ]; + } + + if (message.type === "tool_progress") { + const toolName = readTrimmedString(message.tool_name); + if (!toolName) { + return []; + } + const key = readTrimmedString(message.tool_use_id) ?? `progress:${toolName}`; + return [{ key, toolName, input: null }]; + } + + return []; + } + + private appendSubAgentAction( + state: SubAgentActivityState, + candidate: SubAgentActionCandidate + ): boolean { + const normalizedToolName = readTrimmedString(candidate.toolName); + if (!normalizedToolName) { + return false; + } + + const summary = this.deriveSubAgentActionSummary( + normalizedToolName, + candidate.input + ); + const existingIndex = state.actionIndexByKey.get(candidate.key); + + if (existingIndex !== undefined) { + const existing = state.actions[existingIndex]; + if (!existing) { + return false; + } + const nextSummary = existing.summary ?? summary; + if ( + existing.toolName === normalizedToolName && + existing.summary === nextSummary + ) { + return false; + } + state.actions[existingIndex] = { + ...existing, + toolName: normalizedToolName, + ...(nextSummary ? { summary: nextSummary } : {}), + }; + return true; + } + + state.actions.push({ + index: state.nextActionIndex, + toolName: normalizedToolName, + ...(summary ? { summary } : {}), + }); + state.nextActionIndex += 1; + state.actionKeys.push(candidate.key); + this.trimSubAgentTail(state); + this.rebuildSubAgentActionIndex(state); + return true; + } + + private trimSubAgentTail(state: SubAgentActivityState): void { + while (state.actions.length > MAX_SUB_AGENT_LOG_ENTRIES) { + state.actions.shift(); + state.actionKeys.shift(); + } + } + + private rebuildSubAgentActionIndex(state: SubAgentActivityState): void { + state.actionIndexByKey.clear(); + for (let index = 0; index < state.actionKeys.length; index += 1) { + const key = state.actionKeys[index]; + if (key) { + state.actionIndexByKey.set(key, index); + } + } + } + + private deriveSubAgentActionSummary( + toolName: string, + input: unknown + ): string | undefined { + const runningToolCall = mapClaudeRunningToolCall({ + name: toolName, + callId: `sub-agent-summary-${toolName}`, + input, + output: null, + }); + if (!runningToolCall) { + return undefined; + } + const display = buildToolCallDisplayModel({ + name: runningToolCall.name, + status: runningToolCall.status, + error: runningToolCall.error, + detail: runningToolCall.detail, + metadata: runningToolCall.metadata, + }); + return this.normalizeSubAgentText(display.summary); + } +}