From 30131967483bdaf1389dbb55060cc8bc8150b87b Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 27 May 2026 00:27:25 +0800 Subject: [PATCH] feat(serve): prompt absolute deadline + SSE writer idle timeout (#4514 T2.9) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Squashed: 8 commits for clean rebase onto daemon_mode_b_main. ๐Ÿค– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- docs/users/qwen-serve.md | 30 +- packages/acp-bridge/src/status.test.ts | 11 +- packages/acp-bridge/src/status.ts | 13 + packages/cli/src/commands/serve.ts | 35 +- packages/cli/src/serve/capabilities.ts | 46 +- packages/cli/src/serve/runQwenServe.ts | 101 +- packages/cli/src/serve/server.test.ts | 985 +++++++++++++++++- packages/cli/src/serve/server.ts | 311 +++++- packages/cli/src/serve/types.ts | 39 +- .../sdk-typescript/src/daemon/DaemonClient.ts | 14 + packages/sdk-typescript/src/daemon/types.ts | 11 + .../test/unit/daemon-public-surface.test.ts | 13 + 12 files changed, 1458 insertions(+), 151 deletions(-) diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index a54b9ffc27e..260da0822b7 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -270,11 +270,31 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401 > "alive" until Node's keepalive probes time out โ€” typically ~2 hours > on Linux defaults. On `--hostname 0.0.0.0` deployments behind such > NATs, phantom SSE connections can accumulate and eventually hit the -> 256 `server.maxConnections` ceiling. Stage 2 will add an -> application-level idle deadline (last-byte-written tracking + -> per-connection timeout). Until then, operators on networks that -> swallow RSTs may want to lower `server.keepAliveTimeout` via a -> reverse proxy or accept periodic daemon restarts. +> 256 `server.maxConnections` ceiling. +> +> Set [`--writer-idle-timeout-ms `](#deadlines-and-writer-idle-timeout) +> (issue [#4514](https://github.com/QwenLM/qwen-code/issues/4514) T2.9) +> to close the gap with an explicit application-level idle deadline: +> when no write has successfully flushed for `n` ms the daemon emits +> a terminal `client_evicted` frame with +> `reason: 'writer_idle_timeout'` and closes the stream. The flag is +> off by default to preserve the legacy contract โ€” operators on +> networks that swallow RSTs should pick a value well above the 15s +> heartbeat interval (e.g. `60000`โ€“`300000`) so legitimate idle +> connections aren't evicted while genuinely stuck writers are +> reaped promptly. Pre-flight `caps.features.includes('writer_idle_timeout')` +> from your SDK to confirm the daemon supports it. + +### Deadlines and writer idle timeout + +Issue [#4514](https://github.com/QwenLM/qwen-code/issues/4514) T2.9 ships two opt-in flags that close the long-running / remote-deployment gaps the 15s heartbeat + AbortSignal don't cover. Both are off by default โ€” single-user loopback workflows stay bit-for-bit unchanged. + +| Flag | Env var | Default | What it does | +| ------------------------------ | ----------------------------------- | ------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `--prompt-deadline-ms ` | `QWEN_SERVE_PROMPT_DEADLINE_MS` | unset | Server-side wallclock cap on a single `POST /session/:id/prompt`. On expiry the daemon aborts the prompt's AbortController and returns HTTP `504` with `{code:"prompt_deadline_exceeded", errorKind:"prompt_deadline_exceeded", deadlineMs:n}`. A per-prompt request body field `deadlineMs` can SHORTEN the effective deadline below the flag but never extend it. Capability tag (conditional): `prompt_absolute_deadline`. | +| `--writer-idle-timeout-ms ` | `QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS` | unset | Per-SSE-connection idle deadline. When no write has SUCCESSFULLY flushed for `n` ms โ€” neither a real event nor the 15s heartbeat โ€” the daemon emits a terminal `client_evicted` frame with `data.reason = 'writer_idle_timeout'` (mirrored on `data.errorKind`) and closes the stream. **Pick a value comfortably above the 15s heartbeat** (e.g. `30000`โ€“`300000`) so legitimate idle streams aren't evicted; values `< 15000` WILL evict otherwise-healthy idle connections before the first heartbeat fires (intentional only for tests / short-lived dev sessions). Capability tag (conditional): `writer_idle_timeout`. | + +Both flags accept a positive integer in milliseconds; `0`, `NaN`, non-integer, or negative values are rejected at boot with a clear error message. CLI flag wins over env var; explicit `ServeOptions` field (embedded callers) wins over env. SDK consumers should pre-flight the matching capability tag before relying on either behavior โ€” daemons predating this PR omit both tags and the request `deadlineMs` field is silently dropped. ## Multi-session & multi-workspace deployment diff --git a/packages/acp-bridge/src/status.test.ts b/packages/acp-bridge/src/status.test.ts index e2040400df7..bf89b0c2a27 100644 --- a/packages/acp-bridge/src/status.test.ts +++ b/packages/acp-bridge/src/status.test.ts @@ -20,9 +20,12 @@ describe('SERVE_ERROR_KINDS', () => { // kinds; PR 14 added `'budget_exhausted'` for MCP guardrail // refusals (see #4175 PR 14); PR 16 added `'stat_failed'` for // non-ENOENT stat failures on workspace memory discovery (see - // #4175 PR 16). Future additions append to this list โ€” the - // order is part of the contract so SDK consumers can pattern- - // match without per-kind lookups. + // #4175 PR 16). Issue #4514 T2.9 appended + // `'prompt_deadline_exceeded'` (POST /session/:id/prompt 504) and + // `'writer_idle_timeout'` (terminal SSE client_evicted frame). + // Future additions append to this list โ€” the order is part of the + // contract so SDK consumers can pattern-match without per-kind + // lookups. expect(SERVE_ERROR_KINDS).toEqual([ 'missing_binary', 'blocked_egress', @@ -33,6 +36,8 @@ describe('SERVE_ERROR_KINDS', () => { 'parse_error', 'stat_failed', 'budget_exhausted', + 'prompt_deadline_exceeded', + 'writer_idle_timeout', ]); }); }); diff --git a/packages/acp-bridge/src/status.ts b/packages/acp-bridge/src/status.ts index 7c70c178fd5..d91da130f16 100644 --- a/packages/acp-bridge/src/status.ts +++ b/packages/acp-bridge/src/status.ts @@ -28,6 +28,19 @@ export const SERVE_ERROR_KINDS = [ // Surfaced on per-server `mcp_server` cells (refused at discovery) // and on the workspace-level `mcp_budget` cell (any refusal this pass). 'budget_exhausted', + // Issue #4514 T2.9: a prompt exceeded the server-configured wallclock + // cap (`--prompt-deadline-ms`) or the request's own `deadlineMs` + // (capped at the server flag). Surfaced on the + // `POST /session/:id/prompt` 504 response so callers can branch on a + // typed kind instead of regex-matching the message. + 'prompt_deadline_exceeded', + // Issue #4514 T2.9: an SSE writer's last successful flush was older + // than `--writer-idle-timeout-ms`. The daemon emits a terminal + // `client_evicted` frame with `reason: 'writer_idle_timeout'` before + // tearing the connection down; the kind appears on the frame payload + // (not an HTTP response โ€” by the time we detect the stall the stream + // is already in flight). + 'writer_idle_timeout', ] as const; export type ServeErrorKind = (typeof SERVE_ERROR_KINDS)[number]; diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 1c02af8210e..2a50b4bc5d1 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -42,6 +42,8 @@ interface ServeArgs { 'mcp-client-budget'?: number; 'mcp-budget-mode'?: 'enforce' | 'warn' | 'off'; 'allow-origin'?: string[]; + 'prompt-deadline-ms'?: number; + 'writer-idle-timeout-ms'?: number; } export const serveCommand: CommandModule = { @@ -150,17 +152,19 @@ export const serveCommand: CommandModule = { type: 'string', array: true, description: - 'T2.4 (#4514). Cross-origin allowlist for browser webui clients. ' + - 'Repeatable; each value must be a canonical URL origin ' + - '(`://[:]`, no trailing slash) or `*` for any ' + - 'origin (loud warning; boot refuses if no bearer token is ' + - 'configured. Recommended: pair with --require-auth on loopback so ' + - '/health and /demo are also bearer-gated). When unset, ' + - 'the daemon rejects every request carrying an `Origin` header with ' + - "403 (today's behavior). Matched origins receive proper CORS " + - 'response headers; unmatched still 403. Example: `--allow-origin ' + - 'http://localhost:3000 --allow-origin http://localhost:5173`. ' + - 'Pre-flight via `caps.features.allow_origin`.', + 'T2.4 (#4514). Cross-origin allowlist for browser webui clients.', + }) + .option('prompt-deadline-ms', { + type: 'number', + description: + 'T2.9 (#4514). Server-side wallclock cap on POST /session/:id/prompt (ms). ' + + 'Falls back to QWEN_SERVE_PROMPT_DEADLINE_MS. Positive integer.', + }) + .option('writer-idle-timeout-ms', { + type: 'number', + description: + 'T2.9 (#4514). Per-SSE-connection idle deadline (ms). ' + + 'Falls back to QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS. Positive integer.', }) as unknown as Argv, handler: async (argv) => { if (!argv['http-bridge']) { @@ -236,12 +240,15 @@ export const serveCommand: CommandModule = { requireAuth: argv['require-auth'], mcpClientBudget, mcpBudgetMode: resolvedMcpMode, - // T2.4 (#4514). Pass through verbatim; runQwenServe re-runs - // `parseAllowOriginPatterns` at boot so a malformed entry is - // rejected before the listener binds. ...(argv['allow-origin'] && argv['allow-origin'].length > 0 ? { allowOrigins: argv['allow-origin'] } : {}), + ...(argv['prompt-deadline-ms'] !== undefined + ? { promptDeadlineMs: argv['prompt-deadline-ms'] } + : {}), + ...(argv['writer-idle-timeout-ms'] !== undefined + ? { writerIdleTimeoutMs: argv['writer-idle-timeout-ms'] } + : {}), }); } catch (err) { writeStderrLine( diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index 44bc0f76404..52e7e5282c2 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -221,17 +221,12 @@ export const SERVE_CAPABILITY_REGISTRY = { // status route (extension data on `/capabilities` would inflate the // descriptor shape; we keep the registry uniform). auth_device_flow: { since: 'v1' }, - // #4175 F3 (Commit 6). Daemon advertises which permission mediation - // policies it can run. Clients introspect `modes` to discover the - // closed set of strategies before relying on `permission_partial_vote` - // / `permission_forbidden` SSE events. The active policy for THIS - // daemon is exposed in the `/capabilities` envelope's - // `policy.permission` field โ€” the mode list here is the - // build-supported set, distinct from runtime configuration. permission_mediation: { since: 'v1', modes: ['first-responder', 'designated', 'consensus', 'local-only'], }, + prompt_absolute_deadline: { since: 'v1' }, + writer_idle_timeout: { since: 'v1' }, } as const satisfies Record; export type ServeFeature = keyof typeof SERVE_CAPABILITY_REGISTRY; @@ -239,24 +234,14 @@ export type ServeFeature = keyof typeof SERVE_CAPABILITY_REGISTRY; /** * Per-deployment feature toggles surfaced through `/capabilities`. * - * `requireAuth` controls whether the conditional `require_auth` tag is - * advertised. `mcpPoolActive` (F2 #4175 commit 5) advertises - * `mcp_workspace_pool` + `mcp_pool_restart` together when the daemon - * runs with the pool enabled (default; off only with - * `QWEN_SERVE_NO_MCP_POOL=1`). Other Wave 4 follow-ups can extend - * this object as more deployment-shape capability tags appear (e.g. - * `redact_errors`). + * advertised. */ export interface AdvertiseFeatureToggles { requireAuth?: boolean; mcpPoolActive?: boolean; - /** - * T2.4 (issue #4514). `true` iff the daemon booted with at least one - * `--allow-origin ` entry. Drives advertisement of the - * `allow_origin` capability tag so SDK / webui clients can pre-flight - * the cross-origin path. - */ allowOriginActive?: boolean; + promptDeadlineMs?: number; + writerIdleTimeoutMs?: number; } /** @@ -296,18 +281,21 @@ export const CONDITIONAL_SERVE_FEATURES: ReadonlyMap< (toggles: AdvertiseFeatureToggles) => boolean > = new Map boolean>([ ['require_auth', (toggles) => toggles.requireAuth === true], - // F2 (#4175 commit 5): pool tags advertise as a unit. Both keys - // share the same predicate so SDK clients can rely on - // `mcp_workspace_pool โ‡’ entryCount/entrySummary fields present` - // and `mcp_pool_restart โ‡’ ?entryIndex= + entries[] response shape - // valid` without per-tag pre-flighting. ['mcp_workspace_pool', (toggles) => toggles.mcpPoolActive === true], ['mcp_pool_restart', (toggles) => toggles.mcpPoolActive === true], - // T2.4 (issue #4514): the `allow_origin` tag tracks whether the - // daemon was booted with at least one `--allow-origin` pattern. SDK - // clients pre-flight on it before issuing a cross-origin request - // they would otherwise expect to be 403'd by `denyBrowserOriginCors`. ['allow_origin', (toggles) => toggles.allowOriginActive === true], + [ + 'prompt_absolute_deadline', + (toggles) => + typeof toggles.promptDeadlineMs === 'number' && + toggles.promptDeadlineMs > 0, + ], + [ + 'writer_idle_timeout', + (toggles) => + typeof toggles.writerIdleTimeoutMs === 'number' && + toggles.writerIdleTimeoutMs > 0, + ], ]); export const SERVE_FEATURES = Object.freeze( diff --git a/packages/cli/src/serve/runQwenServe.ts b/packages/cli/src/serve/runQwenServe.ts index 5de43d04740..32e8ce1ca7b 100644 --- a/packages/cli/src/serve/runQwenServe.ts +++ b/packages/cli/src/serve/runQwenServe.ts @@ -20,28 +20,66 @@ import { createBridgeFileSystemAdapter } from './bridgeFileSystemAdapter.js'; import { createDaemonStatusProvider } from './daemonStatusProvider.js'; import { isLoopbackBind } from './loopbackBinds.js'; import { parseAllowOriginPatterns } from './auth.js'; -import { createPermissionAuditPublisher, PermissionAuditRing, } from './permissionAudit.js'; import { createServeApp, resolveBridgeFsFactory } from './server.js'; -// Wenshao review #4335 / 3272581563 โ€” single runtime source of -// truth for the closed permission-policy set. `validatePolicyConfig` -// derives its valid-set from `permission_mediation.modes` so a -// future 5th policy lands in one place. import { SERVE_CAPABILITY_REGISTRY } from './capabilities.js'; import type { ServeOptions } from './types.js'; import type { WorkspaceFileSystemFactory } from './fs/index.js'; -// Wenshao review #4335 / 3272493805 โ€” use the canonical -// `PermissionPolicy` union from acp-bridge instead of inlining -// the four string literals at the let-declaration, the `as` -// cast, and the validation Set. Same drift-protection rationale -// as types.ts/3271978342. import type { PermissionPolicy } from '@qwen-code/acp-bridge'; const QWEN_SERVER_TOKEN_ENV = 'QWEN_SERVER_TOKEN'; +const QWEN_SERVE_PROMPT_DEADLINE_MS_ENV = 'QWEN_SERVE_PROMPT_DEADLINE_MS'; +const QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS_ENV = + 'QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS'; const SHUTDOWN_FORCE_CLOSE_MS = 5_000; +function isPositiveIntegerMs(value: number): boolean { + return Number.isFinite(value) && Number.isInteger(value) && value > 0; +} + +const MAX_TIMEOUT_MS = 2_147_483_647; + +function assertTimerDelayInRange(name: string, value: number): void { + if (value > MAX_TIMEOUT_MS) { + throw new TypeError( + `Invalid ${name}: ${value}. Exceeds maximum JS timer delay of ` + + `${MAX_TIMEOUT_MS} ms (~24.8 days); Node would silently ` + + `compress longer delays to 1ms.`, + ); + } +} + +/** + * Issue #4514 T2.9. Resolve a positive-integer millisecond value from + * an env var. Returns `undefined` when the var is absent (caller falls + * back to the CLI option / `ServeOptions` field), throws when the var + * is present but malformed so a typo fails the boot loudly instead of + * silently disabling the deadline. Whitespace-only values are also + * treated as malformed (rather than silently "unset") โ€” the JSDoc + * "fail loud on typo" promise applies symmetrically. + */ +function parseDeadlineEnv( + envName: string, + raw: string | undefined, +): number | undefined { + if (raw === undefined) return undefined; + // Don't early-return on empty/whitespace: `Number('')` and + // `Number(' ')` both yield `0`, which the positive-integer check + // below rejects with the standard error message. Silently treating + // `QWEN_SERVE_PROMPT_DEADLINE_MS=" "` as "not set" would let a + // shell-substitution typo slip past. + const trimmed = raw.trim(); + const parsed = Number(trimmed); + if (!isPositiveIntegerMs(parsed)) { + throw new Error( + `Invalid ${envName}="${raw}": must be a positive integer (milliseconds).`, + ); + } + return parsed; +} + /** * Wenshao review #4335 / 3271978374 โ€” boot-time policy validation * errors. Replaces the previous substring-matching of "invalid @@ -321,7 +359,28 @@ export async function runQwenServe( typeof rawToken === 'string' && rawToken.trim().length > 0 ? rawToken.trim() : undefined; - const opts: ServeOptions = { ...optsIn, token }; + // T2.9: env-var fallback for the deadline options. Explicit option + // beats the env beats unset (= unlimited). `parseDeadlineEnv` throws + // on malformed values so an `export QWEN_SERVE_PROMPT_DEADLINE_MS=abc` + // typo fails boot loudly instead of silently disabling the cap. + const promptDeadlineMs = + optsIn.promptDeadlineMs ?? + parseDeadlineEnv( + QWEN_SERVE_PROMPT_DEADLINE_MS_ENV, + process.env[QWEN_SERVE_PROMPT_DEADLINE_MS_ENV], + ); + const writerIdleTimeoutMs = + optsIn.writerIdleTimeoutMs ?? + parseDeadlineEnv( + QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS_ENV, + process.env[QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS_ENV], + ); + const opts: ServeOptions = { + ...optsIn, + token, + ...(promptDeadlineMs !== undefined ? { promptDeadlineMs } : {}), + ...(writerIdleTimeoutMs !== undefined ? { writerIdleTimeoutMs } : {}), + }; // BU-sh: catch the `--hostname localhost:4170` / `127.0.0.1:4170` // typo BEFORE the loopback / token check so the operator sees a @@ -498,6 +557,26 @@ export async function runQwenServe( 'Pass mcpClientBudget=N, or set mcpBudgetMode to "warn" or "off".', ); } + // T2.9: validate the deadline options on the explicit option path. + // The env path is already validated inside `parseDeadlineEnv`. Boot- + // loud so an embedded caller passing `{ promptDeadlineMs: -5 }` + // doesn't end up with a daemon that silently fails to enforce the + // cap, leaving the operator believing the timeout is active. + if (opts.promptDeadlineMs !== undefined) { + if (!isPositiveIntegerMs(opts.promptDeadlineMs)) { + throw new TypeError( + `Invalid promptDeadlineMs: ${opts.promptDeadlineMs}. Must be a positive integer (milliseconds).`, + ); + } + assertTimerDelayInRange('promptDeadlineMs', opts.promptDeadlineMs); + } + if (opts.writerIdleTimeoutMs !== undefined) { + if (!isPositiveIntegerMs(opts.writerIdleTimeoutMs)) { + throw new TypeError( + `Invalid writerIdleTimeoutMs: ${opts.writerIdleTimeoutMs}. Must be a positive integer (milliseconds).`, + ); + } + } // Per-handle env overrides: `undefined` value means "scrub this // var from the child env" โ€” important when a different daemon // in the same process set the var globally previously. Always diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 4ee188a090d..90b7d01af01 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -5,12 +5,18 @@ */ import { realpathSync, promises as fsp } from 'node:fs'; +import type { ServerResponse } from 'node:http'; import * as os from 'node:os'; import * as path from 'node:path'; import { fileURLToPath } from 'node:url'; import { describe, it, expect, afterEach, vi } from 'vitest'; import request from 'supertest'; -import { createServeApp, detectFromLoopback } from './server.js'; +import { + createServeApp, + detectFromLoopback, + PromptDeadlineExceededError, + resolvePromptDeadlineMs, +} from './server.js'; import { runQwenServe, type RunHandle } from './runQwenServe.js'; import { CONDITIONAL_SERVE_FEATURES, @@ -157,12 +163,7 @@ const EXPECTED_STAGE1_FEATURES = [ // and PR 21 (`auth_device_flow`); reflect that here so the assertion // matches the real ordering. // -// F2 (#4175 commit 5): `mcp_workspace_pool` + `mcp_pool_restart` are -// also conditional (gated on `mcpPoolActive` toggle, default-true at -// call site in server.ts but default-OFF at the predicate so a -// no-toggle invocation matches the established `require_auth` -// pattern). Both insert AFTER `workspace_mcp_restart` and BEFORE -// `require_auth` so the registry order matches `capabilities.ts`. +// Conditional tags registered in capabilities.ts registry order. const EXPECTED_REGISTERED_FEATURES = [ // Same order as `SERVE_CAPABILITY_REGISTRY` declaration: // ...always-on PR16/17/19/20/21 features, then F2's conditional @@ -184,6 +185,8 @@ const EXPECTED_REGISTERED_FEATURES = [ 'allow_origin', 'auth_device_flow', 'permission_mediation', + 'prompt_absolute_deadline', + 'writer_idle_timeout', ] as const; interface FakeBridgeOpts { @@ -804,12 +807,7 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { } /** - * Wenshao review #4335 / 3272581557 โ€” supertest in the rest of this - * suite always connects from `127.0.0.1`, leaving the prefix-match - * branches in `detectFromLoopback` (the security gate for - * `local-only` permission policy) without direct coverage. Exercise - * the helper synchronously over the address shapes the Round-5 fix - * widened, plus the fail-closed branches. + * Wenshao review #4335 / 3272581557 โ€” detectFromLoopback tests. */ describe('detectFromLoopback (#4335 / 3272581557)', () => { function fakeReq(addr: string | undefined): { @@ -832,14 +830,7 @@ describe('detectFromLoopback (#4335 / 3272581557)', () => { ['1.2.3.4', false], ['::', false], ['fe80::1', false], - // RFC 1918 private addrs that LOOK loopback-adjacent but aren't. ['127', false], - // Note: `'127.'` is structurally `127.`-prefix so the helper - // accepts it (fail-OPEN for that malformed shape). Real - // `req.socket.remoteAddress` values come from the kernel as - // well-formed dotted-decimal IPs, so this only matters for - // pathological synthetic inputs. Documented for transparency. - // Empty / malformed. ['', false], ])('detectFromLoopback(%s) === %s', (addr, expected) => { expect(detectFromLoopback(fakeReq(addr))).toBe(expected); @@ -851,9 +842,6 @@ describe('detectFromLoopback (#4335 / 3272581557)', () => { }); it('does NOT consult X-Forwarded-For or any HTTP header (security)', () => { - // The function takes only the socket-shaped input โ€” even if the - // express request would carry forwarded headers, this helper - // can't see them. Pin the contract. const reqWithForwardedHeader = { socket: { remoteAddress: '10.0.0.1' }, get: (name: string) => @@ -863,6 +851,15 @@ describe('detectFromLoopback (#4335 / 3272581557)', () => { }); }); +function abortableBridgePromptImpl(): FakeBridgeOpts['promptImpl'] { + return (_sid, _req, signal) => + new Promise((resolve) => { + const onAbort = () => resolve({ stopReason: 'cancelled' }); + if (signal?.aborted) onAbort(); + else signal?.addEventListener('abort', onAbort, { once: true }); + }); +} + describe('createServeApp', () => { describe('serve capability registry', () => { it('returns a fresh ordered registered feature list', () => { @@ -932,12 +929,6 @@ describe('createServeApp', () => { feature === 'mcp_workspace_pool' || feature === 'mcp_pool_restart' ) { - // F2 (#4175 commit 5): both pool tags share the - // `mcpPoolActive` predicate and advertise in lockstep. - // Default-OFF at the predicate (matches `require_auth`'s - // pattern); the server.ts call site flips to default-ON via - // `opts.mcpPoolActive !== false`, so a daemon booted without - // the kill switch advertises both tags by default. expect(predicate({ mcpPoolActive: true })).toBe(true); expect(predicate({ mcpPoolActive: false })).toBe(false); expect(predicate({})).toBe(false); @@ -950,12 +941,6 @@ describe('createServeApp', () => { continue; } if (feature === 'allow_origin') { - // T2.4 (#4514): conditional on `allowOriginActive`, which the - // server.ts call site computes from - // `opts.allowOrigins?.length > 0`. Predicate is default-OFF - // (matches `require_auth` / pool pattern) so a baseline - // daemon without `--allow-origin` keeps today's bit-for-bit - // advertisement shape. expect(predicate({ allowOriginActive: true })).toBe(true); expect(predicate({ allowOriginActive: false })).toBe(false); expect(predicate({})).toBe(false); @@ -967,6 +952,30 @@ describe('createServeApp', () => { ); continue; } + if (feature === 'prompt_absolute_deadline') { + expect(predicate({ promptDeadlineMs: 5_000 })).toBe(true); + expect(predicate({ promptDeadlineMs: 0 })).toBe(false); + expect(predicate({})).toBe(false); + expect( + getAdvertisedServeFeatures(undefined, { promptDeadlineMs: 5_000 }), + ).toContain(feature); + expect(getAdvertisedServeFeatures(undefined, {})).not.toContain( + feature, + ); + continue; + } + if (feature === 'writer_idle_timeout') { + expect(predicate({ writerIdleTimeoutMs: 60_000 })).toBe(true); + expect(predicate({ writerIdleTimeoutMs: 0 })).toBe(false); + expect(predicate({})).toBe(false); + expect( + getAdvertisedServeFeatures(undefined, { writerIdleTimeoutMs: 60_000 }), + ).toContain(feature); + expect(getAdvertisedServeFeatures(undefined, {})).not.toContain( + feature, + ); + continue; + } // Future conditional tag. Authors must add a branch above with // the toggle field that drives this predicate. Failing here is // intentional: it forces the new conditional tag to ship with a @@ -3844,7 +3853,11 @@ describe('runQwenServe', () => { await handle.close(); handle = undefined; } + // Scrub any env vars individual tests may have set so leftover + // state can't leak into the next test in this worker. delete process.env['QWEN_SERVER_TOKEN']; + delete process.env['QWEN_SERVE_PROMPT_DEADLINE_MS']; + delete process.env['QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS']; }); it('refuses to bind 0.0.0.0 without a token', async () => { @@ -3937,6 +3950,169 @@ describe('runQwenServe', () => { ).rejects.toThrow(/enforce.*requires.*mcpClientBudget/); }); + // Issue #4514 T2.9: same boot-validation contract as mcpClientBudget + // โ€” embedded callers must hit the same fail-loud TypeError as the + // CLI handler, not a silent uncapped daemon. + it.each([ + ['zero', 0], + ['negative', -5], + ['float', 1.5], + ['NaN', Number.NaN], + ])( + 'rejects invalid promptDeadlineMs (%s) at boot (#4514 T2.9)', + async (_label, value) => { + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + promptDeadlineMs: value, + }), + ).rejects.toThrow(/promptDeadlineMs/); + }, + ); + + it.each([ + ['zero', 0], + ['negative', -5], + ['float', 1.5], + ['NaN', Number.NaN], + ])( + 'rejects invalid writerIdleTimeoutMs (%s) at boot (#4514 T2.9)', + async (_label, value) => { + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + writerIdleTimeoutMs: value, + }), + ).rejects.toThrow(/writerIdleTimeoutMs/); + }, + ); + + it('rejects promptDeadlineMs that exceeds the JS timer cap (#4514 T2.9 wenshao review)', async () => { + // Node silently compresses setTimeout delays > 2^31-1 ms to 1ms + // with a TimeoutOverflowWarning โ€” an operator setting 30 days + // expecting "effectively no cap" would otherwise see every prompt + // 504 instantly. Boot-loud rejection with a clear error pointing + // at the cap prevents the footwound. + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + promptDeadlineMs: 2_147_483_648, // one over 2^31 - 1 + }), + ).rejects.toThrow(/Exceeds maximum JS timer delay/); + }); + + it('accepts writerIdleTimeoutMs above the JS timer cap (#4530 review)', async () => { + handle = await runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + writerIdleTimeoutMs: 2_147_483_648, + }); + const port = (handle.server.address() as { port: number }).port; + const res = await fetch(`http://127.0.0.1:${port}/capabilities`); + const caps = (await res.json()) as { features: string[] }; + expect(caps.features).toContain('writer_idle_timeout'); + }); + + // Env-var scrub for these tests is handled by `afterEach` above โ€” + // no `try/finally` per test needed. + it.each([ + ['empty string', ''], + ['whitespace only', ' '], + ['NaN', 'abc'], + ['float', '1.5'], + ['negative', '-5'], + ['zero', '0'], + ])( + 'rejects invalid QWEN_SERVE_PROMPT_DEADLINE_MS env var (%s) at boot (#4514 T2.9)', + async (_label, value) => { + process.env['QWEN_SERVE_PROMPT_DEADLINE_MS'] = value; + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + }), + ).rejects.toThrow(/QWEN_SERVE_PROMPT_DEADLINE_MS/); + }, + ); + + it('rejects QWEN_SERVE_PROMPT_DEADLINE_MS that exceeds JS timer cap (#4514 T2.9)', async () => { + process.env['QWEN_SERVE_PROMPT_DEADLINE_MS'] = '2147483648'; + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + }), + ).rejects.toThrow(/Exceeds maximum JS timer delay/); + }); + + it('accepts a valid QWEN_SERVE_PROMPT_DEADLINE_MS env var (#4514 T2.9 happy path)', async () => { + // Pin the env-fallback shape end-to-end: env var โ†’ ServeOptions + // field โ†’ /capabilities advertises the conditional tag. Closes + // the "no tests at all" gap wenshao flagged on `parseDeadlineEnv`. + process.env['QWEN_SERVE_PROMPT_DEADLINE_MS'] = '30000'; + handle = await runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + }); + const port = (handle.server.address() as { port: number }).port; + const res = await fetch(`http://127.0.0.1:${port}/capabilities`); + const caps = (await res.json()) as { features: string[] }; + expect(caps.features).toContain('prompt_absolute_deadline'); + }); + + // wenshao review #4530 inline #5: sibling env var + // `QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS` had zero dedicated coverage + // โ€” a copy-paste error reading the wrong env-var name in + // `runQwenServe` would have passed all existing tests. Mirror the + // prompt-deadline env-var plumbing while preserving writer-idle's + // larger arithmetic-only budget range. + it('rejects invalid QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS env var at boot (#4514 T2.9)', async () => { + process.env['QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS'] = 'abc'; + await expect( + runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + }), + ).rejects.toThrow(/QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS/); + }); + + it('accepts QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS above JS timer cap (#4530 review)', async () => { + process.env['QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS'] = '2147483648'; + handle = await runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + }); + const port = (handle.server.address() as { port: number }).port; + const res = await fetch(`http://127.0.0.1:${port}/capabilities`); + const caps = (await res.json()) as { features: string[] }; + expect(caps.features).toContain('writer_idle_timeout'); + }); + + it('accepts a valid QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS env var (#4514 T2.9 happy path)', async () => { + process.env['QWEN_SERVE_WRITER_IDLE_TIMEOUT_MS'] = '60000'; + handle = await runQwenServe({ + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + }); + const port = (handle.server.address() as { port: number }).port; + const res = await fetch(`http://127.0.0.1:${port}/capabilities`); + const caps = (await res.json()) as { features: string[] }; + expect(caps.features).toContain('writer_idle_timeout'); + }); + // Round 6 (wenshao R5 line 216): replaced the R3 `process.env` // mutation tests. `runQwenServe` now passes per-handle env // overrides via `BridgeOptions.childEnvOverrides`, NOT by mutating @@ -6211,3 +6387,742 @@ describe('auth device-flow routes', () => { expect(identified.body).not.toHaveProperty('verificationUri'); }); }); + +// =========================================================================== +// Issue #4514 T2.9: prompt absolute deadline + SSE writer idle timeout +// =========================================================================== + +describe('T2.9 prompt absolute deadline (issue #4514)', () => { + describe('resolvePromptDeadlineMs', () => { + it('returns undefined when the server flag is unset', () => { + // Default off โ€” preserves the legacy "client disconnect is the + // only auto-cancel" behavior bit-for-bit. A request body + // `deadlineMs` is ignored without the server opting in (we don't + // want a client to be able to force a deadline on an operator + // who hasn't asked for one). + expect(resolvePromptDeadlineMs(undefined, undefined)).toBeUndefined(); + expect(resolvePromptDeadlineMs(undefined, 1_000)).toBeUndefined(); + expect(resolvePromptDeadlineMs(0, 1_000)).toBeUndefined(); + }); + + it('uses the server flag when no request override is present', () => { + expect(resolvePromptDeadlineMs(5_000, undefined)).toBe(5_000); + }); + + it('caps the request override at the server flag (request can shorten)', () => { + // Operator is the upper bound: request can lower the deadline + // but never raise it. + expect(resolvePromptDeadlineMs(5_000, 1_000)).toBe(1_000); + }); + + it('rejects request overrides that exceed the server flag', () => { + // The cap is `Math.min`, so an over-bound request never widens + // the effective deadline. This is the test that locks down the + // "cannot extend" contract called out in the issue. + expect(resolvePromptDeadlineMs(5_000, 10_000)).toBe(5_000); + }); + + it('ignores invalid request overrides without dropping the server cap', () => { + // Malformed request override should be caught at the route layer + // (returns 400), but defense-in-depth: the resolver still falls + // back to the server value rather than silently disabling. + expect(resolvePromptDeadlineMs(5_000, 0)).toBe(5_000); + expect(resolvePromptDeadlineMs(5_000, -100)).toBe(5_000); + expect(resolvePromptDeadlineMs(5_000, Number.NaN)).toBe(5_000); + }); + }); + + describe('POST /session/:id/prompt', () => { + it.each([ + ['negative', -5], + ['zero', 0], + ['float', 1.5], + ['string', 'abc'], + ['boolean', true], + ['object', { ms: 500 }], + ])( + // Note: `NaN` / `Infinity` aren't reachable here โ€” JSON.stringify + // converts both to `null`, which the validator correctly treats + // as "absent" (same as `undefined`). The remaining cases exercise + // every reachable branch of the typeof / isFinite / isInteger / + // positive validator. + 'rejects an invalid `deadlineMs` body field (%s) with 400', + async (_label, value) => { + // Symmetric with the `prompt` validator: malformed inputs fail + // loudly so the client doesn't silently lose their deadline + // request. Each branch of the validator (typeof / isFinite / + // isInteger / positive) gets covered. + const bridge = fakeBridge({ + promptImpl: () => { + throw new Error('bridge must not be touched'); + }, + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + prompt: [{ type: 'text', text: 'hi' }], + deadlineMs: value, + }); + expect(res.status).toBe(400); + expect(res.body.code).toBe('invalid_deadline_ms'); + expect(bridge.promptCalls).toHaveLength(0); + }, + ); + + it('returns cleanly when client disconnects in the same tick the deadline fires (wenshao review #3)', async () => { + // Critical regression from wenshao's CHANGES_REQUESTED on #4530: + // when `res.writableEnded` was true at the moment the deadline + // rejection surfaced, the early code `if (err instanceof + // PromptDeadlineExceededError && !res.writableEnded) { ... + // return; }` would skip BOTH the body AND the return, fall + // through to `sendBridgeError`, and try to write 500 to an + // already-ended response โ†’ ERR_STREAM_WRITE_AFTER_END. + // + // We force the race by destroying the client socket + // immediately after the bridge starts the prompt, so by the + // time the 50ms deadline fires the response is already ended. + // The route MUST handle this without throwing โ€” assertion is + // implicit: a thrown uncaughtException would fail the test. + let promptStarted: (() => void) | undefined; + const promptStartedPromise = new Promise((r) => { + promptStarted = r; + }); + const bridge = fakeBridge({ + promptImpl: (_sid, _req, signal) => + new Promise((resolve) => { + promptStarted!(); + const onAbort = () => resolve({ stopReason: 'cancelled' }); + if (signal?.aborted) onAbort(); + else signal?.addEventListener('abort', onAbort, { once: true }); + }), + }); + const localHandle = await runQwenServe( + { + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + promptDeadlineMs: 50, + }, + { bridge }, + ); + try { + const port = (localHandle.server.address() as { port: number }).port; + const http = await import('node:http'); + const reqBody = JSON.stringify({ + prompt: [{ type: 'text', text: 'slow' }], + }); + const httpReq = http.request({ + host: '127.0.0.1', + port, + method: 'POST', + path: '/session/sess-A/prompt', + headers: { + 'content-type': 'application/json', + 'content-length': Buffer.byteLength(reqBody), + }, + }); + httpReq.on('error', () => {}); + httpReq.write(reqBody); + httpReq.end(); + await promptStartedPromise; + // Destroy the client socket so `res.writableEnded` is true by + // the time the 50ms deadline fires. + httpReq.destroy(); + // Give the deadline timer time to fire + the route's catch + // block to handle the race. + await new Promise((r) => setTimeout(r, 200)); + expect(bridge.promptCalls).toHaveLength(1); + // The bridge's signal MUST still have been aborted with the + // typed reason โ€” the cleanup path still runs even though the + // response was already ended. + expect(bridge.promptCalls[0]?.signal?.aborted).toBe(true); + } finally { + await localHandle.close(); + } + }); + + it('fires the server-side deadline and returns 504 with errorKind', async () => { + // 50ms server deadline + a prompt that resolves only on abort: + // the deadline timer must abort the AbortController, the catch + // block must detect the typed reason, and the response must + // carry the structured `errorKind: 'prompt_deadline_exceeded'` + // (not a generic 500 / silent close). + const bridge = fakeBridge({ promptImpl: abortableBridgePromptImpl() }); + const app = createServeApp( + { ...baseOpts, promptDeadlineMs: 50 }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ prompt: [{ type: 'text', text: 'slow' }] }); + expect(res.status).toBe(504); + expect(res.body).toMatchObject({ + code: 'prompt_deadline_exceeded', + errorKind: 'prompt_deadline_exceeded', + deadlineMs: 50, + }); + // The bridge MUST have received an aborted signal so the agent + // can wind down its FIFO slot โ€” otherwise a buggy agent keeps + // the per-session lane blocked forever even though the HTTP + // client got its 504. + expect(bridge.promptCalls).toHaveLength(1); + expect(bridge.promptCalls[0]?.signal?.aborted).toBe(true); + expect(bridge.promptCalls[0]?.signal?.reason).toBeInstanceOf( + PromptDeadlineExceededError, + ); + }); + + it('still returns typed 504 when deadline stderr logging fails', async () => { + const stderrSpy = vi + .spyOn(process.stderr, 'write') + .mockImplementation(() => { + throw new Error('stderr pipe closed'); + }); + try { + const bridge = fakeBridge({ + promptImpl: () => new Promise(() => {}), + }); + const app = createServeApp( + { ...baseOpts, promptDeadlineMs: 50 }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ prompt: [{ type: 'text', text: 'slow' }] }); + expect(res.status).toBe(504); + expect(res.body.errorKind).toBe('prompt_deadline_exceeded'); + } finally { + stderrSpy.mockRestore(); + } + }); + + it('strips route-only deadlineMs before forwarding the prompt body', async () => { + const bridge = fakeBridge({ + promptImpl: async () => ({ stopReason: 'end_turn' }), + }); + const app = createServeApp( + { ...baseOpts, promptDeadlineMs: 5_000 }, + undefined, + { bridge }, + ); + const prompt = [{ type: 'text', text: 'hi' }]; + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + prompt, + deadlineMs: 1_000, + _meta: { trace: 'kept' }, + extra: 'kept', + }); + expect(res.status).toBe(200); + expect(bridge.promptCalls).toHaveLength(1); + expect(bridge.promptCalls[0]?.req).not.toHaveProperty('deadlineMs'); + expect(bridge.promptCalls[0]?.req).toMatchObject({ + sessionId: 'session-A', + prompt, + _meta: { trace: 'kept' }, + extra: 'kept', + }); + }); + + it('caps a per-prompt `deadlineMs` override at the server flag', async () => { + // Server flag 50ms, request asks for 5000ms โ€” the effective + // deadline must be the smaller 50ms. The way we observe it is + // the same 504-with-deadlineMs:50 response: if the cap was + // incorrectly the request's 5000ms, the test would time out + // long before completing. + const bridge = fakeBridge({ promptImpl: abortableBridgePromptImpl() }); + const app = createServeApp( + { ...baseOpts, promptDeadlineMs: 50 }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + prompt: [{ type: 'text', text: 'slow' }], + deadlineMs: 5_000, + }); + expect(res.status).toBe(504); + expect(res.body.deadlineMs).toBe(50); + }); + + it('uses the per-prompt override when shorter than the server flag', async () => { + // Server flag 10s, request 30ms โ€” request wins as the tighter + // bound. Same observability path as above; if the cap was the + // server's 10s the test would hang past its own short timeout. + const bridge = fakeBridge({ promptImpl: abortableBridgePromptImpl() }); + const app = createServeApp( + { ...baseOpts, promptDeadlineMs: 10_000 }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ + prompt: [{ type: 'text', text: 'slow' }], + deadlineMs: 30, + }); + expect(res.status).toBe(504); + expect(res.body.deadlineMs).toBe(30); + }); + + it('still emits 504 when the bridge IGNORES the abort signal (race contract)', async () => { + // The deadline must be a hard server-side guarantee, not contingent + // on the bridge / agent honoring AbortSignal. A buggy agent that + // never resolves its sendPrompt promise would, without the + // Promise.race in the prompt handler, keep the HTTP request open + // indefinitely and never emit the promised 504 โ€” that was the + // Copilot finding on the initial T2.9 commit. This test exercises + // a non-cooperative bridge to lock the contract: deadline 50ms, + // bridge promise never settles, route still returns 504 within a + // reasonable budget. + const bridge = fakeBridge({ + promptImpl: () => new Promise(() => {}), + }); + const app = createServeApp( + { ...baseOpts, promptDeadlineMs: 50 }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ prompt: [{ type: 'text', text: 'slow' }] }); + expect(res.status).toBe(504); + expect(res.body).toMatchObject({ + code: 'prompt_deadline_exceeded', + errorKind: 'prompt_deadline_exceeded', + deadlineMs: 50, + }); + // The signal was still aborted with the typed reason as best- + // effort wind-down โ€” the agent has every chance to clean up + // its FIFO slot even though we no longer wait for it. + expect(bridge.promptCalls[0]?.signal?.aborted).toBe(true); + expect(bridge.promptCalls[0]?.signal?.reason).toBeInstanceOf( + PromptDeadlineExceededError, + ); + }); + + it('does not interfere with normal prompt completion when the flag is unset', async () => { + // The deadline path must be 100% off by default. A 200 OK with + // the bridge's stopReason is the bit-for-bit pre-PR contract. + const bridge = fakeBridge({ + promptImpl: async () => ({ stopReason: 'end_turn' }), + }); + const app = createServeApp(baseOpts, undefined, { bridge }); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ prompt: [{ type: 'text', text: 'hi' }] }); + expect(res.status).toBe(200); + expect(res.body.stopReason).toBe('end_turn'); + }); + + it('does not fire the deadline when the prompt resolves promptly', async () => { + // 5s deadline + an immediate resolve: the timer must not fire + // and must not corrupt the 200 response. Guards against a + // future regression where the timer races the response. + const bridge = fakeBridge({ + promptImpl: async () => ({ stopReason: 'end_turn' }), + }); + const app = createServeApp( + { ...baseOpts, promptDeadlineMs: 5_000 }, + undefined, + { bridge }, + ); + const res = await request(app) + .post('/session/session-A/prompt') + .set('Host', `127.0.0.1:${baseOpts.port}`) + .send({ prompt: [{ type: 'text', text: 'hi' }] }); + expect(res.status).toBe(200); + expect(res.body.stopReason).toBe('end_turn'); + }); + }); + + describe('GET /capabilities', () => { + it('omits `prompt_absolute_deadline` by default', async () => { + const app = createServeApp(baseOpts); + const res = await request(app) + .get('/capabilities') + .set('Host', `127.0.0.1:${baseOpts.port}`); + expect(res.status).toBe(200); + expect(res.body.features).not.toContain('prompt_absolute_deadline'); + }); + + it('advertises `prompt_absolute_deadline` when the flag is set', async () => { + const app = createServeApp({ ...baseOpts, promptDeadlineMs: 5_000 }); + const res = await request(app) + .get('/capabilities') + .set('Host', `127.0.0.1:${baseOpts.port}`); + expect(res.status).toBe(200); + expect(res.body.features).toContain('prompt_absolute_deadline'); + }); + + it('omits `writer_idle_timeout` by default', async () => { + const app = createServeApp(baseOpts); + const res = await request(app) + .get('/capabilities') + .set('Host', `127.0.0.1:${baseOpts.port}`); + expect(res.status).toBe(200); + expect(res.body.features).not.toContain('writer_idle_timeout'); + }); + + it('advertises `writer_idle_timeout` when the flag is set', async () => { + const app = createServeApp({ + ...baseOpts, + writerIdleTimeoutMs: 60_000, + }); + const res = await request(app) + .get('/capabilities') + .set('Host', `127.0.0.1:${baseOpts.port}`); + expect(res.status).toBe(200); + expect(res.body.features).toContain('writer_idle_timeout'); + }); + }); +}); + +describe('T2.9 SSE writer idle timeout (issue #4514)', () => { + let handle: RunHandle | undefined; + afterEach(async () => { + if (handle) { + await handle.close(); + handle = undefined; + } + }); + + it('evicts an idle SSE writer with a terminal client_evicted frame', async () => { + // Bridge yields nothing, ever โ€” simulating an idle stream where + // the only writes the timer would observe are the SSE handshake + // (`retry: 3000`) and (eventually) the 15s heartbeat. With a + // 200ms idle deadline the timer must fire well before the + // heartbeat refreshes `lastWriteAt`. Expected terminal frame: + // `client_evicted` with the new `reason: 'writer_idle_timeout'`. + const bridge = fakeBridge({ + // eslint-disable-next-line require-yield + async *subscribeImpl(_sessionId, _opts) { + // Park forever; the test triggers eviction via the timer. + // No yield: the test asserts that the daemon's idle-timeout + // path fires even when the bridge produces zero frames. + await new Promise(() => {}); + }, + }); + handle = await runQwenServe( + { + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + writerIdleTimeoutMs: 200, + }, + { bridge }, + ); + const port = (handle.server.address() as { port: number }).port; + + const res = await fetch(`http://127.0.0.1:${port}/session/sess-A/events`); + expect(res.status).toBe(200); + + // Read until we see the eviction frame OR the stream closes. The + // 1500ms budget is well below the 15s heartbeat so a regression + // that disables the idle timer would still fail loudly here. + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + let evictedData: unknown; + const deadline = Date.now() + 1_500; + while (Date.now() < deadline) { + const { value, done } = await reader.read(); + if (done) break; + buf += decoder.decode(value, { stream: true }); + let idx: number; + while ((idx = buf.indexOf('\n\n')) !== -1) { + const raw = buf.slice(0, idx); + buf = buf.slice(idx + 2); + if (!raw || raw.startsWith(':') || raw.startsWith('retry:')) continue; + // Parse the frame; look for event: client_evicted. + let eventName = ''; + let dataLine = ''; + for (const line of raw.split('\n')) { + if (line.startsWith('event: ')) eventName = line.slice(7); + else if (line.startsWith('data: ')) dataLine = line.slice(6); + } + if (eventName === 'client_evicted') { + evictedData = JSON.parse(dataLine); + break; + } + } + if (evictedData !== undefined) break; + } + await reader.cancel().catch(() => undefined); + + expect(evictedData).toBeDefined(); + expect(evictedData).toMatchObject({ + v: 1, + type: 'client_evicted', + data: { + reason: 'writer_idle_timeout', + errorKind: 'writer_idle_timeout', + timeoutMs: 200, + }, + }); + }); + + it('does not evict when the writer idle timeout is unset (legacy contract)', async () => { + // Without `writerIdleTimeoutMs`, the existing 15s-heartbeat-only + // behavior must be preserved bit-for-bit. We open a stream, read + // a real event, then wait ~600ms โ€” no client_evicted frame may + // appear in that window. + const bridge = fakeBridge({ + async *subscribeImpl(_sessionId, _opts) { + yield { id: 1, v: 1, type: 'session_update', data: { ok: true } }; + await new Promise(() => {}); + }, + }); + handle = await runQwenServe( + { hostname: '127.0.0.1', port: 0, mode: 'http-bridge' }, + { bridge }, + ); + const port = (handle.server.address() as { port: number }).port; + + const res = await fetch(`http://127.0.0.1:${port}/session/sess-A/events`); + expect(res.status).toBe(200); + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + let sawFirstEvent = false; + let sawEviction = false; + const deadline = Date.now() + 600; + while (Date.now() < deadline) { + const readPromise = reader.read(); + const wakeup = new Promise<{ value: undefined; done: false }>((r) => { + setTimeout( + () => r({ value: undefined, done: false }), + deadline - Date.now() + 10, + ); + }); + const { value, done } = (await Promise.race([readPromise, wakeup])) as { + value: Uint8Array | undefined; + done: boolean; + }; + if (done) break; + if (value === undefined) break; + buf += decoder.decode(value, { stream: true }); + let idx: number; + while ((idx = buf.indexOf('\n\n')) !== -1) { + const raw = buf.slice(0, idx); + buf = buf.slice(idx + 2); + if (!raw || raw.startsWith(':') || raw.startsWith('retry:')) continue; + let eventName = ''; + for (const line of raw.split('\n')) { + if (line.startsWith('event: ')) eventName = line.slice(7); + } + if (eventName === 'session_update') sawFirstEvent = true; + if (eventName === 'client_evicted') sawEviction = true; + } + } + await reader.cancel().catch(() => undefined); + + expect(sawFirstEvent).toBe(true); + expect(sawEviction).toBe(false); + }); + + it('does NOT evict when active writes keep refreshing lastWriteAt (#4514 T2.9 wenshao review)', async () => { + // wenshao flagged that the existing "fires when idle" + "doesn't + // fire when unset" tests don't cover the case where REAL writes + // (event yields, not just heartbeats) refresh `lastWriteAt` + // inside `doWrite`. With idle timeout = 300ms and an event every + // 100ms, the timer should keep deferring โ€” the connection stays + // alive even past several timeout cycles. + const bridge = fakeBridge({ + async *subscribeImpl(_sessionId, _opts) { + // Yield 5 events at ~100ms intervals โ€” well below the 300ms + // idle budget โ€” then park forever. We expect no eviction in + // the read window. + for (let i = 1; i <= 5; i++) { + await new Promise((r) => setTimeout(r, 100)); + yield { + id: i, + v: 1, + type: 'session_update', + data: { tick: i }, + }; + } + await new Promise(() => {}); + }, + }); + handle = await runQwenServe( + { + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + writerIdleTimeoutMs: 300, + }, + { bridge }, + ); + const port = (handle.server.address() as { port: number }).port; + const res = await fetch(`http://127.0.0.1:${port}/session/sess-A/events`); + expect(res.status).toBe(200); + + // Read for ~700ms โ€” long enough that an IDLE writer would have + // been evicted twice over (300ms timeout, polled every ~250ms), + // but the per-100ms event stream refreshes lastWriteAt before the + // check fires. + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + let sawEviction = false; + let sessionUpdates = 0; + const deadline = Date.now() + 700; + while (Date.now() < deadline) { + const readPromise = reader.read(); + const wakeup = new Promise<{ value: undefined; done: false }>((r) => { + setTimeout( + () => r({ value: undefined, done: false }), + Math.max(0, deadline - Date.now() + 10), + ); + }); + const { value, done } = (await Promise.race([readPromise, wakeup])) as { + value: Uint8Array | undefined; + done: boolean; + }; + if (done) break; + if (value === undefined) break; + buf += decoder.decode(value, { stream: true }); + let idx: number; + while ((idx = buf.indexOf('\n\n')) !== -1) { + const raw = buf.slice(0, idx); + buf = buf.slice(idx + 2); + if (!raw || raw.startsWith(':') || raw.startsWith('retry:')) continue; + for (const line of raw.split('\n')) { + if (line.startsWith('event: session_update')) sessionUpdates += 1; + if (line.startsWith('event: client_evicted')) sawEviction = true; + } + } + } + await reader.cancel().catch(() => undefined); + + expect(sessionUpdates).toBeGreaterThanOrEqual(3); + expect(sawEviction).toBe(false); + }); + + it('does NOT evict when a back-pressured write drains within the idle budget', async () => { + const http = await import('node:http'); + type WriteCallback = (error?: Error | null) => void; + const originalWrite = http.ServerResponse.prototype.write as unknown as ( + this: ServerResponse, + chunk: string | Uint8Array, + encodingOrCb?: BufferEncoding | WriteCallback, + cb?: WriteCallback, + ) => boolean; + const writeSpy = vi.spyOn(http.ServerResponse.prototype, 'write'); + let forcedBackpressure = false; + writeSpy.mockImplementation(function ( + this: ServerResponse, + chunk: string | Uint8Array, + encodingOrCb?: BufferEncoding | WriteCallback, + cb?: WriteCallback, + ): boolean { + const wrote = + typeof encodingOrCb === 'function' + ? originalWrite.call(this, chunk, encodingOrCb) + : originalWrite.call(this, chunk, encodingOrCb, cb); + const text = typeof chunk === 'string' ? chunk : chunk.toString(); + if (!forcedBackpressure && text.includes('event: session_update')) { + forcedBackpressure = true; + setTimeout(() => this.emit('drain'), 150); + return false; + } + return wrote; + }); + + try { + const bridge = fakeBridge({ + async *subscribeImpl(_sessionId, _opts) { + yield { + id: 1, + v: 1, + type: 'session_update', + data: { tick: 1 }, + }; + await new Promise(() => {}); + }, + }); + handle = await runQwenServe( + { + hostname: '127.0.0.1', + port: 0, + mode: 'http-bridge', + writerIdleTimeoutMs: 200, + }, + { bridge }, + ); + const port = (handle.server.address() as { port: number }).port; + const res = await fetch(`http://127.0.0.1:${port}/session/sess-A/events`); + expect(res.status).toBe(200); + + const reader = res.body!.getReader(); + const decoder = new TextDecoder(); + let buf = ''; + let sawSessionUpdate = false; + let sawEviction = false; + const deadline = Date.now() + 350; + while (Date.now() < deadline) { + const readPromise = reader.read(); + const wakeup = new Promise<{ value: undefined; done: false }>((r) => { + setTimeout( + () => r({ value: undefined, done: false }), + Math.max(0, deadline - Date.now() + 10), + ); + }); + const { value, done } = (await Promise.race([readPromise, wakeup])) as { + value: Uint8Array | undefined; + done: boolean; + }; + if (done) break; + if (value === undefined) break; + buf += decoder.decode(value, { stream: true }); + let idx: number; + while ((idx = buf.indexOf('\n\n')) !== -1) { + const raw = buf.slice(0, idx); + buf = buf.slice(idx + 2); + if (!raw || raw.startsWith(':') || raw.startsWith('retry:')) continue; + for (const line of raw.split('\n')) { + if (line.startsWith('event: session_update')) { + sawSessionUpdate = true; + } + if (line.startsWith('event: client_evicted')) sawEviction = true; + } + } + } + await reader.cancel().catch(() => undefined); + + expect(forcedBackpressure).toBe(true); + expect(sawSessionUpdate).toBe(true); + expect(sawEviction).toBe(false); + } finally { + writeSpy.mockRestore(); + } + }); +}); + +describe('T2.9 serve-side errorKind taxonomy (issue #4514)', () => { + it('publishes the new error kinds in SERVE_ERROR_KINDS', async () => { + // Lock the serve-side taxonomy contains both T2.9 kinds. The + // mirrored SDK assertion lives in + // `packages/sdk-typescript/test/unit/daemon-public-surface.test.ts` + // (different package, no cross-package import). Together they + // guarantee a PR adding a kind on one side without the other + // fails CI. + const { SERVE_ERROR_KINDS } = await import('@qwen-code/acp-bridge/status'); + expect(SERVE_ERROR_KINDS).toContain('prompt_deadline_exceeded'); + expect(SERVE_ERROR_KINDS).toContain('writer_idle_timeout'); + }); +}); diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 173c14964b4..b414717eb8e 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -214,6 +214,79 @@ export interface ServeAppDeps { deviceFlowProviders?: DeviceFlowProvider[]; } +/** + * Issue #4514 T2.9. Sentinel passed as `AbortController.abort(reason)` + * when a prompt exceeds its server-configured wallclock. The catch + * block on `POST /session/:id/prompt` distinguishes + * `instanceof PromptDeadlineExceededError` (โ†’ HTTP 504 with the typed + * `errorKind`) from the existing client-disconnect path (โ†’ swallow, + * socket is gone). Exported so tests can match on the class identity + * without parsing the response body for the kind string. + */ +export class PromptDeadlineExceededError extends Error { + readonly deadlineMs: number; + constructor(deadlineMs: number) { + super(`prompt exceeded the ${deadlineMs}ms deadline`); + this.name = 'PromptDeadlineExceededError'; + this.deadlineMs = deadlineMs; + } +} + +/** + * Issue #4514 T2.9. Resolve the effective per-prompt wallclock from + * the server flag + an optional request body override. Returns + * `undefined` when no deadline applies (server flag is unset). + * When the server flag is set, the request override may SHORTEN the + * deadline but never EXTEND it โ€” operators stay the upper bound. + * + * Exported (named) for the unit test that asserts the capping + * contract without spinning up an HTTP listener. + */ +export function resolvePromptDeadlineMs( + serverMs: number | undefined, + requestMs: number | undefined, +): number | undefined { + if (serverMs === undefined || !Number.isFinite(serverMs) || serverMs <= 0) { + return undefined; + } + if ( + requestMs === undefined || + !Number.isFinite(requestMs) || + requestMs <= 0 + ) { + return serverMs; + } + return Math.min(serverMs, requestMs); +} + +/** + * Issue #4514 T2.9. Single source of truth for the prompt-deadline 504 + * response: log the operator-facing stderr breadcrumb (so a 504 spike + * at the load balancer can be grepped back to a session) and emit the + * typed JSON body. Keeping the wire format and log line together + * prevents drift between future prompt-deadline response paths. + */ +function emitPromptDeadline504( + res: import('express').Response, + err: PromptDeadlineExceededError, + sessionId: string, +): void { + try { + writeStderrLine( + `qwen serve: prompt deadline fired (session ${sessionId}) โ€” ` + + `deadlineMs=${err.deadlineMs}`, + ); + } catch { + /* stderr pipe closed; 504 response still going out. */ + } + res.status(504).json({ + error: err.message, + code: 'prompt_deadline_exceeded', + errorKind: 'prompt_deadline_exceeded', + deadlineMs: err.deadlineMs, + }); +} + /** * Build the Express app for `qwen serve`. Pure function โ€” no side effects on * the network or process; `runQwenServe` does the listen/signal handling. @@ -671,19 +744,17 @@ export function createServeApp( // ONLY when the operator opted in. Tag presence = behavior is // on; older daemons without this PR omit the tag and SDKs that // post-PR feature-detect on it stay backward compatible. - // - // F2 (#4175 commit 5): `mcpPoolActive` advertises - // `mcp_workspace_pool` + `mcp_pool_restart` together. Defaults - // to `true` when omitted so daemons that don't explicitly set - // the option still advertise the F2 surface; operators flip it - // to `false` only when `QWEN_SERVE_NO_MCP_POOL=1` is in scope. features: getAdvertisedServeFeatures(undefined, { requireAuth: opts.requireAuth === true, mcpPoolActive: opts.mcpPoolActive !== false, - // T2.4 (issue #4514): advertise `allow_origin` iff the daemon - // was booted with at least one `--allow-origin` pattern. allowOriginActive: opts.allowOrigins !== undefined && opts.allowOrigins.length > 0, + ...(opts.promptDeadlineMs !== undefined + ? { promptDeadlineMs: opts.promptDeadlineMs } + : {}), + ...(opts.writerIdleTimeoutMs !== undefined + ? { writerIdleTimeoutMs: opts.writerIdleTimeoutMs } + : {}), }), modelServices: [], // #3803 ยง02: surface the bound workspace so clients can detect @@ -1261,6 +1332,29 @@ export function createServeApp( }); return; } + // T2.9: validate the optional per-prompt `deadlineMs` override BEFORE + // we touch the abort controller โ€” a malformed value is operator- + // visible client error (400) rather than silently dropped (which + // would let the client believe their deadline was active when it + // wasn't). Capping vs the server flag happens later, after we + // know what the server is willing to enforce. + const rawRequestDeadline = body['deadlineMs']; + let requestDeadlineMs: number | undefined; + if (rawRequestDeadline !== undefined && rawRequestDeadline !== null) { + if ( + typeof rawRequestDeadline !== 'number' || + !Number.isFinite(rawRequestDeadline) || + !Number.isInteger(rawRequestDeadline) || + rawRequestDeadline <= 0 + ) { + res.status(400).json({ + error: '`deadlineMs` must be a positive integer (milliseconds)', + code: 'invalid_deadline_ms', + }); + return; + } + requestDeadlineMs = rawRequestDeadline; + } // Propagate HTTP-client disconnect to an ACP cancel notification so // the agent winds down promptly and the per-session FIFO doesn't // stay blocked on a dead client. Detached after the prompt settles. @@ -1278,35 +1372,97 @@ export function createServeApp( if (!res.writableEnded) abort.abort(); }; res.once('close', onResClose); + // T2.9: arm the server-side wallclock deadline (if configured). + // `resolvePromptDeadlineMs` returns `undefined` when the server + // flag is unset, preserving the legacy "client disconnect is + // the only auto-cancel" behavior bit-for-bit. When a deadline IS + // configured we Promise.race the bridge call against an explicit + // rejecting timer โ€” without the race, a non-cooperative agent + // that ignores AbortSignal could keep the HTTP request open + // indefinitely (the FIXME in `httpAcpBridge.ts` `sendPrompt` was + // promised closed by T2.9 in the PR description; relying on the + // bridge alone wouldn't deliver). The race makes the 504 a hard + // guarantee independent of bridge cooperation; `abort.abort` is + // still called as best-effort wind-down so the agent's FIFO slot + // is freed if it does honor the signal. + const effectiveDeadlineMs = resolvePromptDeadlineMs( + opts.promptDeadlineMs, + requestDeadlineMs, + ); + const forwardedBody = { ...body }; + delete forwardedBody['deadlineMs']; + let deadlineTimer: NodeJS.Timeout | undefined; + const deadlinePromise: Promise | undefined = + effectiveDeadlineMs !== undefined + ? new Promise((_, reject) => { + deadlineTimer = setTimeout(() => { + const err = new PromptDeadlineExceededError(effectiveDeadlineMs); + // Reject FIRST so Promise.race resolves deterministically + // with the typed deadline error; the bridge's own + // AbortError rejection (if the agent honors abort) + // would otherwise race the route's microtask queue and + // surface as a generic AbortError in the catch path. + reject(err); + if (!abort.signal.aborted) abort.abort(err); + }, effectiveDeadlineMs); + // unref so a still-armed timer can't keep the daemon alive + // past shutdown. + deadlineTimer.unref(); + }) + : undefined; const clientId = parseClientIdHeader(req, res); if (clientId === null) { res.off('close', onResClose); + if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); return; } try { - // SECURITY NOTE: this `...(body as object)` passthrough is + // SECURITY NOTE: this `...forwardedBody` passthrough is // intentional โ€” the bridge / ACP SDK ignores fields it // doesn't recognize (ACP-spec `_meta` etc are forwarded // wholesale to the agent, which is the documented behavior). - // `sessionId` and `prompt` are forced to the route's view to - // prevent body-spoofing of the routing key. If a future - // bridge version starts trusting an additional field by name, - // that field becomes a client-controlled input surface โ€” at - // that point switch this to an explicit pick. The same - // pattern repeats on cancel / model below; review them all - // together when adding new bridge-trusted fields. - const result = await bridge.sendPrompt( + // `sessionId`, `prompt`, and the route-only `deadlineMs` are + // forced/stripped so the child never sees uncapped client input. + const bridgePromise = bridge.sendPrompt( sessionId, { - ...(body as object), + ...forwardedBody, sessionId, prompt, } as Parameters[1], abort.signal, clientId !== undefined ? { clientId } : undefined, ); + // T2.9: when the deadline race fires first, the underlying + // bridge promise becomes an orphan that may eventually settle + // minutes later (especially against a buggy agent). Tail-attach + // a no-op handler so its eventual rejection doesn't surface as + // an unhandledRejection โ€” the 504 has already been sent and the + // route has no further use for the result. + if (deadlinePromise !== undefined) { + bridgePromise.catch(() => undefined); + } + const result = await (deadlinePromise !== undefined + ? Promise.race([bridgePromise, deadlinePromise]) + : bridgePromise); res.status(200).json(result); } catch (err) { + // T2.9: the deadline race won โ€” emit the typed 504 directly. + // This is the primary deadline-exceeded path now that the + // route races the bridge against its own timer. + // + // The `return` MUST fire on every `PromptDeadlineExceededError`, + // including the writableEnded race (client disconnected in the + // same tick the deadline timer fired). Without the early return, + // the typed error would fall through to the AbortError branch + // (false โ€” not a DOMException), then `sendBridgeError`, which + // would call `res.status(500).json(...)` on an already-ended + // response and trip `ERR_STREAM_WRITE_AFTER_END`. wenshao + // review #4530 inline #3 (Critical). + if (err instanceof PromptDeadlineExceededError) { + if (!res.writableEnded) emitPromptDeadline504(res, err, sessionId); + return; + } // The HTTP client disconnecting fires the abort path above and // the bridge re-throws as `AbortError`. That's a normal // wind-down, not an error worth a 500 + stderr stack trace. @@ -1336,6 +1492,7 @@ export function createServeApp( }); } finally { res.off('close', onResClose); + if (deadlineTimer !== undefined) clearTimeout(deadlineTimer); } }); @@ -1935,6 +2092,21 @@ export function createServeApp( // SSE frames on the wire. The chain is single-flight: each call // waits for the previous write to settle before scheduling its own. let writeChain: Promise = Promise.resolve(); + // T2.9: epoch (ms) of the last write that fully resolved โ€” either + // synchronous `res.write` returned `true`, or the async `drain` + // fired. The idle-timeout interval below compares + // `Date.now() - lastWriteAt` against the configured budget; a + // writer that stalls indefinitely on `drain` will never refresh + // this stamp, so the timer fires and forces cleanup. Initialized + // to "now" because cleanup runs only after the FIRST stall, and + // the SSE handshake itself counts as activity. + // + // Gated on `trackWriterIdle` so the default (flag unset) avoids + // a per-chunk `Date.now()` on a chatty stream โ€” SSE writers can + // be in the hundreds-to-thousands of frames per session. + const trackWriterIdle = + opts.writerIdleTimeoutMs !== undefined && opts.writerIdleTimeoutMs > 0; + let lastWriteAt = trackWriterIdle ? Date.now() : 0; const doWrite = (chunk: string): Promise => new Promise((resolve, reject) => { if (res.writableEnded) { @@ -1957,12 +2129,14 @@ export function createServeApp( return; } if (ok) { + if (trackWriterIdle) lastWriteAt = Date.now(); resolve(); return; } const onDrain = () => { res.off('close', onClose); res.off('error', onError); + if (trackWriterIdle) lastWriteAt = Date.now(); resolve(); }; const onClose = () => { @@ -2001,13 +2175,16 @@ export function createServeApp( // notice a dead client through write-back-pressure. Comment frame is // ignored by EventSource. // - // KNOWN GAP: this only catches dead connections via write - // back-pressure on heartbeat itself. A network partition without TCP - // RST can leave the connection looking alive (no FIN received) for - // however long Node's keepalive probes take to time out โ€” usually - // ~2 hours by default, configurable via `server.keepAliveTimeout`. - // Stage 2 may add an explicit application-level idle timeout - // (last-byte-written tracking + per-connection deadline). + // T2.9 (issue #4514): the 15s heartbeat detects a TCP-dead writer + // via `drain` back-pressure on the comment frame itself. The + // `--writer-idle-timeout-ms` flag below adds the orthogonal + // application-level guard: if the LAST SUCCESSFUL FLUSH (any + // write โ€” heartbeat, replay frame, live event) is older than the + // configured budget, the writer is considered stuck (NAT silently + // dropping flows, peer process frozen, etc.) and we force a + // terminal `client_evicted` frame + cleanup. The historical "Stage + // 2 may add an explicit application-level idle timeout" gap + // referenced here is now closed when the flag is set. const heartbeatTimer = setInterval(() => { if (!res.writableEnded) { // Heartbeat writes are best-effort; failure swallowed via the @@ -2017,10 +2194,94 @@ export function createServeApp( }, 15_000); heartbeatTimer.unref(); + // T2.9: declare the idle-timer slot up-front so `cleanup` below can + // clear it unconditionally. The actual interval is armed only when + // `--writer-idle-timeout-ms` is configured. + let idleTimer: NodeJS.Timeout | undefined; + const cleanup = () => { clearInterval(heartbeatTimer); + if (idleTimer !== undefined) clearInterval(idleTimer); abort.abort(); }; + + // T2.9: arm the SSE writer idle timeout (if configured). Distinct + // from the heartbeat above: heartbeat = "try to ping every 15s"; + // this = "if no write SUCCEEDED for N ms, force-evict." Values + // BELOW the 15s heartbeat interval WILL evict otherwise-healthy + // idle connections before the first heartbeat fires โ€” they're not + // a no-op. Production deployments should pick a value comfortably + // above 15s (e.g. 30000โ€“300000ms) so legitimate idle streams stay + // alive and only genuinely stuck writers are reaped; small values + // are useful for tests / short-lived dev sessions. The interval + // polls at 1/4 the budget (bounded by [250ms, 5s]) so tests + // using short budgets still detect promptly, while long + // production budgets stay cheap. Values below roughly 1000ms all + // use the 250ms polling floor, so eviction can lag until the next + // tick instead of landing at exact millisecond precision. + if (trackWriterIdle) { + // Narrowed by `trackWriterIdle`; the const assertion keeps + // TypeScript happy inside the closure without re-reading opts. + const writerIdleTimeoutMs = opts.writerIdleTimeoutMs as number; + const checkIntervalMs = Math.max( + 250, + Math.min(5_000, Math.floor(writerIdleTimeoutMs / 4)), + ); + idleTimer = setInterval(() => { + if (res.writableEnded) return; + const idleForMs = Date.now() - lastWriteAt; + if (idleForMs < writerIdleTimeoutMs) return; + // Reuse the existing `client_evicted` taxonomy from + // `eventBus.ts` so SDK reducers branch on the same frame type + // they already handle for queue-overflow eviction; the new + // `reason` slot is the differentiator. Write DIRECTLY here + // (bypassing `writeWithBackpressure`) because the chain may + // already be stuck on a `drain` that will never come โ€” which + // is the exact scenario this timer exists to catch. If the + // kernel send buffer has room the client sees the frame; if + // not, the client gets EPIPE on next read. Either way the + // socket is closed in the next two statements, so any drop + // is bounded. + try { + res.write( + formatSseFrame({ + v: 1, + type: 'client_evicted', + data: { + reason: 'writer_idle_timeout', + errorKind: 'writer_idle_timeout', + idleForMs, + timeoutMs: writerIdleTimeoutMs, + }, + }), + ); + } catch { + /* socket already destroyed; nothing to send. */ + } + // wenshao review #4530 inline #2: wrap stderr + res.end so an + // EPIPE on the stderr pipe (or a synchronous throw from + // `res.end()` against a destroyed socket) can't escape this + // interval callback. If it did, `cleanup()` wouldn't run, the + // heartbeat + idle timers would never clear, and every + // subsequent tick would re-throw โ€” turning one transient + // failure into a permanent uncaughtException loop. + try { + writeStderrLine( + `qwen serve: evicting SSE client (session ${sessionId}) โ€” ` + + `writer idle for ${idleForMs}ms > ${writerIdleTimeoutMs}ms timeout`, + ); + } catch { + /* stderr pipe closed; eviction is still happening. */ + } + cleanup(); + try { + if (!res.writableEnded) res.end(); + } catch { + /* socket already destroyed; nothing more to do. */ + } + }, checkIntervalMs); + idleTimer.unref(); + } req.on('close', cleanup); // Swallow socket-level write errors. When the underlying TCP connection // dies (RST, mid-flight kill -9), the next `res.write` throws EPIPE. diff --git a/packages/cli/src/serve/types.ts b/packages/cli/src/serve/types.ts index 03e807e57c5..385ed33b90f 100644 --- a/packages/cli/src/serve/types.ts +++ b/packages/cli/src/serve/types.ts @@ -142,41 +142,22 @@ export interface ServeOptions { /** * F2 (#4175 commit 5). Whether the daemon advertises the * `mcp_workspace_pool` + `mcp_pool_restart` capability tags. - * Defaults to `true` (the F2 pool is always on except under the - * env-var kill switch). Operators set this to `false` when - * `QWEN_SERVE_NO_MCP_POOL=1` is in scope so SDK clients pre-flighting - * on the tags don't speculatively send `?entryIndex=` queries the - * legacy single-entry path can't honor. The tag advertisement is - * orthogonal to ACP child behavior โ€” the child reads its own env - * var copy independently โ€” but they're driven from the same source - * via `runQwenServe.ts`. */ mcpPoolActive?: boolean; /** * T2.4 (issue #4514). Cross-origin allowlist for browser webui - * deployments. Each entry is either the `*` literal (any origin - * allowed, advertised loudly in the boot breadcrumb) or a canonical - * URL origin (`://[:]`, no trailing slash / path / - * userinfo / query). When at least one pattern is configured, the - * daemon installs `allowOriginCors` instead of `denyBrowserOriginCors` - * โ€” matched cross-origin requests get proper CORS response headers - * (`Access-Control-Allow-Origin: `, `Vary: Origin`, - * standard methods / headers / max-age, exposed `Retry-After`), - * unmatched cross-origin requests still get a 403 with the same error - * envelope as today. - * - * Empty / undefined preserves the default wall (any `Origin` header โ†’ - * 403 from `denyBrowserOriginCors`). Boot validates each entry through - * `parseAllowOriginPatterns` in - * `packages/cli/src/serve/auth.ts`; malformed entries throw - * `InvalidAllowOriginPatternError` and refuse to start. - * - * Loopback self-hits are unaffected โ€” the demo-page Origin-strip - * shim (`server.ts` near `cachedSelfOrigins`) runs first and removes - * the `Origin` header for self-loopback addresses, so neither the - * old wall nor the new allowlist needs to know about them. + * deployments. */ allowOrigins?: string[]; + /** + * Issue #4514 T2.9. Server-side wallclock cap on a single + * `POST /session/:id/prompt` from receipt to completion. + */ + promptDeadlineMs?: number; + /** + * Issue #4514 T2.9. Per-SSE-connection idle deadline. + */ + writerIdleTimeoutMs?: number; } /** diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 26c5d441c52..6aa8c8f40db 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -227,6 +227,20 @@ export interface PromptRequest { prompt: PromptContentBlock[]; /** Optional ACP _meta passthrough. */ _meta?: Record | null; + /** + * Issue #4514 T2.9. Per-prompt wallclock cap (positive integer ms). + * The effective deadline is `min(server flag, this)` โ€” the request + * can shorten, never extend. When omitted, the server's + * `--prompt-deadline-ms` flag governs alone (unlimited when both + * are unset). On expiry the daemon returns 504 + + * `errorKind: 'prompt_deadline_exceeded'`. + * + * Daemons without T2.9 (no `prompt_absolute_deadline` capability + * tag) silently ignore the field โ€” pre-flight + * `caps.features.includes('prompt_absolute_deadline')` before + * relying on it. + */ + deadlineMs?: number; [key: string]: unknown; } diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index caccc949890..aa20792fc25 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -193,6 +193,17 @@ export const DAEMON_ERROR_KINDS = [ // Issue #4175 PR 14: budget refusal under `--mcp-budget-mode=enforce`. // Mirrors the serve-side `SERVE_ERROR_KINDS` addition. 'budget_exhausted', + // Issue #4514 T2.9: a prompt exceeded the daemon-configured wallclock + // cap (or the request's own `deadlineMs`, capped at the server flag). + // Surfaced on the `POST /session/:id/prompt` 504 response. Mirrors + // the serve-side `SERVE_ERROR_KINDS` addition. + 'prompt_deadline_exceeded', + // Issue #4514 T2.9: an SSE writer's last successful flush was older + // than the daemon's writer-idle deadline. Daemon emits a terminal + // `client_evicted` frame with `reason: 'writer_idle_timeout'`; the + // kind appears on that frame's `errorKind` field. Mirrors the + // serve-side `SERVE_ERROR_KINDS` addition. + 'writer_idle_timeout', ] as const; export type DaemonErrorKind = (typeof DAEMON_ERROR_KINDS)[number]; diff --git a/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts b/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts index 46ede5bedb9..7d8067e8d5e 100644 --- a/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts +++ b/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts @@ -126,4 +126,17 @@ describe('public SDK entry โ€” typed daemon event surface (#4217)', () => { expect(typeof Public.createDaemonAuthState).toBe('function'); expect(typeof Public.DEVICE_FLOW_EXPIRY_GRACE_MS).toBe('number'); }); + + it('mirrors the T2.9 errorKind additions in DAEMON_ERROR_KINDS (issue #4514)', () => { + // The SDK-side `DAEMON_ERROR_KINDS` is hand-mirrored from the + // serve-side `SERVE_ERROR_KINDS` in `acp-bridge/src/status.ts`. + // T2.9 added two kinds (`prompt_deadline_exceeded` for the + // POST /session/:id/prompt 504, `writer_idle_timeout` for the + // terminal SSE client_evicted frame). Lock them so a future PR + // that bumps the serve list without touching the SDK list fails + // here instead of shipping a typed-on-server-but-unknown-on-SDK + // mismatch. + expect(Public.DAEMON_ERROR_KINDS).toContain('prompt_deadline_exceeded'); + expect(Public.DAEMON_ERROR_KINDS).toContain('writer_idle_timeout'); + }); });