feat(core): QWEN_STREAM_IDLE_TIMEOUT_MS env knob for the stream idle timeout

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

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,238 @@ 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('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;
+    });
   });
 });

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 debug
+ * warning) 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(