From 10441fa0565a254f876f79602f8af4dc408b5acb Mon Sep 17 00:00:00 2001 From: Zero Liu Date: Fri, 12 Jun 2026 18:27:16 +0800 Subject: [PATCH] fix(agent-mode): recover from stalled Claude SDK streams instead of hanging (#2595) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * fix(agent-mode): recover from a stalled Claude SDK stream instead of hanging the turn A half-open/dropped streaming response from the Claude Agent SDK could stop yielding mid-message without ever emitting a terminal `result`. The turn loop in `ClaudeSdkBackendProcess.prompt()` only ends on `result`, so it parked forever: the turn stayed "running" indefinitely and the user had to interrupt and re-prompt to recover (observed after approving a plan, mid tool-call streaming). Wrap the SDK stream in `guardStreamStall`, a small async-generator that arms an idle timer only while an assistant message is actively streaming (after the first content block, until `message_stop`) so tool runs, first-token latency, and permission waits aren't timed. On a mid-stream stall it aborts the query (via a wired `AbortController`) and throws a clear error the session surfaces as the turn's in-chat error, plus a transient `Notice` toast (wired at the descriptor layer so `sdk/` stays UI-free). Adds focused unit tests for the guard and integration coverage in the backend. Co-Authored-By: Claude Opus 4.8 (1M context) * fix(agent-mode): surface stream-stall error in chat only, drop the Notice toast The stall error already renders inline as the turn's error — the thrown `STREAM_STALL_MESSAGE` flows through `AgentSession`'s catch into `markMessageError`, which appends `**Error:** …` to the assistant turn — so the extra `Notice` toast was redundant. Remove the `notifyUser` hook and its descriptor wiring; `onStall` now just records the stall in the frame trace. Co-Authored-By: Claude Opus 4.8 (1M context) * refactor(agent-mode): scope the stream-stall guard to the Claude SDK by name Rename `streamStallGuard` → `sdkStreamStallGuard` (`guardSdkStreamStall`, `SDK_STREAM_STALL_*`) and document that it is Claude Agent SDK only. The guard keys off Anthropic streaming events (`content_block_*` / `message_*`); the ACP backends (opencode, codex) speak a different protocol — `session/update` notifications over JSON-RPC request/response — and are not covered by it. The `SDKMessage` type signature already prevents misuse on an ACP stream. No behavior change. Co-Authored-By: Claude Opus 4.8 (1M context) * docs(agent-mode): trim the stream-stall guard header comment Drop the ACP-exclusion paragraph; the "Claude Agent SDK only" title line and the `SDKMessage` type signature already convey the scope. Co-Authored-By: Claude Opus 4.8 (1M context) --------- Co-authored-by: Claude Opus 4.8 (1M context) --- .../sdk/ClaudeSdkBackendProcess.test.ts | 93 +++++++++ src/agentMode/sdk/ClaudeSdkBackendProcess.ts | 17 +- src/agentMode/sdk/sdkStreamStallGuard.test.ts | 188 ++++++++++++++++++ src/agentMode/sdk/sdkStreamStallGuard.ts | 91 +++++++++ 4 files changed, 388 insertions(+), 1 deletion(-) create mode 100644 src/agentMode/sdk/sdkStreamStallGuard.test.ts create mode 100644 src/agentMode/sdk/sdkStreamStallGuard.ts diff --git a/src/agentMode/sdk/ClaudeSdkBackendProcess.test.ts b/src/agentMode/sdk/ClaudeSdkBackendProcess.test.ts index 498b1936..8c85f5cc 100644 --- a/src/agentMode/sdk/ClaudeSdkBackendProcess.test.ts +++ b/src/agentMode/sdk/ClaudeSdkBackendProcess.test.ts @@ -669,3 +669,96 @@ describe("ClaudeSdkBackendProcess.prompt auth gate", () => { expect(checkAuth).toHaveBeenCalledTimes(2); }); }); + +describe("ClaudeSdkBackendProcess.prompt stream-stall watchdog", () => { + beforeEach(() => { + queryMock.mockReset(); + createSdkMcpServerMock.mockClear(); + }); + + function makeProc() { + return new ClaudeSdkBackendProcess({ + pathToClaudeCodeExecutable: "/usr/local/bin/claude", + // eslint-disable-next-line @typescript-eslint/no-explicit-any + app: { vault: {} } as any, + clientVersion: "1.2.3", + descriptor: fakeDescriptor(), + }); + } + + /** + * A query whose stream emits a couple of mid-message deltas (arming the + * watchdog) and then goes silent forever — until the backend's abort + * controller fires, at which point the generator returns (the SDK "stops and + * cleans up"). Reproduces a dropped/half-open response with no terminal + * `result`, which would otherwise park `for await` and wedge the turn. + */ + function makeStallingQuery(arg: unknown) { + const { options } = arg as { options: { abortController: AbortController } }; + const { signal } = options.abortController; + const iter = (async function* () { + yield streamEvent({ type: "message_start", message: {} }); + yield streamEvent({ + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "Draf" }, + }); + await new Promise((resolve) => { + if (signal.aborted) resolve(); + else signal.addEventListener("abort", () => resolve(), { once: true }); + }); + })(); + return Object.assign(iter, { + interrupt: jest.fn().mockResolvedValue(undefined), + setModel: jest.fn().mockResolvedValue(undefined), + setPermissionMode: jest.fn().mockResolvedValue(undefined), + }); + } + + it("aborts the turn and rejects when the stream stalls mid-message", async () => { + queryMock.mockImplementation((arg: unknown) => makeStallingQuery(arg)); + const proc = makeProc(); + const { sessionId } = await proc.newSession({ cwd: "/vault", mcpServers: [] }); + proc.registerSessionHandler(sessionId, () => {}); + + jest.useFakeTimers(); + try { + const turn = proc.prompt({ sessionId, prompt: [{ type: "text", text: "draft a plan" }] }); + // The thrown stall error is what `AgentSession` renders as the in-chat + // turn error via `markMessageError`. + const assertion = expect(turn).rejects.toThrow(/stalled/i); + // Past the idle window; advanceTimersByTimeAsync flushes microtasks so the + // two deltas are consumed and the watchdog timer fires. + await jest.advanceTimersByTimeAsync(61_000); + await assertion; + } finally { + jest.useRealTimers(); + } + // The query was aborted (not left dangling) so the turn can be retried. + const call = getPromptQueryCalls()[0][0] as { options: { abortController: AbortController } }; + expect(call.options.abortController.signal.aborted).toBe(true); + }); + + it("passes an abort controller to query() and never fires while the stream is healthy", async () => { + queryMock.mockImplementation(() => + makeQuery([ + streamEvent({ type: "message_start", message: {} }), + streamEvent({ + type: "content_block_delta", + index: 0, + delta: { type: "text_delta", text: "ok" }, + }), + streamEvent({ type: "message_stop" }), + resultMessage(), + ]) + ); + const proc = makeProc(); + const { sessionId } = await proc.newSession({ cwd: "/vault", mcpServers: [] }); + proc.registerSessionHandler(sessionId, () => {}); + + const resp = await proc.prompt({ sessionId, prompt: [{ type: "text", text: "hi" }] }); + expect(resp.stopReason).toBe("end_turn"); + const call = getPromptQueryCalls()[0][0] as { options: { abortController?: unknown } }; + expect(call.options.abortController).toBeInstanceOf(AbortController); + }); +}); diff --git a/src/agentMode/sdk/ClaudeSdkBackendProcess.ts b/src/agentMode/sdk/ClaudeSdkBackendProcess.ts index 14b91812..b828bbaa 100644 --- a/src/agentMode/sdk/ClaudeSdkBackendProcess.ts +++ b/src/agentMode/sdk/ClaudeSdkBackendProcess.ts @@ -71,6 +71,7 @@ import { logSdkOutbound, logSdkOutboundResult, } from "./sdkDebugTap"; +import { guardSdkStreamStall } from "./sdkStreamStallGuard"; interface SessionState { cwd: string | null; @@ -365,15 +366,29 @@ export class ClaudeSdkBackendProcess implements BackendProcess { params.sessionId ); + // Abort the query if the response stream goes half-open mid-message: that + // would otherwise park the loop below forever and wedge the turn in a + // permanent "running" state. `guardSdkStreamStall` owns the watchdog and + // throws `SDK_STREAM_STALL_MESSAGE` (surfaced to the user) if it trips. + const turnAbort = new AbortController(); + options.abortController = turnAbort; + const q = query({ prompt: promptStream, options }); session.active = q; session.firstPromptStarted = true; + const stream = guardSdkStreamStall(q, { + abortController: turnAbort, + // The thrown stall error lands as an in-chat error on the turn (via + // `AgentSession`'s catch → `markMessageError`); this just records it in + // the frame trace. + onStall: (idleMs) => logSdkError("←", "stream:stalled", { idleMs }, params.sessionId), + }); const translatorState = createTranslatorState(); let stopReason: StopReason = "end_turn"; let resultErrorMessage: string | null = null; try { - for await (const sdkMsg of q) { + for await (const sdkMsg of stream) { if (this.shuttingDown) break; logSdkInbound(describeSdkMessage(sdkMsg), sdkMsg, params.sessionId); const events = translateSdkMessage(sdkMsg, params.sessionId, translatorState); diff --git a/src/agentMode/sdk/sdkStreamStallGuard.test.ts b/src/agentMode/sdk/sdkStreamStallGuard.test.ts new file mode 100644 index 00000000..d3270fac --- /dev/null +++ b/src/agentMode/sdk/sdkStreamStallGuard.test.ts @@ -0,0 +1,188 @@ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; +import { guardSdkStreamStall, SDK_STREAM_STALL_MESSAGE } from "./sdkStreamStallGuard"; + +function streamEvent(innerType: string): SDKMessage { + return { type: "stream_event", event: { type: innerType } } as unknown as SDKMessage; +} + +function resultMessage(): SDKMessage { + return { type: "result", subtype: "success" } as unknown as SDKMessage; +} + +async function collect(stream: AsyncIterable): Promise { + const seen: string[] = []; + for await (const m of stream) { + seen.push( + m.type === "stream_event" ? `stream:${(m as { event: { type: string } }).event.type}` : m.type + ); + } + return seen; +} + +/** A source whose messages and completion are driven by the test. */ +function controllableSource() { + const queue: SDKMessage[] = []; + let wake: (() => void) | null = null; + let done = false; + const wakeUp = (): void => { + const w = wake; + wake = null; + w?.(); + }; + return { + push: (m: SDKMessage): void => { + queue.push(m); + wakeUp(); + }, + finish: (): void => { + done = true; + wakeUp(); + }, + iterable: { + async *[Symbol.asyncIterator]() { + for (;;) { + while (queue.length) yield queue.shift() as SDKMessage; + if (done) return; + await new Promise((r) => { + wake = r; + }); + } + }, + } as AsyncIterable, + }; +} + +describe("guardSdkStreamStall", () => { + it("yields every message unchanged and never trips on a clean stream", async () => { + async function* source(): AsyncGenerator { + yield streamEvent("message_start"); + yield streamEvent("content_block_delta"); + yield streamEvent("message_stop"); + yield resultMessage(); + } + const abortController = new AbortController(); + const onStall = jest.fn(); + + const seen = await collect(guardSdkStreamStall(source(), { abortController, onStall })); + + expect(seen).toEqual([ + "stream:message_start", + "stream:content_block_delta", + "stream:message_stop", + "result", + ]); + expect(onStall).not.toHaveBeenCalled(); + expect(abortController.signal.aborted).toBe(false); + }); + + it("aborts and throws the stall error when a message stalls mid-stream", async () => { + const abortController = new AbortController(); + const onStall = jest.fn(); + // Emits two mid-message chunks (arming the watchdog) then hangs until the + // guard aborts — a half-open response with no terminal `result`. + async function* source(): AsyncGenerator { + yield streamEvent("message_start"); + yield streamEvent("content_block_delta"); + await new Promise((resolve) => { + abortController.signal.addEventListener("abort", () => resolve(), { once: true }); + }); + } + + jest.useFakeTimers(); + try { + const run = collect( + guardSdkStreamStall(source(), { abortController, timeoutMs: 1_000, onStall }) + ); + const assertion = expect(run).rejects.toThrow(SDK_STREAM_STALL_MESSAGE); + await jest.advanceTimersByTimeAsync(1_500); + await assertion; + } finally { + jest.useRealTimers(); + } + expect(onStall).toHaveBeenCalledWith(1_000); + expect(abortController.signal.aborted).toBe(true); + }); + + it("does not start timing until the first content event arrives", async () => { + const abortController = new AbortController(); + const onStall = jest.fn(); + const src = controllableSource(); + + jest.useFakeTimers(); + try { + const run = collect( + guardSdkStreamStall(src.iterable, { abortController, timeoutMs: 1_000, onStall }) + ); + + src.push(streamEvent("message_start")); + await jest.advanceTimersByTimeAsync(5_000); + expect(onStall).not.toHaveBeenCalled(); + expect(abortController.signal.aborted).toBe(false); + + src.push(streamEvent("content_block_delta")); + await jest.advanceTimersByTimeAsync(0); + src.push(streamEvent("message_stop")); + src.push(resultMessage()); + src.finish(); + await expect(run).resolves.toEqual([ + "stream:message_start", + "stream:content_block_delta", + "stream:message_stop", + "result", + ]); + } finally { + jest.useRealTimers(); + } + }); + + it("does not trip during a long quiet gap between messages", async () => { + const abortController = new AbortController(); + const onStall = jest.fn(); + const src = controllableSource(); + + jest.useFakeTimers(); + try { + const run = collect( + guardSdkStreamStall(src.iterable, { abortController, timeoutMs: 1_000, onStall }) + ); + + // Stream a complete message, then go quiet *between* messages. + src.push(streamEvent("message_start")); + src.push(streamEvent("content_block_delta")); + src.push(streamEvent("message_stop")); + await jest.advanceTimersByTimeAsync(0); + + // A gap far longer than the window must not trip the guard here. + await jest.advanceTimersByTimeAsync(5_000); + expect(onStall).not.toHaveBeenCalled(); + expect(abortController.signal.aborted).toBe(false); + + // The stream resumes and ends normally. + src.push(resultMessage()); + src.finish(); + await expect(run).resolves.toEqual([ + "stream:message_start", + "stream:content_block_delta", + "stream:message_stop", + "result", + ]); + } finally { + jest.useRealTimers(); + } + }); + + it("re-throws a real source error unchanged instead of a stall", async () => { + const abortController = new AbortController(); + const onStall = jest.fn(); + async function* source(): AsyncGenerator { + yield streamEvent("message_start"); + throw new Error("network down"); + } + + await expect( + collect(guardSdkStreamStall(source(), { abortController, onStall })) + ).rejects.toThrow("network down"); + expect(onStall).not.toHaveBeenCalled(); + expect(abortController.signal.aborted).toBe(false); + }); +}); diff --git a/src/agentMode/sdk/sdkStreamStallGuard.ts b/src/agentMode/sdk/sdkStreamStallGuard.ts new file mode 100644 index 00000000..69389123 --- /dev/null +++ b/src/agentMode/sdk/sdkStreamStallGuard.ts @@ -0,0 +1,91 @@ +/** + * Mid-stream stall guard — **Claude Agent SDK only.** + * + * Why it exists: the driver loop in `ClaudeSdkBackendProcess.prompt()` advances + * only when the `query()` async-iterator yields. If a streaming response goes + * half-open mid-message (so no terminal `result` ever arrives), `for await` + * would park forever and wedge the turn in a permanent "running" state. + * `guardSdkStreamStall` wraps the stream and, *while an assistant message is + * actively streaming*, aborts the query when no chunk arrives within + * `timeoutMs` — turning a silent hang into a surfaced error the session can + * recover from. + * + * Streamed tokens arrive sub-second once content starts, so a multi-second + * mid-message gap means the stream died. The timer is armed only after the + * first content block event and until `message_stop`; first-token latency and + * gaps *between* messages (tool execution, permission waits) are never timed. + */ +import type { SDKMessage } from "@anthropic-ai/claude-agent-sdk"; + +export const SDK_STREAM_STALL_TIMEOUT_MS = 60_000; + +export const SDK_STREAM_STALL_MESSAGE = + `Claude stopped responding — the response stream stalled mid-reply (no output for ` + + `${SDK_STREAM_STALL_TIMEOUT_MS / 1000}s) and the turn was ended. Send your message again to continue.`; + +export interface SdkStreamStallGuardOptions { + /** Aborted on stall so the SDK stops and cleans up the in-flight request. */ + abortController: AbortController; + /** Override the default idle window (mainly for tests). */ + timeoutMs?: number; + /** Invoked once when a stall is detected, before the abort (for logging). */ + onStall?: (timeoutMs: number) => void; +} + +function streamEventType(msg: SDKMessage): string | null { + if (msg.type !== "stream_event") return null; + const evType = (msg as { event?: { type?: unknown } }).event?.type; + return typeof evType === "string" ? evType : null; +} + +function isContentStreamEvent(evType: string | null): boolean { + return ( + evType === "content_block_start" || + evType === "content_block_delta" || + evType === "content_block_stop" + ); +} + +/** + * Yields every message from `source` unchanged. If the stream stalls mid- + * message, aborts `abortController` and — once the underlying iterator unwinds + * — throws `Error(SDK_STREAM_STALL_MESSAGE)`. Real transport errors propagate + * as-is. + */ +export async function* guardSdkStreamStall( + source: AsyncIterable, + { abortController, timeoutMs = SDK_STREAM_STALL_TIMEOUT_MS, onStall }: SdkStreamStallGuardOptions +): AsyncGenerator { + let stalled = false; + let contentStarted = false; + let timer: number | undefined; + const disarm = (): void => { + if (timer !== undefined) { + window.clearTimeout(timer); + timer = undefined; + } + }; + try { + for await (const msg of source) { + disarm(); + const evType = streamEventType(msg); + if (isContentStreamEvent(evType)) contentStarted = true; + if (evType === "message_stop") contentStarted = false; + if (contentStarted && evType !== "message_stop") { + timer = window.setTimeout(() => { + stalled = true; + onStall?.(timeoutMs); + abortController.abort(); + }, timeoutMs); + } + yield msg; + } + } catch (e) { + // The abort surfaces as an iterator throw; suppress it so the stall error + // below wins. Anything else is a real transport error — re-throw it. + if (!stalled) throw e; + } finally { + disarm(); + } + if (stalled) throw new Error(SDK_STREAM_STALL_MESSAGE); +}