fix(streaming): cap overall stream duration to free trickling provider streams (#766)

Third-party gateways can trickle content deltas (e.g. a large tool_use
input_json_delta) just under the idle-watchdog window and never send
message_stop. Each delta resets the idle timer, so the in-kernel watchdog
never fires and the request hangs forever — the desktop shows "running"
with slowly growing tokens and a permanently-pending trace span (#766).

Add an overall-duration watchdog (CLAUDE_STREAM_MAX_DURATION_MS) that is
armed once and never reset by incoming chunks; the desktop injects 600000ms
via buildChildEnv. Orthogonal to the idle watchdog: idle catches a fully
silent stream, this catches an endless trickle that never completes. CLI
default stays 0 (disabled) for backward compatibility.

Verified end-to-end by replaying the issue's real captured SSE against the
actual kernel: trickle hangs without the cap, aborts cleanly with it.

src/server/__tests__/conversation-service.test.ts

@@ -182,6 +182,42 @@ describe('ConversationService', () => {
     await expect(fs.stat(path.dirname(env.CLAUDE_CODE_DIAGNOSTICS_FILE))).resolves.toBeTruthy()
   })
 
+  test('buildChildEnv injects stream watchdog + overall max-duration so a trickling provider stream cannot hang the desktop forever (#766)', async () => {
+    const prev = process.env.CLAUDE_STREAM_MAX_DURATION_MS
+    delete process.env.CLAUDE_STREAM_MAX_DURATION_MS
+    try {
+      const service = new ConversationService() as any
+      const env = (await service.buildChildEnv('/tmp')) as Record<string, string>
+
+      // Idle watchdog frees a fully-silent stream after 240s...
+      expect(env.CLAUDE_ENABLE_STREAM_WATCHDOG).toBe('1')
+      expect(env.CLAUDE_STREAM_IDLE_TIMEOUT_MS).toBe('240000')
+      // ...but the idle timer is reset by EVERY SSE event, so an upstream that
+      // trickles content deltas (a large tool_use input_json_delta) just under
+      // 240s apart keeps it alive forever. The overall-duration cap is NOT reset
+      // by chunks and is what actually frees that case (#766).
+      expect(env.CLAUDE_STREAM_MAX_DURATION_MS).toBe('600000')
+      // Non-streaming fallback stays off — its retry loop also hangs the UI (#766).
+      expect(env.CLAUDE_CODE_DISABLE_NONSTREAMING_FALLBACK).toBe('1')
+    } finally {
+      if (prev === undefined) delete process.env.CLAUDE_STREAM_MAX_DURATION_MS
+      else process.env.CLAUDE_STREAM_MAX_DURATION_MS = prev
+    }
+  })
+
+  test('buildChildEnv lets caller env override the stream max-duration cap (#766)', async () => {
+    const prev = process.env.CLAUDE_STREAM_MAX_DURATION_MS
+    process.env.CLAUDE_STREAM_MAX_DURATION_MS = '120000'
+    try {
+      const service = new ConversationService() as any
+      const env = (await service.buildChildEnv('/tmp')) as Record<string, string>
+      expect(env.CLAUDE_STREAM_MAX_DURATION_MS).toBe('120000')
+    } finally {
+      if (prev === undefined) delete process.env.CLAUDE_STREAM_MAX_DURATION_MS
+      else process.env.CLAUDE_STREAM_MAX_DURATION_MS = prev
+    }
+  })
+
   test('builds hidden CLI spawn options for desktop session subprocesses', () => {
     const env = { CLAUDECODE: '1' }
 

src/server/services/conversationService.ts

@@ -1114,6 +1114,13 @@ export class ConversationService {
       // CLI's 90s idle default kills healthy streams (#766). 240s still frees
       // a truly dead connection without shooting slow ones.
       CLAUDE_STREAM_IDLE_TIMEOUT_MS: cleanEnv.CLAUDE_STREAM_IDLE_TIMEOUT_MS || '240000',
+      // Overall wall-clock cap for one streaming response, NOT reset by chunks.
+      // The 240s idle timer above is reset by every SSE event, so an upstream
+      // that trickles content deltas (e.g. a huge tool_use input_json_delta)
+      // just under 240s apart keeps it alive forever and the request hangs with
+      // no completion (#766: "卡住" with slowly growing tokens). This independent
+      // cap frees such a stream after a fixed duration regardless of trickle.
+      CLAUDE_STREAM_MAX_DURATION_MS: cleanEnv.CLAUDE_STREAM_MAX_DURATION_MS || '600000',
       // When a stream does get aborted, retry as streaming instead of falling
       // back to non-streaming: a non-streaming request must wait for the FULL
       // generation before the first response byte, so slow providers can never

src/services/api/claude.ts

@@ -1944,11 +1944,22 @@ async function* queryModel(
     const STREAM_IDLE_TIMEOUT_MS =
       parseInt(process.env.CLAUDE_STREAM_IDLE_TIMEOUT_MS || "", 10) || 90_000;
     const STREAM_IDLE_WARNING_MS = STREAM_IDLE_TIMEOUT_MS / 2;
+    // Overall wall-clock cap for a single streaming response. UNLIKE the idle
+    // timer, this is NEVER reset by incoming chunks, so it catches upstreams that
+    // trickle content deltas (e.g. a large tool_use input_json_delta) just fast
+    // enough to keep resetting the idle timer but never send message_stop — the
+    // idle watchdog can then never fire and the request hangs forever (#766).
+    // 0 disables it (terminal CLI default); the desktop injects a value.
+    const STREAM_MAX_DURATION_MS =
+      parseInt(process.env.CLAUDE_STREAM_MAX_DURATION_MS || "", 10) || 0;
     let streamIdleAborted = false;
+    // Which watchdog tripped, so the thrown error message is accurate.
+    let streamAbortReason: "idle" | "max_duration" | null = null;
     // performance.now() snapshot when watchdog fires, for measuring abort propagation delay
     let streamWatchdogFiredAt: number | null = null;
     let streamIdleWarningTimer: ReturnType<typeof setTimeout> | null = null;
     let streamIdleTimer: ReturnType<typeof setTimeout> | null = null;
+    let streamMaxDurationTimer: ReturnType<typeof setTimeout> | null = null;
     function clearStreamIdleTimers(): void {
       if (streamIdleWarningTimer !== null) {
         clearTimeout(streamIdleWarningTimer);
@@ -1977,6 +1988,7 @@ async function* queryModel(
       );
       streamIdleTimer = setTimeout(() => {
         streamIdleAborted = true;
+        streamAbortReason = "idle";
         streamWatchdogFiredAt = performance.now();
         logForDebugging(
           `Streaming idle timeout: no chunks received for ${STREAM_IDLE_TIMEOUT_MS / 1000}s, aborting stream`,
@@ -1994,6 +2006,31 @@ async function* queryModel(
       }, STREAM_IDLE_TIMEOUT_MS);
     }
     resetStreamIdleTimer();
+    // Arm the overall-duration watchdog exactly once. It is intentionally NOT
+    // re-armed in resetStreamIdleTimer(), so a steady trickle of chunks cannot
+    // keep the request alive forever (#766).
+    if (streamWatchdogEnabled && STREAM_MAX_DURATION_MS > 0) {
+      streamMaxDurationTimer = setTimeout(() => {
+        streamIdleAborted = true;
+        streamAbortReason = "max_duration";
+        streamWatchdogFiredAt = performance.now();
+        logForDebugging(
+          `Streaming max duration exceeded: no completion after ${STREAM_MAX_DURATION_MS / 1000}s, aborting stream`,
+          { level: "error" },
+        );
+        logForDiagnosticsNoPII("error", "cli_streaming_max_duration_exceeded", {
+          timeoutMs: STREAM_MAX_DURATION_MS,
+        });
+        logEvent("tengu_streaming_idle_timeout", {
+          model:
+            options.model as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
+          request_id: (streamRequestId ??
+            "unknown") as AnalyticsMetadata_I_VERIFIED_THIS_IS_NOT_CODE_OR_FILEPATHS,
+          timeout_ms: STREAM_MAX_DURATION_MS,
+        });
+        releaseStreamResources();
+      }, STREAM_MAX_DURATION_MS);
+    }
 
     startSessionActivity("api_call");
     try {
@@ -2371,6 +2408,10 @@ async function* queryModel(
       }
       // Clear the idle timeout watchdog now that the stream loop has exited
       clearStreamIdleTimers();
+      if (streamMaxDurationTimer !== null) {
+        clearTimeout(streamMaxDurationTimer);
+        streamMaxDurationTimer = null;
+      }
 
       // If the stream was aborted by our idle timeout watchdog, fall back to
       // non-streaming retry rather than treating it as a completed stream.
@@ -2398,7 +2439,11 @@ async function* queryModel(
         // Prevent double-emit: this throw lands in the catch block below,
         // whose exit_path='error' probe guards on streamWatchdogFiredAt.
         streamWatchdogFiredAt = null;
-        throw new Error("Stream idle timeout - no chunks received");
+        throw new Error(
+          streamAbortReason === "max_duration"
+            ? "Stream max duration exceeded - no completion received"
+            : "Stream idle timeout - no chunks received",
+        );
       }
 
       // Detect when the stream completed without producing any assistant messages.
@@ -2471,6 +2516,10 @@ async function* queryModel(
     } catch (streamingError) {
       // Clear the idle timeout watchdog on error path too
       clearStreamIdleTimers();
+      if (streamMaxDurationTimer !== null) {
+        clearTimeout(streamMaxDurationTimer);
+        streamMaxDurationTimer = null;
+      }
 
       // Instrumentation: if the watchdog had already fired and the for-await
       // threw (rather than exiting cleanly), record that the loop DID exit and