Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
wenshao merged 8 commits into from
Jun 25, 2026
Merged

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
8aa7869
feat(core): allow QWEN_STREAM_IDLE_TIMEOUT_MS env to tune the stream …
doudouOUC Jun 25, 2026
8732aeb
fix(core): bound + resolve-once the stream idle timeout env
doudouOUC Jun 25, 2026
641201d
test(core): harden stream-idle env fallback tests to verify the defau…
doudouOUC Jun 25, 2026
b2553c7
fix(core): harden stream-idle timeout resolution (Codex review)
doudouOUC Jun 25, 2026
76fc6ef
fix(core): address stream-idle timeout review feedback
doudouOUC Jun 25, 2026
b8204d5
fix(core): address remaining stream-idle timeout review comments
doudouOUC Jun 25, 2026
5e28b04
fix(core): fix JSDoc (console.warn not debug) + add boundary acceptan…
doudouOUC Jun 25, 2026
98434a6
codex: address PR review feedback (#5845)
doudouOUC Jun 25, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions packages/core/src/core/openaiContentGenerator/constants.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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';
Expand Down
269 changes: 268 additions & 1 deletion packages/core/src/core/openaiContentGenerator/pipeline.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Comment thread
doudouOUC marked this conversation as resolved.
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', () => ({
Expand Down Expand Up @@ -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 () => {
Expand Down Expand Up @@ -3422,5 +3431,263 @@ describe('ContentGenerationPipeline', () => {
// Proves the bypass: the handler (which would strip the code) is skipped.
expect(mockErrorHandler.handle).not.toHaveBeenCalled();
});

Comment thread
doudouOUC marked this conversation as resolved.
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
Comment thread
doudouOUC marked this conversation as resolved.
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;
});
});
Comment thread
doudouOUC marked this conversation as resolved.
});
Loading
Loading