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