From f793895fb303cd7cc420a1000203e0a35e85d5b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=A8=8B=E5=BA=8F=E5=91=98=E9=98=BF=E6=B1=9F=28Relakkes?= =?UTF-8?q?=29?= Date: Mon, 15 Jun 2026 14:27:03 +0800 Subject: [PATCH] fix(streaming): cap overall stream duration to free trickling provider streams (#766) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../__tests__/conversation-service.test.ts | 36 +++++++++++++ src/server/services/conversationService.ts | 7 +++ src/services/api/claude.ts | 51 ++++++++++++++++++- 3 files changed, 93 insertions(+), 1 deletion(-) diff --git a/src/server/__tests__/conversation-service.test.ts b/src/server/__tests__/conversation-service.test.ts index 6f23aa08..76935042 100644 --- a/src/server/__tests__/conversation-service.test.ts +++ b/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 + + // 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 + 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' } diff --git a/src/server/services/conversationService.ts b/src/server/services/conversationService.ts index de1a54fe..2f665438 100644 --- a/src/server/services/conversationService.ts +++ b/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 diff --git a/src/services/api/claude.ts b/src/services/api/claude.ts index 5824bd3e..0032e2cd 100644 --- a/src/services/api/claude.ts +++ b/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 | null = null; let streamIdleTimer: ReturnType | null = null; + let streamMaxDurationTimer: ReturnType | 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