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 02a09ba44b..b2bc2a189d 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,3 +1,10 @@ +import type { + NonNullableUsage, + Options, + Query, + SDKMessage, + SDKUserMessage, +} from "@anthropic-ai/claude-agent-sdk"; import { afterEach, expect, test, vi } from "vitest"; import { createTestLogger } from "../../../../test-utils/test-logger.js"; @@ -276,6 +283,453 @@ afterEach(() => { queryFactory.mockReset(); }); +function deferred() { + let resolve!: () => void; + const promise = new Promise((done) => { + resolve = done; + }); + return { promise, resolve }; +} + +function unsupportedQueryOperation(): never { + throw new Error("Unexpected query operation in runtime lifecycle test"); +} + +interface RuntimeQueryHooks { + initialize?: (flags: Parameters[0]) => Promise; + retire?: () => Promise; + beforeReply?: () => Promise; +} + +function createRuntimeQuery(prompt: AsyncIterable, hooks: RuntimeQueryHooks = {}) { + const usage: NonNullableUsage = { + input_tokens: 1, + output_tokens: 1, + cache_creation_input_tokens: 0, + cache_read_input_tokens: 0, + cache_creation: { ephemeral_1h_input_tokens: 0, ephemeral_5m_input_tokens: 0 }, + server_tool_use: { web_fetch_requests: 0, web_search_requests: 0 }, + service_tier: "standard", + inference_geo: "", + speed: "standard", + iterations: [], + output_tokens_details: { thinking_tokens: 0 }, + }; + const output = createAsyncQueue(); + const received: string[] = []; + let closed = false; + const stream = (async function* () { + for (;;) { + const next = await output.next(); + if (next.done) return; + yield next.value; + } + })(); + const returnStream = stream.return.bind(stream); + const query: Query = Object.assign(stream, { + interrupt: async () => undefined, + setPermissionMode: async () => undefined, + setModel: async () => undefined, + applyFlagSettings: async (flags: Parameters[0]) => { + await hooks.initialize?.(flags); + }, + return: async () => { + output.end(); + await hooks.retire?.(); + return returnStream(); + }, + supportedCommands: async () => [], + supportedModels: async () => [], + close: () => { + closed = true; + output.end(); + }, + setMcpPermissionModeOverride: unsupportedQueryOperation, + setMaxThinkingTokens: unsupportedQueryOperation, + initializationResult: unsupportedQueryOperation, + reinitialize: unsupportedQueryOperation, + supportedAgents: unsupportedQueryOperation, + mcpServerStatus: unsupportedQueryOperation, + getContextUsage: unsupportedQueryOperation, + usage_EXPERIMENTAL_MAY_CHANGE_DO_NOT_RELY_ON_THIS_API_YET: unsupportedQueryOperation, + readFile: unsupportedQueryOperation, + reloadPlugins: unsupportedQueryOperation, + reloadSkills: unsupportedQueryOperation, + accountInfo: unsupportedQueryOperation, + rewindFiles: unsupportedQueryOperation, + seedReadState: unsupportedQueryOperation, + reconnectMcpServer: unsupportedQueryOperation, + toggleMcpServer: unsupportedQueryOperation, + setMcpServers: unsupportedQueryOperation, + streamInput: unsupportedQueryOperation, + stopTask: unsupportedQueryOperation, + backgroundTasks: unsupportedQueryOperation, + }); + void (async () => { + for await (const message of prompt) { + const text = extractPromptText(message); + received.push(text); + await hooks.beforeReply?.(); + output.push({ + type: "assistant", + uuid: crypto.randomUUID(), + session_id: "runtime-lifecycle", + parent_tool_use_id: null, + message: { + id: crypto.randomUUID(), + type: "message", + role: "assistant", + model: "opus", + content: [{ type: "text", text: `reply to ${text}`, citations: null }], + stop_reason: "end_turn", + stop_sequence: null, + usage, + container: null, + context_management: null, + diagnostics: null, + stop_details: null, + }, + }); + output.push({ + type: "result", + subtype: "success", + uuid: crypto.randomUUID(), + session_id: "runtime-lifecycle", + duration_ms: 1, + duration_api_ms: 1, + is_error: false, + num_turns: 1, + result: `reply to ${text}`, + stop_reason: "end_turn", + total_cost_usd: 0, + usage, + modelUsage: {}, + permission_denials: [], + }); + } + })(); + return { query, received, isClosed: () => closed }; +} + +test("command discovery overlapping startup preserves replies to subsequent prompts", async () => { + const lookup = deferred(); + let lookups = 0; + const queries: ReturnType[] = []; + const session = await new ClaudeAgentClient({ + logger: createTestLogger(), + resolveBinary: async () => { + if (++lookups === 1) await lookup.promise; + return "/test/claude/bin"; + }, + queryFactory: ({ prompt }) => { + if (typeof prompt === "string") throw new Error("Expected streaming input"); + const runtime = createRuntimeQuery(prompt); + queries.push(runtime); + return runtime.query; + }, + }).createSession({ provider: "claude", cwd: process.cwd() }); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + try { + const commands = session.listCommands!(); + const first = session.startTurn("first"); + // Finish all work not blocked by executable discovery before releasing it. + await new Promise((resolve) => setImmediate(resolve)); + lookup.resolve(); + await Promise.all([commands, first]); + await new Promise((resolve) => setImmediate(resolve)); + await session.startTurn("hello"); + await new Promise((resolve) => setImmediate(resolve)); + expect(collectAssistantText(events)).toBe("reply to firstreply to hello"); + expect(events.filter((event) => event.type === "turn_completed")).toHaveLength(2); + expect(queries).toHaveLength(1); + expect(queries[0].received).toEqual(["first", "hello"]); + } finally { + lookup.resolve(); + for (const runtime of queries) runtime.query.close(); + await session.close(); + } +}); + +test("closing during query initialization disposes the process before initialization settles", async () => { + const initializing = deferred(); + const release = deferred(); + const queries: ReturnType[] = []; + const session = await new ClaudeAgentClient({ + logger: createTestLogger(), + resolveBinary: async () => "/test/claude/bin", + queryFactory: ({ prompt }) => { + if (typeof prompt === "string") throw new Error("Expected streaming input"); + const runtime = createRuntimeQuery(prompt, { + initialize: async () => { + initializing.resolve(); + await release.promise; + }, + }); + queries.push(runtime); + return runtime.query; + }, + }).createSession({ provider: "claude", model: "claude-opus-5-5", cwd: process.cwd() }); + const discovery = session.listCommands!(); + const rejected = expect(discovery).rejects.toThrow("Claude session is closed"); + try { + await initializing.promise; + await session.close(); + expect(queries[0].isClosed()).toBe(true); + } finally { + release.resolve(); + await rejected; + await session.close(); + } +}); + +async function createRuntimeHarness( + options: { + resolveBinary?: () => Promise; + hooks?: RuntimeQueryHooks; + } = {}, +) { + const launches: Options[] = []; + const queries: ReturnType[] = []; + const session = await new ClaudeAgentClient({ + logger: createTestLogger(), + resolveBinary: options.resolveBinary ?? (async () => "/test/claude/bin"), + queryFactory: ({ prompt, options: launch }) => { + if (typeof prompt === "string") throw new Error("Expected streaming input"); + launches.push(launch); + const runtime = createRuntimeQuery(prompt, options.hooks); + queries.push(runtime); + return runtime.query; + }, + }).createSession({ + provider: "claude", + model: "claude-opus-5-5", + thinkingOptionId: "low", + cwd: process.cwd(), + }); + return { session, queries, launches }; +} + +test.each([true, false])( + "Fast mode changed to %s during startup reaches the launched query", + async (enabled) => { + const lookup = deferred(); + const entered = deferred(); + const applied: unknown[] = []; + const { session, queries } = await createRuntimeHarness({ + resolveBinary: async () => { + entered.resolve(); + await lookup.promise; + return "/test/claude/bin"; + }, + hooks: { + initialize: async (flags) => { + applied.push(flags.fastMode); + }, + }, + }); + await session.setFeature!("fast_mode", !enabled); + const discovery = session.listCommands!(); + try { + await entered.promise; + await session.setFeature!("fast_mode", enabled); + lookup.resolve(); + await discovery; + await session.listCommands!(); + expect(applied).toEqual([enabled]); + expect(queries).toHaveLength(1); + } finally { + lookup.resolve(); + await discovery; + await session.close(); + } + }, +); + +test("command discovery leaves a thinking change pending until the active turn finishes", async () => { + const reply = deferred(); + const { session, queries, launches } = await createRuntimeHarness({ + hooks: { beforeReply: () => reply.promise }, + }); + try { + const turn = collectUntilTerminal(streamSession(session, "first")); + await new Promise((resolve) => setImmediate(resolve)); + await session.setThinkingOption!("high"); + await session.listCommands!(); + expect(queries).toHaveLength(1); + reply.resolve(); + expect(collectAssistantText(await turn)).toBe("reply to first"); + const next = await collectUntilTerminal(streamSession(session, "hello")); + expect(collectAssistantText(next)).toBe("reply to hello"); + expect(launches.map((launch) => launch.effort)).toEqual(["low", "high"]); + } finally { + reply.resolve(); + await session.close(); + } +}); + +test("command discovery and a prompt share replacement until the old runtime retires", async () => { + const retiring = deferred(); + const release = deferred(); + const { session, queries } = await createRuntimeHarness({ + hooks: { + retire: async () => { + retiring.resolve(); + await release.promise; + }, + }, + }); + try { + await collectUntilTerminal(streamSession(session, "first")); + await session.setThinkingOption!("high"); + const commands = session.listCommands!(); + await retiring.promise; + const next = collectUntilTerminal(streamSession(session, "second")); + await new Promise((resolve) => setImmediate(resolve)); + expect(queries).toHaveLength(1); + release.resolve(); + await commands; + expect(collectAssistantText(await next)).toBe("reply to second"); + expect(collectAssistantText(await collectUntilTerminal(streamSession(session, "third")))).toBe( + "reply to third", + ); + expect(queries).toHaveLength(2); + expect(queries[0].isClosed()).toBe(true); + expect(queries[1].received).toEqual(["second", "third"]); + } finally { + release.resolve(); + await session.close(); + } +}); + +test("a thinking change during startup remains pending for the following turn", async () => { + const lookup = deferred(); + const entered = deferred(); + const { session, launches } = await createRuntimeHarness({ + resolveBinary: async () => { + entered.resolve(); + await lookup.promise; + return "/test/claude/bin"; + }, + }); + try { + await session.setThinkingOption!("medium"); + const first = collectUntilTerminal(streamSession(session, "first")); + await entered.promise; + await session.setThinkingOption!("high"); + lookup.resolve(); + expect(collectAssistantText(await first)).toBe("reply to first"); + expect(collectAssistantText(await collectUntilTerminal(streamSession(session, "second")))).toBe( + "reply to second", + ); + expect(launches.map((launch) => launch.effort)).toEqual(["medium", "high"]); + } finally { + lookup.resolve(); + await session.close(); + } +}); + +test("closing during executable discovery never launches a late query", async () => { + const lookup = deferred(); + const entered = deferred(); + const { session, queries } = await createRuntimeHarness({ + resolveBinary: async () => { + entered.resolve(); + await lookup.promise; + return "/test/claude/bin"; + }, + }); + const discovery = session.listCommands!(); + const rejected = expect(discovery).rejects.toThrow("Claude session is closed"); + await entered.promise; + await session.close(); + lookup.resolve(); + await rejected; + expect(queries).toHaveLength(0); + await expect(session.listCommands!()).rejects.toThrow("Claude session is closed"); +}); + +test("a failed shared initialization releases the runtime for a subsequent prompt", async () => { + let attempts = 0; + const { session, queries } = await createRuntimeHarness({ + hooks: { + initialize: async () => { + if (++attempts === 1) throw new Error("initialization rejected"); + }, + }, + }); + try { + const discovery = session.listCommands!(); + const rejected = expect(discovery).rejects.toThrow("initialization rejected"); + const failed = await collectUntilTerminal(streamSession(session, "first")); + await rejected; + expect(failed.at(-1)).toMatchObject({ type: "turn_failed", error: "initialization rejected" }); + expect(queries[0].isClosed()).toBe(true); + expect(queries[0].received).toEqual([]); + const recovered = await collectUntilTerminal(streamSession(session, "retry")); + expect(collectAssistantText(recovered)).toBe("reply to retry"); + expect(queries).toHaveLength(2); + } finally { + await session.close(); + } +}); + +test("a new turn waits for a naturally ended runtime to finish cleanup", async () => { + const retiring = deferred(); + const release = deferred(); + const { session, queries } = await createRuntimeHarness({ + hooks: { + retire: async () => { + retiring.resolve(); + await release.promise; + }, + }, + }); + try { + await collectUntilTerminal(streamSession(session, "first")); + queries[0].query.close(); + await retiring.promise; + const next = collectUntilTerminal(streamSession(session, "second")); + await new Promise((resolve) => setImmediate(resolve)); + expect(queries).toHaveLength(1); + release.resolve(); + expect(collectAssistantText(await next)).toBe("reply to second"); + expect(queries).toHaveLength(2); + } finally { + release.resolve(); + await session.close(); + } +}); + +test("an interrupted startup failure does not finish its replacement turn twice", async () => { + const entered = deferred(); + const release = deferred(); + const { session } = await createRuntimeHarness({ + hooks: { + initialize: async () => { + entered.resolve(); + await release.promise; + throw new Error("initialization rejected"); + }, + }, + }); + const events: AgentStreamEvent[] = []; + session.subscribe((event) => events.push(event)); + try { + const first = session.startTurn("first"); + await entered.promise; + await session.interrupt(); + const replacement = session.startTurn("second"); + release.resolve(); + await Promise.all([first, replacement]); + expect(events.filter((event) => event.type === "turn_canceled")).toHaveLength(1); + expect(events.filter((event) => event.type === "turn_failed")).toHaveLength(1); + } finally { + release.resolve(); + await session.close(); + } +}); + test("interrupt only calls query.interrupt and leaves the query open", async () => { const logger = createTestLogger(); const queries: ScriptedQuery[] = []; diff --git a/packages/server/src/server/agent/providers/claude/agent.real.e2e.test.ts b/packages/server/src/server/agent/providers/claude/agent.real.e2e.test.ts index 74ff780aac..acac20db10 100644 --- a/packages/server/src/server/agent/providers/claude/agent.real.e2e.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.real.e2e.test.ts @@ -143,10 +143,6 @@ function getLatestCompletedBashCall(events: AgentStreamEvent[]): ToolCallTimelin .find((item) => item.status === "completed" && item.name.toLowerCase() === "bash"); } -function getInternalQuery(session: AgentSession): unknown { - return (session as AgentSession & { query?: unknown }).query ?? null; -} - async function createSession(params?: { cwdPrefix?: string; modeId?: string; @@ -316,7 +312,7 @@ describe("ClaudeAgentSession integration", () => { } }, 60_000); - test("interrupts a running Bash turn and continues on the same query", async () => { + test("interrupts a running Bash turn and continues the same conversation", async () => { const handle = await createSession({ cwdPrefix: "claude-agent-interrupt-continue-", }); @@ -340,9 +336,9 @@ describe("ClaudeAgentSession integration", () => { event.item.name.toLowerCase() === "bash", 45_000, ); - const firstQuery = getInternalQuery(handle.session); + const firstSessionId = (await handle.session.getRuntimeInfo()).sessionId; - expect(firstQuery).toBeTruthy(); + expect(firstSessionId).toEqual(expect.any(String)); await handle.session.interrupt(); @@ -360,9 +356,9 @@ describe("ClaudeAgentSession integration", () => { const followUpEvents = await collectUntilTerminal( streamSession(handle.session, "Respond with exactly: AFTER_INTERRUPT_OK"), ); - const secondQuery = getInternalQuery(handle.session); + const secondSessionId = (await handle.session.getRuntimeInfo()).sessionId; - expect(secondQuery).toBe(firstQuery); + expect(secondSessionId).toBe(firstSessionId); expect(compactText(getAssistantText(followUpEvents))).toContain("after_interrupt_ok"); expect(followUpEvents.at(-1)).toMatchObject({ type: "turn_completed", diff --git a/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts b/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts index 5393926520..1c9c89963d 100644 --- a/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts +++ b/packages/server/src/server/agent/providers/claude/agent.redesign.test.ts @@ -478,31 +478,23 @@ test("logs redacted query summary and never leaks sentinel secrets", async () => test("interruptActiveTurn only interrupts the active query without info logs", async () => { const spy = createSpyLogger(); const session = await createSessionWithLogger(spy.logger); - const internal: { - query: { - interrupt: () => Promise; - return?: () => Promise; - close?: () => void; - } | null; - input: { end: () => void } | null; - queryRestartNeeded: boolean; - mainTurnInFlight: boolean; - interruptActiveTurn: () => Promise; - } = asInternals(session); - const interrupt = vi.fn(async () => undefined); - const queryReturn = vi.fn(async () => undefined); - const end = vi.fn(() => undefined); - internal.query = { - interrupt, - return: queryReturn, - close: vi.fn(() => undefined), - }; - internal.input = { end }; - internal.queryRestartNeeded = false; - internal.mainTurnInFlight = true; + let acknowledgeInit!: () => void; + const initialized = new Promise((resolve) => { + acknowledgeInit = resolve; + }); + const activeQuery = createBaseQueryMock( + vi.fn(() => { + acknowledgeInit(); + return new Promise(() => {}); + }), + ); + activeQuery.next.mockResolvedValueOnce({ done: false, value: claudeTurnInit() }); + sdkQueryFactory.mockImplementation(() => activeQuery); + await session.startTurn("hello"); + await initialized; try { - await internal.interruptActiveTurn(); + await session.interrupt(); const interruptInfoMessages = extractStringLogArgs(spy.info.mock.calls).filter((message) => message.includes("interruptActiveTurn"), @@ -513,12 +505,11 @@ test("interruptActiveTurn only interrupts the active query without info logs", a expect(interruptInfoMessages).toEqual([]); expect(interruptDebugMessages).toEqual([]); - expect(interrupt).toHaveBeenCalledTimes(1); - expect(queryReturn).not.toHaveBeenCalled(); - expect(end).not.toHaveBeenCalled(); - expect(internal.query).not.toBeNull(); - expect(internal.input).not.toBeNull(); - expect(internal.queryRestartNeeded).toBe(false); + expect(activeQuery.interrupt).toHaveBeenCalledTimes(1); + expect(activeQuery.return).not.toHaveBeenCalled(); + expect(activeQuery.close).not.toHaveBeenCalled(); + await session.listCommands!(); + expect(sdkQueryFactory).toHaveBeenCalledTimes(1); } finally { 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 d505689a95..c17df58525 100644 --- a/packages/server/src/server/agent/providers/claude/agent.ts +++ b/packages/server/src/server/agent/providers/claude/agent.ts @@ -2057,6 +2057,14 @@ class ClaudeContextUsageState { } } +interface ClaudeRuntime { + query: Query; + input: AsyncMessageInput; + process: { child: ChildProcess | null }; + pump: Promise | null; + closing: Promise | null; +} + class ClaudeAgentSession implements AgentSession { readonly provider = "claude" as const; readonly capabilities = CLAUDE_CAPABILITIES; @@ -2070,14 +2078,12 @@ class ClaudeAgentSession implements AgentSession { private readonly logger: Logger; private readonly queryFactory?: ClaudeQueryFactory; private readonly resolveBinary: () => Promise; - private query: Query | null = null; + private runtime: ClaudeRuntime | null = null; + private runtimeStarting: Promise | null = null; private readonly harnessEnvironment: Record; private readonly usageSessionKey = randomUUID(); - private childProcess: ChildProcess | null = null; - private input: AsyncMessageInput | null = null; - /** The exact SDK query/input pair that owns the current foreground turn. */ - private activeForegroundQuery: Query | null = null; - private activeForegroundInput: AsyncMessageInput | null = null; + /** The runtime that owns the foreground or autonomous turn. */ + private activeTurnRuntime: ClaudeRuntime | null = null; /** * Messages pushed into the live SDK input (steers, and a foreground turn's own prompt) whose * command_lifecycle has not reached "started". Interrupting has to withdraw them, or the SDK @@ -2133,7 +2139,6 @@ class ClaudeAgentSession implements AgentSession { private lastRuntimeModel: string | null = null; private compacting = false; private compactionMarkerOpen = false; - private queryPumpPromise: Promise | null = null; private queryRestartNeeded = false; private pendingInterruptAbort = false; /** @@ -2319,19 +2324,15 @@ class ClaudeAgentSession implements AgentSession { this.notifySubscribers({ type: "turn_started", provider: "claude" }); try { - await this.ensureQuery(); - if (cancelIssued) { - // Stopped while Claude was still starting up: withdrawn by never sending it. + const runtime = await this.ensureRuntime(); + if (this.closed || cancelIssued || this.activeForegroundTurnId !== turnId) { + // Stopped during startup: withdraw this turn's message by never sending it. if (sdkUserMessageId) this.unstartedMessageUuids.delete(sdkUserMessageId); return { turnId }; } - if (!this.input) { - throw new Error("Claude session input stream not initialized"); - } - this.activeForegroundQuery = this.query; - this.activeForegroundInput = this.input; - this.startQueryPump(); - this.input.push(sdkMessage); + this.activeTurnRuntime = runtime; + runtime.input.push(sdkMessage); + this.startQueryPump(runtime); setTimeout(() => { if (this.activeForegroundTurnId === turnId) { this.emitSubmittedUserMessage(sdkMessage, turnId, options?.clientMessageId); @@ -2339,9 +2340,13 @@ class ClaudeAgentSession implements AgentSession { }, 0); } catch (error) { if (sdkUserMessageId) this.unstartedMessageUuids.delete(sdkUserMessageId); - this.finishForegroundTurn( - this.buildTurnFailedEvent(error instanceof Error ? error.message : "Claude stream failed"), - ); + if (this.activeForegroundTurnId === turnId) { + this.finishForegroundTurn( + this.buildTurnFailedEvent( + error instanceof Error ? error.message : "Claude stream failed", + ), + ); + } } return { turnId }; @@ -2361,23 +2366,20 @@ class ClaudeAgentSession implements AgentSession { // Capture both ends of the live SDK stream before creating or delivering the message. There // is deliberately no await below: a finished A cannot make this input point at a later B. - const query = this.activeForegroundQuery; - const input = this.activeForegroundInput; - if (!query || !input || this.query !== query || this.input !== input) { + const runtime = this.activeTurnRuntime; + if (!runtime || this.runtime !== runtime) { return { status: "unavailable" }; } const message = this.toSdkUserMessage(prompt); message.priority = "next"; if ( (this.activeForegroundTurnId ?? this.autonomousTurn?.id) !== options.expectedTurnId || - this.activeForegroundQuery !== query || - this.activeForegroundInput !== input || - this.query !== query || - this.input !== input + this.activeTurnRuntime !== runtime || + this.runtime !== runtime ) { return { status: "unavailable" }; } - this.enqueueSteer(input, message, options.clearPendingPermissions === true); + this.enqueueSteer(runtime.input, message, options.clearPendingPermissions === true); return { status: "accepted" }; } @@ -2469,7 +2471,7 @@ class ClaudeAgentSession implements AgentSession { const normalized = isPermissionMode(modeId) ? modeId : "default"; assertClaudeModeCanRun(normalized, this.harnessEnvironment); const previousMode = this.currentMode; - const launchesQuery = !this.query || this.queryRestartNeeded; + const launchesQuery = !this.runtime || this.queryRestartNeeded; const activeQuery = await this.ensureQuery(normalized); try { await activeQuery.setPermissionMode(normalized); @@ -2568,7 +2570,7 @@ class ClaudeAgentSession implements AgentSession { ...this.config.featureValues, fast_mode: enabled, }; - const activeQuery = query ?? this.query; + const activeQuery = query ?? this.runtime?.query; if (activeQuery) { await activeQuery.applyFlagSettings({ fastMode: enabled }); } @@ -2731,8 +2733,8 @@ class ClaudeAgentSession implements AgentSession { sessionId: this.claudeSessionId, turnId: this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? undefined, turnState: this.turnState, - hasQuery: Boolean(this.query), - hasInput: Boolean(this.input), + hasQuery: Boolean(this.runtime), + hasInput: Boolean(this.runtime), hasActiveForegroundTurnId: Boolean(this.activeForegroundTurnId), }, "provider.claude.session_close.start", @@ -2742,35 +2744,17 @@ class ClaudeAgentSession implements AgentSession { this.cancelCurrentTurn?.(); this.subscribers.clear(); this.activeForegroundTurnId = null; - this.activeForegroundQuery = null; - this.activeForegroundInput = null; + this.activeTurnRuntime = null; this.autonomousTurn = null; this.cancelCurrentTurn = null; this.turnState = "idle"; this.sidechainTracker.clear(); this.taskProtocolSource.reset(); - this.input?.end(); - this.query?.close?.(); - await this.awaitWithTimeout(this.query?.interrupt?.(), "close query interrupt"); - await this.awaitWithTimeout(this.query?.return?.(), "close query return"); - this.query = null; - this.input = null; - // Terminate the entire process tree (claude + MCP children) to prevent - // orphan accumulation. The SDK's internal cleanup may only kill the - // direct child process. - if (this.childProcess) { - const result = await terminateWithTreeKill(this.childProcess, { - gracefulTimeoutMs: 2_000, - forceTimeoutMs: 2_000, - }); - if (result === "kill-timeout") { - this.logger.warn( - { pid: this.childProcess.pid, agentId: this.agentId }, - "Claude process tree did not report exit after SIGKILL", - ); - } - this.childProcess = null; - } + const runtime = this.runtime; + this.runtime = null; + if (runtime) await this.disposeRuntime(runtime); + // A pending launch checks closed before spawning and after initialization. Do not + // wait on executable discovery here: closing must work even when discovery stalls. if (this.persistSession === false && this.claudeSessionId) { // Claude Code currently ignores --no-session-persistence outside --print mode // (see `claude --help`), so the SDK's persistSession=false is silently dropped @@ -2984,7 +2968,7 @@ class ClaudeAgentSession implements AgentSession { } private async ensureFreshQuery(): Promise { - if (this.query) { + if (this.runtime) { this.queryRestartNeeded = true; } return this.ensureQuery(); @@ -3164,75 +3148,99 @@ class ClaudeAgentSession implements AgentSession { } private async ensureQuery(launchMode: PermissionMode = this.currentMode): Promise { - if (this.query && !this.queryRestartNeeded) { - return this.query; + return (await this.ensureRuntime(launchMode)).query; + } + + private ensureRuntime(launchMode: PermissionMode = this.currentMode): Promise { + if (this.closed) return Promise.reject(new Error("Claude session is closed")); + if (this.runtimeStarting) return this.runtimeStarting; + if ( + this.runtime && + !this.runtime.closing && + (!this.queryRestartNeeded || this.activeTurnRuntime === this.runtime) + ) { + return Promise.resolve(this.runtime); } - if (this.queryRestartNeeded && this.query) { - const oldQuery = this.query; - const oldInput = this.input; - // Null out query/input BEFORE awaiting the old iterator's return so the - // old pump sees this.query !== activeQuery and skips failActiveTurns. - this.query = null; - this.input = null; - this.queryPumpPromise = null; - this.queryRestartNeeded = false; - // Ending the input retires the process on purpose. Detach first so its - // exit is not reported as a crash. - const retiredChild = this.childProcess; - this.childProcess = null; - if (retiredChild) this.failRunningRuntimeTasks(); - oldInput?.end(); - oldQuery.close?.(); - try { - await oldQuery.return?.(); - } catch { - /* ignore */ - } - // Tree-kill the old process tree now that the SDK has cleaned up. - // If we skip this, MCP children of the previous claude process can - // survive as orphans when the session spawns a replacement query. - if (retiredChild) { - await terminateWithTreeKill(retiredChild, { - gracefulTimeoutMs: 2_000, - forceTimeoutMs: 2_000, - }).catch(() => { - /* process may already be dead */ - }); - } - } + const starting = this.createRuntime(launchMode).finally(() => { + if (this.runtimeStarting === starting) this.runtimeStarting = null; + }); + this.runtimeStarting = starting; + return starting; + } - // Preserve claudeSessionId across query recreation so buildOptions() passes - // resume: sessionId and the new query continues the existing conversation. + private async createRuntime(launchMode: PermissionMode): Promise { + const retired = this.runtime; + this.runtime = null; + if (retired) { + this.failRunningRuntimeTasks(); + await this.disposeRuntime(retired); + } + if (this.closed) throw new Error("Claude session is closed"); + + // Consume only the settings captured by this launch. A change during an await + // remains pending for the next turn, and the pump never initiates a restart. + this.queryRestartNeeded = false; this.persistence = null; - - const input = createAsyncMessageInput(); const options = await this.buildOptions(launchMode); + if (this.closed) throw new Error("Claude session is closed"); this.logger.debug({ options: summarizeClaudeOptionsForLog(options) }, "claude query"); - this.input = input; // A fresh Claude process has no turn of its own in flight. this.mainTurnInFlight = false; - this.query = claudeQuery( + const input = createAsyncMessageInput(); + const process: ClaudeRuntime["process"] = { child: null }; + const query = claudeQuery( { prompt: input.iterable, options }, { runtimeSettings: this.runtimeSettings, launchEnv: this.launchEnv, queryFactory: this.queryFactory, onChildProcess: (child) => { - this.childProcess = child; + process.child = child; child.once("exit", (code, signal) => this.handleRuntimeExit(child, code, signal)); }, }, ); - const fastMode = this.resolveFastModeSetting(); - if (fastMode !== null) { - await this.query.applyFlagSettings({ fastMode }); + const runtime: ClaudeRuntime = { query, input, process, pump: null, closing: null }; + this.runtime = runtime; + try { + const fastMode = this.resolveFastModeSetting(); + if (fastMode !== null) await query.applyFlagSettings({ fastMode }); + if (this.closed) throw new Error("Claude session is closed"); + if (this.runtime !== runtime) throw new Error("Claude runtime exited during initialization"); + return runtime; + } catch (error) { + if (this.runtime === runtime) this.runtime = null; + await this.disposeRuntime(runtime); + throw error; + } + } + + private disposeRuntime(runtime: ClaudeRuntime): Promise { + runtime.closing ??= this.closeRuntime(runtime); + return runtime.closing; + } + + private async closeRuntime(runtime: ClaudeRuntime): Promise { + const child = runtime.process.child; + runtime.process.child = null; + runtime.input.end(); + runtime.query.close?.(); + await this.awaitWithTimeout(runtime.query.interrupt?.(), "close query interrupt"); + await this.awaitWithTimeout(runtime.query.return?.(), "close query return"); + // Claude's SDK cleanup does not own MCP children; retire the entire tree. + if (child) { + const result = await terminateWithTreeKill(child, { + gracefulTimeoutMs: 2_000, + forceTimeoutMs: 2_000, + }); + if (result === "kill-timeout") { + this.logger.warn( + { pid: child.pid, agentId: this.agentId }, + "Claude process tree did not report exit after SIGKILL", + ); + } } - // Do not kick off background control-plane queries here. Methods like - // supportedCommands()/setPermissionMode() may execute immediately after - // ensureQuery() (for listCommands()/setMode()), and sharing the same query - // control plane can cause those calls to wait behind supportedModels(). - return this.query; } private async awaitWithTimeout( @@ -3351,17 +3359,6 @@ class ClaudeAgentSession implements AgentSession { const sdkEnv = this.harnessEnvironment; assertClaudeModeCanRun(permissionMode, sdkEnv); - const claudeBinary = await this.resolveBinary(); - this.logger.debug( - { - claudeBinary, - pathEnvKey: resolvePathEnvKey(), - pathIncludesClaudeLocalBin: (process.env["Path"] ?? process.env["PATH"] ?? "") - .toLowerCase() - .includes("\\.local\\bin"), - }, - "Resolved Claude executable", - ); const sessionBinding: Pick = {}; if (this.pendingFreshSessionId) { sessionBinding.sessionId = this.pendingFreshSessionId; @@ -3379,7 +3376,6 @@ class ClaudeAgentSession implements AgentSession { allowDangerouslySkipPermissions: true, agents: this.defaults?.agents, canUseTool: this.handlePermissionRequest, - pathToClaudeCodeExecutable: claudeBinary, // Use Claude Code preset system prompt and load CLAUDE.md files // Append provider-agnostic system prompts for agents. systemPrompt: { @@ -3429,7 +3425,18 @@ class ClaudeAgentSession implements AgentSession { ...this.runtimeSettings.disallowedTools, ]; } - return base; + const claudeBinary = await this.resolveBinary(); + this.logger.debug( + { + claudeBinary, + pathEnvKey: resolvePathEnvKey(), + pathIncludesClaudeLocalBin: (process.env["Path"] ?? process.env["PATH"] ?? "") + .toLowerCase() + .includes("\\.local\\bin"), + }, + "Resolved Claude executable", + ); + return { pathToClaudeCodeExecutable: claudeBinary, ...base }; } private buildSettingsOptions( @@ -3685,8 +3692,7 @@ class ClaudeAgentSession implements AgentSession { } this.notifySubscribers(event); this.activeForegroundTurnId = null; - this.activeForegroundQuery = null; - this.activeForegroundInput = null; + this.activeTurnRuntime = null; this.cancelCurrentTurn = null; this.activeTurnHasAssistantText = false; this.compactionMarkerOpen = false; @@ -3704,15 +3710,13 @@ class ClaudeAgentSession implements AgentSession { this.compactionMarkerOpen = false; if (this.activeForegroundTurnId) { this.activeForegroundTurnId = null; - this.activeForegroundQuery = null; - this.activeForegroundInput = null; + this.activeTurnRuntime = null; this.cancelCurrentTurn = null; this.activeTurnHasAssistantText = false; this.syncTurnState("foreground turn terminal"); } else if (this.autonomousTurn) { this.autonomousTurn = null; - this.activeForegroundQuery = null; - this.activeForegroundInput = null; + this.activeTurnRuntime = null; this.activeTurnHasAssistantText = false; this.syncTurnState("autonomous turn terminal"); } @@ -3726,8 +3730,7 @@ class ClaudeAgentSession implements AgentSession { this.autonomousTurn = { id: this.createTurnId("autonomous"), }; - this.activeForegroundQuery = this.query; - this.activeForegroundInput = this.input; + this.activeTurnRuntime = this.runtime; this.activeTurnHasAssistantText = false; this.contextUsage.beginTurn(); this.notifySubscribers({ type: "turn_started", provider: "claude" }); @@ -3740,8 +3743,7 @@ class ClaudeAgentSession implements AgentSession { } this.notifySubscribers({ type: "turn_completed", provider: "claude" }); this.autonomousTurn = null; - this.activeForegroundQuery = null; - this.activeForegroundInput = null; + this.activeTurnRuntime = null; this.activeTurnHasAssistantText = false; this.compactionMarkerOpen = false; this.syncTurnState("autonomous turn completed"); @@ -3769,10 +3771,11 @@ class ClaudeAgentSession implements AgentSession { code: number | null, signal: NodeJS.Signals | null, ): void { - if (this.closed || this.childProcess !== child) { + const runtime = this.runtime; + if (this.closed || !runtime || runtime.process.child !== child) { return; } - this.childProcess = null; + runtime.process.child = null; this.logger.warn( { agentId: this.agentId, pid: child.pid, code, signal }, "Claude runtime exited unexpectedly", @@ -3781,13 +3784,19 @@ class ClaudeAgentSession implements AgentSession { if (this.activeForegroundTurnId || this.autonomousTurn) { // The pump is about to throw. It waits for stderr to flush and reports the // real cause; reporting here first would replace that with a bare exit code - // and detach this.query, which is how the pump recognizes its own stream. + // and detach the runtime, which is how the pump recognizes its own stream. return; } // Drop the dead handles so the next write spawns a fresh process instead of // failing against a dead transport forever. - this.query = null; - this.input = null; + void this.disposeRuntime(runtime) + .then(() => { + if (this.runtime === runtime) this.runtime = null; + return undefined; + }) + .catch((error) => { + this.logger.warn({ err: error }, "Failed to retire exited Claude runtime"); + }); this.dispatchEvents([ this.buildTurnFailedEvent( `Claude stopped unexpectedly (${signal ? `signal ${signal}` : `exit code ${code ?? "unknown"}`}). Any background shells, monitors or other work it had running were terminated with it.`, @@ -3803,12 +3812,12 @@ class ClaudeAgentSession implements AgentSession { ); } - private startQueryPump(): void { - if (this.closed || this.queryPumpPromise) { + private startQueryPump(runtime: ClaudeRuntime): void { + if (this.closed || runtime.pump) { return; } - const pump = this.runQueryPump().catch((error) => { + const pump = this.runQueryPump(runtime).catch((error) => { this.logger.trace( { agentId: this.agentId, @@ -3821,33 +3830,16 @@ class ClaudeAgentSession implements AgentSession { ); }); - this.queryPumpPromise = pump; + runtime.pump = pump; void pump.finally(() => { - if (this.queryPumpPromise === pump) { - this.queryPumpPromise = null; + if (runtime.pump === pump) { + runtime.pump = null; } }); } - private async runQueryPump(): Promise { - let activeQuery: Query; - try { - activeQuery = await this.ensureQuery(); - } catch (error) { - this.logger.trace( - { - agentId: this.agentId, - provider: "claude", - sessionId: this.claudeSessionId, - turnId: this.activeForegroundTurnId ?? this.autonomousTurn?.id ?? undefined, - err: error, - }, - "provider.claude.query_pump.init_failed", - ); - this.failActiveTurns(error instanceof Error ? error.message : "Claude stream failed"); - return; - } - + private async runQueryPump(runtime: ClaudeRuntime): Promise { + const activeQuery = runtime.query; let consecutiveInterruptAbortRecoveries = 0; const logRawMessage = (message: SDKMessage): void => { this.logger.trace( @@ -3865,9 +3857,10 @@ class ClaudeAgentSession implements AgentSession { ); }; const handlePumpedMessage = async (message: SDKMessage): Promise => { + if (this.runtime !== runtime) return true; logRawMessage(message); consecutiveInterruptAbortRecoveries = 0; - if (await this.handleMissingResumedConversation(message, activeQuery)) { + if (await this.handleMissingResumedConversation(message, runtime)) { return true; } await this.routeSdkMessageFromPump(message); @@ -3882,19 +3875,19 @@ class ClaudeAgentSession implements AgentSession { return false; }; try { - while (!this.closed && this.query === activeQuery) { + while (!this.closed && this.runtime === runtime) { try { if (await drainActiveQuery()) { return; } - if (!this.closed && this.query === activeQuery) { + if (!this.closed && this.runtime === runtime) { this.failActiveTurns("Claude stream ended before terminal result"); } return; } catch (error) { if ( !this.closed && - this.query === activeQuery && + this.runtime === runtime && this.shouldRecoverInterruptedQueryAbort(error, consecutiveInterruptAbortRecoveries) ) { consecutiveInterruptAbortRecoveries += 1; @@ -3904,7 +3897,7 @@ class ClaudeAgentSession implements AgentSession { ); continue; } - if (!this.closed && this.query === activeQuery) { + if (!this.closed && this.runtime === runtime) { await this.awaitRecentStderrAfterProcessExit(error); this.failActiveTurns(error instanceof Error ? error.message : "Claude stream failed"); } @@ -3912,10 +3905,8 @@ class ClaudeAgentSession implements AgentSession { } } } finally { - if (this.query === activeQuery) { - this.query = null; - this.input = null; - } + await this.disposeRuntime(runtime); + if (this.runtime === runtime) this.runtime = null; } } @@ -4087,7 +4078,7 @@ class ClaudeAgentSession implements AgentSession { private async handleMissingResumedConversation( message: SDKMessage, - activeQuery: Query, + runtime: ClaudeRuntime, ): Promise { const staleResumeError = this.readMissingResumedConversationError(message); if (!staleResumeError) { @@ -4102,29 +4093,8 @@ class ClaudeAgentSession implements AgentSession { ); this.failActiveTurns(staleResumeError); - // Ending the input retires the process on purpose. Detach first so its exit - // is not reported as a crash. - const retiredChild = this.childProcess; - this.childProcess = null; - this.input?.end(); - await this.awaitWithTimeout( - activeQuery.return?.(), - "query pump return on missing resumed conversation", - ); - // Tree-kill for the same reason the restart path does: MCP children of the - // retired claude process outlive it otherwise. - if (retiredChild) { - await terminateWithTreeKill(retiredChild, { - gracefulTimeoutMs: 2_000, - forceTimeoutMs: 2_000, - }).catch(() => { - /* process may already be dead */ - }); - } - if (this.query === activeQuery) { - this.query = null; - this.input = null; - } + // Clear the conversation before yielding to teardown, so a later request + // cannot resume a session that has already been rejected. this.persistence = null; this.persistedHistory = []; this.persistedProviderSubagentEvents = []; @@ -4133,9 +4103,10 @@ class ClaudeAgentSession implements AgentSession { this.queryRestartNeeded = false; this.autonomousTurn = null; this.activeForegroundTurnId = null; - this.activeForegroundQuery = null; - this.activeForegroundInput = null; + this.activeTurnRuntime = null; this.syncTurnState("missing resumed conversation"); + await this.disposeRuntime(runtime); + if (this.runtime === runtime) this.runtime = null; return true; } @@ -4146,7 +4117,7 @@ class ClaudeAgentSession implements AgentSession { * perTaskStopAffordance), and no result would follow to consume pendingInterruptAbort. */ private async interruptActiveTurn(claudeStartedTurn = this.mainTurnInFlight): Promise { - const queryToInterrupt = this.query; + const queryToInterrupt = this.runtime?.query; if (!queryToInterrupt || typeof queryToInterrupt.interrupt !== "function") { this.logger.trace( { diff --git a/packages/server/src/server/daemon-e2e/claude-thinking-memory.real.e2e.test.ts b/packages/server/src/server/daemon-e2e/claude-thinking-memory.real.e2e.test.ts index c1640a7d2d..82324a980e 100644 --- a/packages/server/src/server/daemon-e2e/claude-thinking-memory.real.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/claude-thinking-memory.real.e2e.test.ts @@ -60,7 +60,7 @@ describe("daemon E2E (real claude) - thinking effort memory", () => { } }); - test("changing thinking effort preserves the previous conversation", async () => { + test("concurrent command discovery and thinking changes preserve replies and conversation", async () => { const logger = pino({ level: "silent" }); const cwd = tmpCwd(); const daemon = await createTestPaseoDaemon({ @@ -88,10 +88,13 @@ describe("daemon E2E (real claude) - thinking effort memory", () => { model: model.id, }); - await client.sendMessage( - agent.id, - "Remember the code phrase PASEO_MEMORY_56. Reply exactly: ACK_56", - ); + await Promise.all([ + client.listCommands(agent.id), + client.sendMessage( + agent.id, + "Remember the code phrase PASEO_MEMORY_56. Reply exactly: ACK_56", + ), + ]); const firstFinish = await client.waitForFinish(agent.id, 180_000); expect(firstFinish.status).toBe("idle"); expect(firstFinish.final?.lastError).toBeUndefined(); @@ -99,16 +102,25 @@ describe("daemon E2E (real claude) - thinking effort memory", () => { await client.setAgentThinkingOption(agent.id, "low"); - await client.sendMessage( - agent.id, - "What code phrase did I ask you to remember? Reply exactly with that code phrase and nothing else.", - ); + await Promise.all([ + client.listCommands(agent.id), + client.sendMessage( + agent.id, + "What code phrase did I ask you to remember? Reply exactly with that code phrase and nothing else.", + ), + ]); const secondFinish = await client.waitForFinish(agent.id, 180_000); expect(secondFinish.status).toBe("idle"); expect(secondFinish.final?.lastError).toBeUndefined(); const assistantText = await getAssistantText(client, agent.id); expect(compactText(assistantText)).toContain("paseo_memory_56"); + + await client.sendMessage(agent.id, "Reply exactly: FOLLOWUP_56"); + const followup = await client.waitForFinish(agent.id, 180_000); + expect(followup.status).toBe("idle"); + expect(followup.final?.lastError).toBeUndefined(); + expect(compactText(await getAssistantText(client, agent.id))).toContain("followup_56"); } finally { await client.close().catch(() => undefined); await daemon.close().catch(() => undefined);