fix(agent-mode): recover from stalled Claude SDK streams instead of hanging (#2595)

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

* 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) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

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<void>((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);
+  });
+});

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);

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<SDKMessage>): Promise<string[]> {
+  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<void>((r) => {
+            wake = r;
+          });
+        }
+      },
+    } as AsyncIterable<SDKMessage>,
+  };
+}
+
+describe("guardSdkStreamStall", () => {
+  it("yields every message unchanged and never trips on a clean stream", async () => {
+    async function* source(): AsyncGenerator<SDKMessage> {
+      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<SDKMessage> {
+      yield streamEvent("message_start");
+      yield streamEvent("content_block_delta");
+      await new Promise<void>((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<SDKMessage> {
+      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);
+  });
+});

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<SDKMessage>,
+  { abortController, timeoutMs = SDK_STREAM_STALL_TIMEOUT_MS, onStall }: SdkStreamStallGuardOptions
+): AsyncGenerator<SDKMessage> {
+  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);
+}