diff --git a/packages/core/src/core/openaiContentGenerator/constants.ts b/packages/core/src/core/openaiContentGenerator/constants.ts index 7c10f6301da..cc444d70a1b 100644 --- a/packages/core/src/core/openaiContentGenerator/constants.ts +++ b/packages/core/src/core/openaiContentGenerator/constants.ts @@ -3,6 +3,15 @@ export const DEFAULT_TIMEOUT = 120000; // only bounds connect + first response, so a stream that returns 200 then // goes silent is otherwise unbounded; this watchdog aborts it. export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 120000; +// Env override (deployment knob) for the streaming inactivity timeout, so a +// daemon deployment can tune it without code — the same way the QWEN_SERVE_* +// params are set. An explicit ContentGeneratorConfig.streamIdleTimeoutMs still +// takes precedence; a malformed value is ignored (falls back to the default). +export const QWEN_STREAM_IDLE_TIMEOUT_MS_ENV = 'QWEN_STREAM_IDLE_TIMEOUT_MS'; +// Maximum JS timer delay (~24.8 days). setTimeout silently compresses larger +// delays to 1ms, which would make the watchdog fire almost immediately, so an +// idle timeout above this is treated as invalid. +export const MAX_STREAM_IDLE_TIMEOUT_MS = 2_147_483_647; export const DEFAULT_MAX_RETRIES = 3; export const DEFAULT_OPENAI_BASE_URL = 'https://api.openai.com/v1'; diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts index d0854e475ef..01e5939d33d 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.test.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.test.ts @@ -21,7 +21,11 @@ import { StreamingToolCallParser } from './streamingToolCallParser.js'; import type { Config } from '../../config/config.js'; import { AuthType, type ContentGeneratorConfig } from '../contentGenerator.js'; import type { OpenAICompatibleProvider } from './provider/index.js'; -import { DEFAULT_STREAM_IDLE_TIMEOUT_MS } from './constants.js'; +import { + DEFAULT_STREAM_IDLE_TIMEOUT_MS, + MAX_STREAM_IDLE_TIMEOUT_MS, + QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, +} from './constants.js'; // Mock dependencies vi.mock('./converter.js', () => ({ @@ -3075,10 +3079,15 @@ describe('ContentGenerationPipeline', () => { return r; }, ); + // Clean baseline: ignore any ambient QWEN_STREAM_IDLE_TIMEOUT_MS from the + // dev/CI shell so the default-timeout tests aren't silently overridden. + // Env-specific tests re-stub it; afterEach unstubs everything. + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, undefined); vi.useFakeTimers(); }); afterEach(() => { vi.useRealTimers(); + vi.unstubAllEnvs(); }); it('aborts and throws ETIMEDOUT when the stream is silent past the idle timeout', async () => { @@ -3422,5 +3431,263 @@ describe('ContentGenerationPipeline', () => { // Proves the bypass: the handler (which would strip the code) is skipped. expect(mockErrorHandler.handle).not.toHaveBeenCalled(); }); + + it('honors QWEN_STREAM_IDLE_TIMEOUT_MS when no explicit config is set', async () => { + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, '3000'); + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(); // no explicit streamIdleTimeoutMs → env applies + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + await vi.advanceTimersByTimeAsync(2999); + expect(settled).toBe(false); // not yet at the env value + await vi.advanceTimersByTimeAsync(1); + await consume; + expect(settled).toBe(true); // tripped at 3000ms from the env + }); + + it('lets an explicit streamIdleTimeoutMs config take precedence over the env', async () => { + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, '1000'); + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(5000); // config 5000 wins over env 1000 + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + await vi.advanceTimersByTimeAsync(1000); // env value — must NOT trip + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(4000); // reach the config value (5000) + await consume; + expect(settled).toBe(true); + }); + + it('ignores a malformed QWEN_STREAM_IDLE_TIMEOUT_MS and falls back to the default', async () => { + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, 'not-a-number'); + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(); // no config; invalid env → default (120000ms) + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + // The effective timeout must be the default: not tripped just before it + // (so a malformed value did not become 0/NaN and fire immediately), and + // tripped exactly at it (so the default — not some other value — is used). + await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS - 1); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await consume; + expect(settled).toBe(true); + }); + + it('ignores an oversized QWEN_STREAM_IDLE_TIMEOUT_MS (beyond the timer ceiling)', async () => { + // A value above the JS timer ceiling must be rejected (fall back to the + // default), not used verbatim. If it were used, the watchdog would be + // scheduled ~24.8 days out, so advancing only to the default would never + // trip it — asserting it trips AT the default proves the value was + // rejected. (In real Node such a delay is silently compressed to 1ms, + // which would make the watchdog fire almost immediately.) + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, '9999999999'); + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(); // no config; oversized env → default (120000ms) + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS - 1); + expect(settled).toBe(false); // not before the default → not used verbatim + await vi.advanceTimersByTimeAsync(1); + await consume; + expect(settled).toBe(true); // trips at the default + }); + + it('rejects a non-decimal QWEN_STREAM_IDLE_TIMEOUT_MS (hex/scientific) and uses the default', async () => { + // Number('0x10') === 16; a strict decimal-integer check must reject it so + // a typo can't silently become a 16ms timeout. + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, '0x10'); + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(); // no config; non-decimal env → default + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS - 1); + expect(settled).toBe(false); // would have tripped at 16ms if '0x10' parsed + await vi.advanceTimersByTimeAsync(1); + await consume; + expect(settled).toBe(true); // trips at the default + }); + + it('rejects an out-of-range config streamIdleTimeoutMs and falls back', async () => { + // A config value above the timer ceiling would overflow setTimeout; it + // must be rejected (fall back to env/default), not used verbatim. + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(MAX_STREAM_IDLE_TIMEOUT_MS + 1); // oversized config + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS - 1); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await consume; + expect(settled).toBe(true); // trips at the default (config rejected) + }); + + it('accepts the exact MAX_STREAM_IDLE_TIMEOUT_MS boundary value', async () => { + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + // The exact ceiling must be accepted (not rejected as out-of-range). + // Guards against an off-by-one changing `<=` to `<`. + const p = buildPipeline(MAX_STREAM_IDLE_TIMEOUT_MS); + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + // Must NOT trip at the default (which would mean the ceiling was rejected). + await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS); + expect(settled).toBe(false); + gated.end(); + await consume; + }); + + it('falls back from an invalid config to the env value (config→env cascade)', async () => { + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, '4000'); + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + // Config is oversized → rejected; env = 4000 → used (not default 120000). + const p = buildPipeline(MAX_STREAM_IDLE_TIMEOUT_MS + 1); + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().catch(() => (settled = true)); + await vi.advanceTimersByTimeAsync(3999); + expect(settled).toBe(false); // not yet at the env value + await vi.advanceTimersByTimeAsync(1); + await consume; + expect(settled).toBe(true); // trips at 4000ms from the env (not 120000 default) + }); + + it('disables the watchdog when QWEN_STREAM_IDLE_TIMEOUT_MS=0', async () => { + vi.stubEnv(QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, '0'); + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(); // no config; env=0 → disabled + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().then( + () => (settled = true), + () => (settled = true), + ); + // Well past the default — must NOT trip (watchdog disabled). + await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS + 60000); + expect(settled).toBe(false); + gated.end(); + await consume; + }); + + it('disables the watchdog with a negative config value', async () => { + const gated = gatedStream(); // silent + (mockClient.chat.completions.create as Mock).mockResolvedValue( + gated.stream, + ); + const p = buildPipeline(-1); // negative → disabled (idleMs > 0 guard) + const gen = await p.executeStream( + streamingRequest(new AbortController().signal), + 'id', + ); + let settled = false; + const consume = (async () => { + for await (const _ of gen) { + /* drain */ + } + })().then( + () => (settled = true), + () => (settled = true), + ); + await vi.advanceTimersByTimeAsync(DEFAULT_STREAM_IDLE_TIMEOUT_MS + 60000); + expect(settled).toBe(false); + gated.end(); + await consume; + }); }); }); diff --git a/packages/core/src/core/openaiContentGenerator/pipeline.ts b/packages/core/src/core/openaiContentGenerator/pipeline.ts index fe3ce4d2a9f..e46178920a3 100644 --- a/packages/core/src/core/openaiContentGenerator/pipeline.ts +++ b/packages/core/src/core/openaiContentGenerator/pipeline.ts @@ -20,7 +20,11 @@ import type { PipelineConfig, RequestContext } from './types.js'; import { redactProxyError } from '../../utils/runtimeFetchOptions.js'; import { runtimeDiagnostics } from '../../utils/runtimeDiagnostics.js'; import { createChildAbortController } from '../../utils/abortController.js'; -import { DEFAULT_STREAM_IDLE_TIMEOUT_MS } from './constants.js'; +import { + DEFAULT_STREAM_IDLE_TIMEOUT_MS, + MAX_STREAM_IDLE_TIMEOUT_MS, + QWEN_STREAM_IDLE_TIMEOUT_MS_ENV, +} from './constants.js'; import { createDebugLogger } from '../../utils/debugLogger.js'; const debugLogger = createDebugLogger('OPENAI_PIPELINE'); @@ -58,6 +62,53 @@ export class StreamInactivityTimeoutError extends Error { } } +/** + * Resolve the effective streaming inactivity timeout (ms). Precedence: + * explicit `ContentGeneratorConfig.streamIdleTimeoutMs` (programmatic, wins — + * including `0` to disable) > the `QWEN_STREAM_IDLE_TIMEOUT_MS` env deployment + * knob > the built-in default. A malformed env value is ignored (with a + * `console.warn`) rather than failing the request. + */ +function resolveStreamIdleTimeoutMs(config: ContentGeneratorConfig): number { + // 1. Explicit config field (programmatic) wins: + // - `<= 0` disables the watchdog (downstream `idleMs > 0` guard skips it). + // - Values above the JS timer ceiling are rejected: setTimeout silently + // compresses them to 1ms, which would fire near-immediately. + // - NaN/Infinity/non-integer are invalid. + const fromConfig = config.streamIdleTimeoutMs; + if (typeof fromConfig === 'number') { + if (Number.isInteger(fromConfig) && fromConfig <= MAX_STREAM_IDLE_TIMEOUT_MS) { + return fromConfig; + } + // eslint-disable-next-line no-console + console.warn( + `[qwen-code] Ignoring out-of-range streamIdleTimeoutMs=${fromConfig} ` + + `(expected an integer in (-∞, ${MAX_STREAM_IDLE_TIMEOUT_MS}]); ` + + `falling back to ${QWEN_STREAM_IDLE_TIMEOUT_MS_ENV}/default.`, + ); + } + // 2. Env deployment knob. Strict decimal integer only — reject hex/scientific + // notation/floats/signs so a typo can't silently become a surprising + // timeout. `0` disables; values above the timer ceiling are rejected. + const raw = process.env[QWEN_STREAM_IDLE_TIMEOUT_MS_ENV]; + const trimmed = raw?.trim(); + if (trimmed) { + if (/^\d+$/.test(trimmed)) { + const parsed = Number(trimmed); + if (parsed <= MAX_STREAM_IDLE_TIMEOUT_MS) { + return parsed; + } + } + // eslint-disable-next-line no-console + console.warn( + `[qwen-code] Ignoring invalid ${QWEN_STREAM_IDLE_TIMEOUT_MS_ENV}="${raw}" ` + + `(expected an integer of milliseconds in [0, ${MAX_STREAM_IDLE_TIMEOUT_MS}]); ` + + `using default ${DEFAULT_STREAM_IDLE_TIMEOUT_MS}ms.`, + ); + } + return DEFAULT_STREAM_IDLE_TIMEOUT_MS; +} + /** * Wraps a streaming chunk source with an inactivity watchdog. If no chunk * arrives for `idleMs`, `abortRequest()` is invoked (to abort the underlying @@ -130,10 +181,16 @@ export type { PipelineConfig } from './types.js'; export class ContentGenerationPipeline { client: OpenAI; private contentGeneratorConfig: ContentGeneratorConfig; + // Resolved once (config field > env > default) so the env read + any + // invalid-value warning happen per pipeline, not per streaming request. + private readonly streamIdleTimeoutMs: number; constructor(private config: PipelineConfig) { this.contentGeneratorConfig = config.contentGeneratorConfig; this.client = this.config.provider.buildClient(); + this.streamIdleTimeoutMs = resolveStreamIdleTimeoutMs( + this.contentGeneratorConfig, + ); } async execute( @@ -205,9 +262,7 @@ export class ContentGenerationPipeline { // response, so a stream that returns 200 then goes silent is otherwise // unbounded. Abort + surface a retryable ETIMEDOUT after `idleMs` of no // chunks. `<= 0` disables it. - const idleMs = - this.contentGeneratorConfig.streamIdleTimeoutMs ?? - DEFAULT_STREAM_IDLE_TIMEOUT_MS; + const idleMs = this.streamIdleTimeoutMs; const guarded = idleMs > 0 ? withStreamInactivityTimeout(