From ce3378bc4e5ec63a98f398053757f6636b5506d5 Mon Sep 17 00:00:00 2001 From: Jacob Kim Date: Fri, 7 Aug 2026 22:43:46 +0900 Subject: [PATCH] fix(provider): recover Claude startup before first prompt --- electron/providers/claude-sdk-runtime.ts | 101 ++++++++++++++++++++-- electron/providers/runtime.ts | 7 ++ tests/claude-sdk-runtime.test.ts | 95 ++++++++++++++++++++ tests/provider-lifecycle-contract.test.ts | 12 ++- 4 files changed, 208 insertions(+), 7 deletions(-) diff --git a/electron/providers/claude-sdk-runtime.ts b/electron/providers/claude-sdk-runtime.ts index ec432f80..db30fed5 100644 --- a/electron/providers/claude-sdk-runtime.ts +++ b/electron/providers/claude-sdk-runtime.ts @@ -3292,6 +3292,13 @@ export function resolveClaudeTurnStopReason(args: { return args.currentStopReason; } +export function resolveClaudeStreamTerminalStopReason(args: { + abortRequested: boolean; + currentStopReason?: string; +}): string | undefined { + return args.abortRequested ? "user_abort" : args.currentStopReason; +} + export function buildClaudeReadOnlyPromptOptions(args: { cwd: string; model: string; @@ -4145,6 +4152,42 @@ export async function waitForClaudeMcpReadiness(args: { : null; } +function isClaudeInitialStartupMessage(message: SDKMessage) { + return ( + message.type === "system" && + (message as SDKSystemMessage).subtype === "init" + ); +} + +/** + * A streaming-input query can finish after SDK initialization but before it + * consumes the first queued user message. That startup-only close is safe to + * retry because the model has not produced output or invoked a tool yet. + */ +export async function* recoverClaudeStreamBeforeInitialTurnWork(args: { + initialStream: AsyncIterable; + createRecoveryStream: () => AsyncIterable; + isAbortRequested: () => boolean; + onRecovery?: () => void; +}): AsyncGenerator { + let startupOnly = true; + for await (const message of args.initialStream) { + if (!isClaudeInitialStartupMessage(message)) { + startupOnly = false; + } + yield message; + } + + if (!startupOnly || args.isAbortRequested()) { + return; + } + + args.onRecovery?.(); + for await (const message of args.createRecoveryStream()) { + yield message; + } +} + type ClaudeMcpAuthenticateResult = { authUrl?: unknown; authorizationUrl?: unknown; @@ -4689,9 +4732,10 @@ export async function streamClaudeWithSdk( // responder is registered after that push for the same reason — a steer // arriving during the gate must not jump ahead of the primary message. inputQueue = new SteerableUserMessageQueue(); + let queryOptions: Options | null = null; const queryResult = queryFn({ prompt: inputQueue, - options: buildClaudeQueryOptions({ + options: (queryOptions = buildClaudeQueryOptions({ cwd: runtimeCwd, claudeExecutablePath, runtimeOptions: args.runtimeOptions, @@ -5086,13 +5130,15 @@ export async function streamClaudeWithSdk( throw error; } }, - }), + })), }) as Query; stream = queryResult; // Register abort handler using the official Query.close() method const gateAbort = new AbortController(); + let abortRequested = false; args.registerAbort?.(() => { + abortRequested = true; gateAbort.abort(); inputQueue?.close(); stream?.close(); @@ -5126,8 +5172,16 @@ export async function streamClaudeWithSdk( } } + const initialPromptMessage = buildClaudeSDKUserMessage({ + text: providerPrompt, + }); if (!gateAbort.signal.aborted) { - inputQueue.push(buildClaudeSDKUserMessage({ text: providerPrompt })); + const accepted = inputQueue.push(initialPromptMessage); + if (!accepted && !abortRequested) { + throw new Error( + "Claude input queue closed before the initial prompt was accepted.", + ); + } } args.registerSteerResponder?.(async ({ text }) => { if ( @@ -5164,8 +5218,39 @@ export async function streamClaudeWithSdk( const claudeDebugStream = args.runtimeOptions?.debug ?? process.env.STAVE_CLAUDE_DEBUG === "1"; const subagentTracker = new SubagentProgressTracker(); + const recoverableStream = recoverClaudeStreamBeforeInitialTurnWork({ + initialStream: queryResult, + isAbortRequested: () => abortRequested, + onRecovery: () => { + console.warn( + "[claude-sdk-runtime] Claude query closed before initial turn work; retrying with the prompt preloaded", + { taskId: args.taskId }, + ); + }, + createRecoveryStream: () => { + inputQueue?.close(); + queryResult.close(); - for await (const message of stream) { + const recoveryInputQueue = new SteerableUserMessageQueue(); + if (!recoveryInputQueue.push(initialPromptMessage)) { + throw new Error( + "Claude recovery input queue closed before the initial prompt was accepted.", + ); + } + if (!queryOptions) { + throw new Error("Claude query options were unavailable for recovery."); + } + inputQueue = recoveryInputQueue; + const recoveryQuery = queryFn({ + prompt: recoveryInputQueue, + options: queryOptions, + }) as Query; + stream = recoveryQuery; + return recoveryQuery; + }, + }); + + for await (const message of recoverableStream) { if ( message.type === "system" && (message as SDKSystemMessage).subtype === "init" @@ -5310,8 +5395,12 @@ export async function streamClaudeWithSdk( } } - const done: BridgeEvent = finalStopReason - ? { type: "done", stop_reason: finalStopReason } + const terminalStopReason = resolveClaudeStreamTerminalStopReason({ + abortRequested, + currentStopReason: finalStopReason, + }); + const done: BridgeEvent = terminalStopReason + ? { type: "done", stop_reason: terminalStopReason } : { type: "done" }; if (eventCollector.overflowed) { for (const overflowEvent of CLAUDE_OVERFLOW_TAIL_EVENTS) { diff --git a/electron/providers/runtime.ts b/electron/providers/runtime.ts index 8d80aea3..8ab878fe 100644 --- a/electron/providers/runtime.ts +++ b/electron/providers/runtime.ts @@ -979,6 +979,13 @@ async function runProviderTurn( emittedCounts.set(key, (emittedCounts.get(key) ?? 0) + 1); } for (const event of events) { + // The shared lifecycle owns the final abort classification. A timed-out + // adapter can return its locally collected user-abort terminal after the + // live callback was correctly suppressed; replaying it here would hide + // the outer runtime_failure terminal. + if (abortRequested && event.type === "done") { + continue; + } const key = JSON.stringify(event); const remaining = emittedCounts.get(key) ?? 0; if (remaining > 0) { diff --git a/tests/claude-sdk-runtime.test.ts b/tests/claude-sdk-runtime.test.ts index 5c7febba..5c1b4cd4 100644 --- a/tests/claude-sdk-runtime.test.ts +++ b/tests/claude-sdk-runtime.test.ts @@ -20,6 +20,8 @@ import { mapClaudeMessageToEvents, parseClaudeQuestionList, parseClaudeRouteClassificationJson, + recoverClaudeStreamBeforeInitialTurnWork, + resolveClaudeStreamTerminalStopReason, resolveClaudeTurnStopReason, resolveClaudeDisallowedTools, resolveClaudePlanModeApprovalScope, @@ -587,6 +589,99 @@ describe("resolveClaudeTurnStopReason", () => { }); }); +describe("resolveClaudeStreamTerminalStopReason", () => { + test("preserves an abort when the SDK iterator closes without a result", () => { + expect( + resolveClaudeStreamTerminalStopReason({ + abortRequested: true, + currentStopReason: undefined, + }), + ).toBe("user_abort"); + }); + + test("keeps the SDK stop reason when the turn was not aborted", () => { + expect( + resolveClaudeStreamTerminalStopReason({ + abortRequested: false, + currentStopReason: "max_tokens", + }), + ).toBe("max_tokens"); + }); +}); + +describe("recoverClaudeStreamBeforeInitialTurnWork", () => { + test("retries when the readiness query ends after init but before turn work", async () => { + async function* initialStream() { + yield { type: "system", subtype: "init", session_id: "cold-start" }; + } + async function* recoveryStream() { + yield { type: "assistant", message: { content: [] } }; + yield { type: "result", subtype: "success" }; + } + + let recoveryCount = 0; + const messages: Array<{ type: string }> = []; + for await (const message of recoverClaudeStreamBeforeInitialTurnWork({ + initialStream: initialStream() as AsyncIterable, + createRecoveryStream: () => { + recoveryCount += 1; + return recoveryStream() as AsyncIterable; + }, + isAbortRequested: () => false, + })) { + messages.push(message); + } + + expect(recoveryCount).toBe(1); + expect(messages.map((message) => message.type)).toEqual([ + "system", + "assistant", + "result", + ]); + }); + + test("does not retry after the provider begins turn work", async () => { + async function* initialStream() { + yield { type: "system", subtype: "init", session_id: "started" }; + yield { type: "assistant", message: { content: [] } }; + } + + let recoveryCount = 0; + for await (const _message of recoverClaudeStreamBeforeInitialTurnWork({ + initialStream: initialStream() as AsyncIterable, + createRecoveryStream: () => { + recoveryCount += 1; + return initialStream() as AsyncIterable; + }, + isAbortRequested: () => false, + })) { + // Consume the public stream boundary. + } + + expect(recoveryCount).toBe(0); + }); + + test("does not retry a user-aborted startup", async () => { + async function* initialStream() { + yield { type: "system", subtype: "init", session_id: "aborted" }; + } + + let recoveryCount = 0; + for await (const _message of recoverClaudeStreamBeforeInitialTurnWork({ + initialStream: initialStream() as AsyncIterable, + createRecoveryStream: () => { + recoveryCount += 1; + return initialStream() as AsyncIterable; + }, + isAbortRequested: () => true, + })) { + // Consume the public stream boundary. + } + + expect(recoveryCount).toBe(0); + }); +}); + describe("buildClaudeApprovalPermissionResult", () => { test("returns an allow payload with updated input for approved tools", () => { expect( diff --git a/tests/provider-lifecycle-contract.test.ts b/tests/provider-lifecycle-contract.test.ts index 596f3faf..cde77a45 100644 --- a/tests/provider-lifecycle-contract.test.ts +++ b/tests/provider-lifecycle-contract.test.ts @@ -50,7 +50,9 @@ async function runMockAdapter(args: AdapterArgs) { args.registerAbort?.(resolve); }); adapterState.pendingDecisionCount = 0; - return []; + const doneEvent = { type: "done", stop_reason: "user_abort" } as const; + args.onEvent?.(doneEvent); + return [doneEvent]; } const textEvent = { type: "text", text: "working" } as const; @@ -175,6 +177,10 @@ for (const providerId of ["claude-code", "codex"] as const) { expect(turn.events.filter((event) => event.type === "done")).toHaveLength( 1, ); + expect(turn.events.at(-1)).toEqual({ + type: "done", + stop_reason: "runtime_failure", + }); expect(getProviderRuntimeLifecycleSnapshot()).toMatchObject({ activeSessionCount: 0, activeStreamCount: 0, @@ -211,6 +217,10 @@ for (const providerId of ["claude-code", "codex"] as const) { expect(turn.events.filter((event) => event.type === "done")).toHaveLength( 1, ); + expect(turn.events.at(-1)).toEqual({ + type: "done", + stop_reason: "runtime_failure", + }); expect(getProviderRuntimeLifecycleSnapshot()).toMatchObject({ activeSessionCount: 0, activeStreamCount: 0,