From 033f1ca2f9f3ac546e9d533d728feafecdc33669 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Sun, 17 May 2026 17:10:25 +0800 Subject: [PATCH 1/5] feat(serve): SSE replay sizing + slow_client_warning backpressure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit #4175 Wave 2.5 PR 10. Closes the SSE replay / backpressure knobs called out in #3803 §02 so chatty Stage 1 sessions get an honest reconnect window and operators get a heads-up signal before clients are summarily evicted. - **`DEFAULT_RING_SIZE` 4000 → 8000.** Per-session replay ring depth now matches the #3803 §02 target for chatty sessions. - **`--event-ring-size `** CLI flag (default 8000) lets operators tune the ring per daemon. Threaded `ServeOptions` → `BridgeOptions.eventRingSize` → both `new EventBus()` construction sites (fresh sessions + restore path). Validation is fail-CLOSED (positive finite integer; 0 / NaN / negative throw at boot). - **`slow_client_warning` SSE frame.** When a subscriber's queue crosses 75% full the bus force-pushes a synthetic `slow_client_warning` to that subscriber once per overflow episode, carrying `{queueSize, maxQueued, lastEventId}`. The flag re-arms after the queue drains below 37.5% (hysteresis, no flap near threshold). If the queue actually overflows after the warning, the existing `client_evicted` terminal frame path still fires. Like `client_evicted`, the warning has no `id` (synthetic frame; must not burn a sequence slot for other subscribers). - **`?maxQueued=N`** query param on `GET /session/:id/events` (range `[16, 2048]`, default 256). Lets cold reconnect clients pre-size their per-subscriber backlog so a large `Last-Event-ID: 0` replay doesn't trip the warning on the first publish. Range rationale: lower bound 16 (smaller is useless for any replay); upper bound 2048 (so a single subscriber can't pin ~1 MB just by asking). Out-of-range / non-decimal returns `400 invalid_max_queued` BEFORE opening the SSE stream — clean 4xx beats half-opening a stream + emitting a `stream_error` (which EventSource would auto-reconnect on). - **`slow_client_warning` capability tag** — single source of truth for the warning frame + `?maxQueued` query param + ring-size knob. Old daemons silently lack all of these; pre-flight via `caps.features`. - **SDK extensions** (`@qwen-code/sdk`): typed `DaemonSlowClientWarningEvent` (added to known event union and `DaemonStreamLifecycleEvent`); schema-validated by a new `isSlowClientWarningData` predicate; reducer (`reduceDaemonSessionEvent`) increments `slowClientWarningCount` + stores `lastSlowClientWarning`. Warning is **non-terminal** — `alive` stays true (only `client_evicted` / `stream_error` / `session_died` close the stream). Re-exported from the public SDK entry. - **Docs**: `qwen-serve-protocol.md` updates the features list (adds `slow_client_warning` and the previously-missing `client_identity` to match reality post-#4231), documents the `?maxQueued` query param, adds the warning frame to the event table, and notes the new default ring size. `qwen-serve.md` adds the `--event-ring-size` flag row. Tests: 19 eventBus (4 new: warning at 75%, once per episode, no `id` on the synthetic frame, hysteresis re-arm), 106 bridge (2 new: validate eventRingSize accept/reject), 111 server (4 new: ?maxQueued accept/absent/non-decimal/out-of-range + EXPECTED_STAGE1_FEATURES update), 14 SDK daemonEvents (2 new: schema validation + non-terminal reducer behavior). 321 focused tests total, all green. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- docs/developers/qwen-serve-protocol.md | 41 +++--- docs/users/qwen-serve.md | 19 +-- packages/cli/src/commands/serve.ts | 11 ++ packages/cli/src/serve/capabilities.ts | 5 + packages/cli/src/serve/eventBus.test.ts | 117 +++++++++++++++++- packages/cli/src/serve/eventBus.ts | 82 ++++++++++-- packages/cli/src/serve/httpAcpBridge.test.ts | 29 +++++ packages/cli/src/serve/httpAcpBridge.ts | 34 ++++- packages/cli/src/serve/runQwenServe.ts | 3 + packages/cli/src/serve/server.test.ts | 85 +++++++++++++ packages/cli/src/serve/server.ts | 61 +++++++++ packages/cli/src/serve/types.ts | 10 ++ packages/sdk-typescript/src/daemon/events.ts | 55 ++++++++ packages/sdk-typescript/src/daemon/index.ts | 2 + packages/sdk-typescript/src/index.ts | 2 + .../test/unit/daemonEvents.test.ts | 66 ++++++++++ 16 files changed, 584 insertions(+), 38 deletions(-) diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index bfdb849e6dc..fb89d28d151 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -91,13 +91,16 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design ['health', 'capabilities', 'session_create', 'session_scope_override', 'session_load', 'unstable_session_resume', 'session_list', 'session_prompt', 'session_cancel', 'session_events', - 'session_set_model', 'permission_vote'] + 'slow_client_warning', + 'session_set_model', 'client_identity', 'permission_vote'] ``` `session_scope_override` is the negotiation handle for the per-request `sessionScope` field on `POST /session` (see below). Older daemons silently ignore the field, so SDK clients should pre-flight `caps.features` for this tag before sending it. `session_load` and `unstable_session_resume` advertise the explicit-restore routes (`POST /session/:id/load` and `POST /session/:id/resume`). Older daemons return `404` for these paths, so SDK clients should pre-flight `caps.features` before calling. The `unstable_` prefix on `unstable_session_resume` mirrors the underlying ACP method (`connection.unstable_resumeSession`) — the daemon's wire shape is committed for v1, but the ACP method name itself may change before ACP marks resume stable. +`slow_client_warning` covers two co-released SSE backpressure knobs introduced in #4175 Wave 2.5 PR 10: (a) the daemon emits a `slow_client_warning` synthetic event-stream frame when a subscriber's queue crosses 75% full, once per overflow episode (rearmed after the queue drains below 37.5%); (b) `GET /session/:id/events` accepts a `?maxQueued=N` query param (range `[16, 2048]`) to pre-size the per-subscriber backlog for cold reconnects against a large replay ring. The daemon-wide ring size is controlled by `--event-ring-size` (default **8000**, per #3803 §02). Old daemons silently lack both — pre-flight this tag before opting in. + ## Routes > **Stage 1 limitation — no `DELETE /session/:id`.** Sessions live until @@ -341,6 +344,12 @@ Accept: text/event-stream Last-Event-ID: 42 ← optional, replays from after id 42 ``` +Query params: + +| Param | Required | Notes | +| ----------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `maxQueued` | no | Per-subscriber live-backlog cap. Range `[16, 2048]`, default 256. Bump for cold reconnects with `Last-Event-ID: 0` against a large replay ring so the force-pushed replay frames don't immediately trip eviction. Out-of-range / non-decimal values return `400 invalid_max_queued` before the SSE handshake opens. Pre-flight `caps.features.slow_client_warning` — old daemons silently ignore the param. | + Frame format. The `data:` line is the **full event envelope**, JSON-stringified on a single line — `{id?, v, type, data, originatorClientId?}`. The ACP-specific payload (`sessionUpdate`, `requestPermission` arguments, etc.) sits under the envelope's `data` field; the envelope's own `type` matches the SSE `event:` line. ``` @@ -360,28 +369,30 @@ data: {"v":1,"type":"client_evicted","data":{"reason":"queue_overflow","droppedA The SSE-level `id:` / `event:` lines duplicate `envelope.id` / `envelope.type` for EventSource compatibility. Raw-`fetch` consumers (the SDK's `parseSseStream`) read everything off the JSON envelope and ignore the SSE preamble lines. -| Event type | Trigger | -| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `session_update` | Any ACP `sessionUpdate` notification (LLM chunks, tool calls, usage) | -| `permission_request` | Agent asked for tool approval | -| `permission_resolved` | Some client voted on a permission via `POST /permission/:requestId` | -| `model_switched` | `POST /session/:id/model` succeeded | -| `model_switch_failed` | `POST /session/:id/model` rejected | -| `session_died` | Agent child crashed unexpectedly. **Terminal: SSE stream closes after this frame; the session is gone from `byId`.** Subscribers should reconnect via `POST /session` to spawn a fresh one. | -| `client_evicted` | Subscriber-local: queue overflow. **Terminal: SSE stream closes after this frame** (no `id` — synthetic). Other subscribers on the same session continue. | -| `stream_error` | Daemon-side error during fan-out. **Terminal: SSE stream closes after this frame** (no `id` — synthetic). | +| Event type | Trigger | +| --------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `session_update` | Any ACP `sessionUpdate` notification (LLM chunks, tool calls, usage) | +| `permission_request` | Agent asked for tool approval | +| `permission_resolved` | Some client voted on a permission via `POST /permission/:requestId` | +| `model_switched` | `POST /session/:id/model` succeeded | +| `model_switch_failed` | `POST /session/:id/model` rejected | +| `session_died` | Agent child crashed unexpectedly. **Terminal: SSE stream closes after this frame; the session is gone from `byId`.** Subscribers should reconnect via `POST /session` to spawn a fresh one. | +| `slow_client_warning` | Subscriber-local: queue ≥ 75% full. **Non-terminal** — the stream continues; the warning is a heads-up before eviction. Carries `{queueSize, maxQueued, lastEventId}`. Fires ONCE per overflow episode; re-arms after the queue drains below 37.5%. No `id` (synthetic). Pre-flight `caps.features.slow_client_warning`. | +| `client_evicted` | Subscriber-local: queue overflow. **Terminal: SSE stream closes after this frame** (no `id` — synthetic). Other subscribers on the same session continue. | +| `stream_error` | Daemon-side error during fan-out. **Terminal: SSE stream closes after this frame** (no `id` — synthetic). | Reconnect semantics: -- Send `Last-Event-ID: ` to replay events with `id > n` from the per-session ring (default depth 4000) +- Send `Last-Event-ID: ` to replay events with `id > n` from the per-session ring (default depth **8000**, tunable via `qwen serve --event-ring-size `) - **Gap detection (client-side):** if `` predates the oldest event still in the ring (e.g. you reconnect with `Last-Event-ID: 50` but the ring now holds 200–1199), the daemon replays from the oldest available event without raising. Compare the first replayed event's `id` against `n + 1`; any difference is the size of the lost window. Stage 2 will inject an explicit `stream_gap` synthetic frame on the daemon side; in Stage 1 detection is the client's responsibility. - IDs are monotonic per session, starting at 1 -- Synthetic terminal frames (`client_evicted`, `stream_error`) intentionally omit `id` so they don't burn a sequence slot for other subscribers +- Synthetic frames (`client_evicted`, `slow_client_warning`, `stream_error`) intentionally omit `id` so they don't burn a sequence slot for other subscribers Backpressure: -- Per-subscriber queue defaults to `maxQueued: 256` live items (replay frames during reconnect bypass the cap) -- On overflow the bus emits the `client_evicted` terminal frame and closes the subscription +- Per-subscriber queue defaults to `maxQueued: 256` live items (replay frames during reconnect bypass the cap). Override via `?maxQueued=N` (range `[16, 2048]`) on the SSE request. +- When a subscriber's queue crosses 75% full the bus force-pushes a `slow_client_warning` synthetic frame to that subscriber (once per overflow episode; re-armed after drain below 37.5%). The stream stays open — the warning is a heads-up so the client can drain faster or detach + reconnect cleanly. +- If the queue actually overflows the warning, the bus emits the `client_evicted` terminal frame and closes the subscription. ### `POST /permission/:requestId` diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index 80655dc269a..3d4fca9cd54 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -107,15 +107,16 @@ The token comparison is constant-time (SHA-256 + `crypto.timingSafeEqual`); 401 ## CLI flags -| Flag | Default | Purpose | -| ----------------------- | --------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `--port ` | `4170` | TCP port. `0` = OS-assigned ephemeral port. | -| `--hostname ` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. | -| `--token ` | — | Bearer token. Falls back to `QWEN_SERVER_TOKEN` env var (with leading/trailing whitespace stripped — handy for `$(cat token.txt)`). | -| `--max-sessions ` | `20` | Cap on concurrent live sessions. New `POST /session` requests that would spawn a fresh child return `503` (with `Retry-After: 5`) when the cap is hit; attaches to existing sessions are NOT counted. Set to `0` to disable. Sized for single-user / small-team usage; raise it if your deployment has the RAM/FD headroom (~30–50 MB per session). | -| `--workspace ` | `process.cwd()` | Absolute workspace path this daemon binds to (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02 — 1 daemon = 1 workspace). `POST /session` requests with a mismatched `cwd` return `400 workspace_mismatch`. For multi-workspace deployments, run one `qwen serve` per workspace on separate ports. | -| `--max-connections ` | `256` | Listener-level TCP connection cap (`server.maxConnections`). Bounds raw socket count irrespective of session count — slow / phantom SSE clients get rejected at accept time once full. Raise alongside `--max-sessions` if your deployment expects many SSE subscribers per session. | -| `--http-bridge` | `true` | Stage 1 mode: one `qwen --acp` child per daemon (bound to one workspace at boot, per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02); N sessions multiplex onto that child via ACP `newSession()`. Stage 2 native in-process becomes available later. | +| Flag | Default | Purpose | +| ----------------------- | --------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `--port ` | `4170` | TCP port. `0` = OS-assigned ephemeral port. | +| `--hostname ` | `127.0.0.1` | Bind interface. Anything beyond loopback requires a token. | +| `--token ` | — | Bearer token. Falls back to `QWEN_SERVER_TOKEN` env var (with leading/trailing whitespace stripped — handy for `$(cat token.txt)`). | +| `--max-sessions ` | `20` | Cap on concurrent live sessions. New `POST /session` requests that would spawn a fresh child return `503` (with `Retry-After: 5`) when the cap is hit; attaches to existing sessions are NOT counted. Set to `0` to disable. Sized for single-user / small-team usage; raise it if your deployment has the RAM/FD headroom (~30–50 MB per session). | +| `--workspace ` | `process.cwd()` | Absolute workspace path this daemon binds to (per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02 — 1 daemon = 1 workspace). `POST /session` requests with a mismatched `cwd` return `400 workspace_mismatch`. For multi-workspace deployments, run one `qwen serve` per workspace on separate ports. | +| `--max-connections ` | `256` | Listener-level TCP connection cap (`server.maxConnections`). Bounds raw socket count irrespective of session count — slow / phantom SSE clients get rejected at accept time once full. Raise alongside `--max-sessions` if your deployment expects many SSE subscribers per session. | +| `--event-ring-size ` | `8000` | Per-session SSE replay ring depth (#3803 §02 target). Sets the backlog available to `GET /session/:id/events` with `Last-Event-ID: N`. Larger = more reconnect headroom at the cost of a few hundred KB extra RAM per session. SDK clients can additionally request a larger per-subscriber backlog cap on a specific subscription via `?maxQueued=N` (range `[16, 2048]`, default 256). Daemons also emit a non-terminal `slow_client_warning` SSE frame at 75% queue fill so clients can drain / reconnect before getting evicted. Pre-flight `caps.features.slow_client_warning`. | +| `--http-bridge` | `true` | Stage 1 mode: one `qwen --acp` child per daemon (bound to one workspace at boot, per [#3803](https://github.com/QwenLM/qwen-code/issues/3803) §02); N sessions multiplex onto that child via ACP `newSession()`. Stage 2 native in-process becomes available later. | > **Sizing the load knobs.** `--max-sessions` is the **new-child** cap. > Three other layers also limit load — when sizing for a high-concurrency diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 20262145cce..b0452d59171 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -30,6 +30,7 @@ interface ServeArgs { token?: string; 'max-sessions': number; 'max-connections': number; + 'event-ring-size': number; workspace?: string; // Read from the kebab-case key only — the camelCase mirror that yargs // synthesizes is convenient for handlers but type-confusing here. The @@ -84,6 +85,15 @@ export const serveCommand: CommandModule = { 'sockets — slow/phantom SSE clients get rejected at accept time once full. ' + 'Set to 0 to disable.', }) + .option('event-ring-size', { + type: 'number', + default: 8000, + description: + 'Per-session SSE replay ring depth (#3803 §02 target). Sets the ' + + 'replay backlog available to `GET /session/:id/events?Last-Event-ID=N` ' + + 'reconnects. Larger = more reconnect headroom at the cost of a few ' + + 'hundred KB extra RAM per session. Must be a positive finite integer.', + }) .option('http-bridge', { type: 'boolean', default: true, @@ -122,6 +132,7 @@ export const serveCommand: CommandModule = { mode: 'http-bridge', maxSessions: argv['max-sessions'], maxConnections: argv['max-connections'], + eventRingSize: argv['event-ring-size'], workspace: argv.workspace, }); } catch (err) { diff --git a/packages/cli/src/serve/capabilities.ts b/packages/cli/src/serve/capabilities.ts index e022e27b74f..3f96142ed58 100644 --- a/packages/cli/src/serve/capabilities.ts +++ b/packages/cli/src/serve/capabilities.ts @@ -36,6 +36,11 @@ export const SERVE_CAPABILITY_REGISTRY = { session_prompt: { since: 'v1' }, session_cancel: { since: 'v1' }, session_events: { since: 'v1' }, + // Daemon emits `slow_client_warning` synthetic frames at 75% queue + // fill and honors `?maxQueued=N` (range [16, 2048]) on + // `GET /session/:id/events`. Old daemons silently lack both — SDK + // clients pre-flight this tag before opting in. + slow_client_warning: { since: 'v1' }, session_set_model: { since: 'v1' }, client_identity: { since: 'v1' }, permission_vote: { since: 'v1' }, diff --git a/packages/cli/src/serve/eventBus.test.ts b/packages/cli/src/serve/eventBus.test.ts index 063ceed97f4..c3d02d3dd46 100644 --- a/packages/cli/src/serve/eventBus.test.ts +++ b/packages/cli/src/serve/eventBus.test.ts @@ -97,13 +97,15 @@ describe('EventBus', () => { aborts.forEach((c) => c.abort()); }); - it('evicts a slow subscriber when its queue overflows', async () => { + it('evicts a slow subscriber when its queue overflows (warning precedes eviction)', async () => { const bus = new EventBus(); const abort = new AbortController(); const iter = bus.subscribe({ maxQueued: 2, signal: abort.signal }); - // Publish 3 events without draining the iterator. Queue cap is 2; the - // 3rd should trip the eviction path and append a `client_evicted` + // Publish 3 events without draining the iterator. Queue cap is 2; + // event 2 fills the queue to 100% (above the 75% warn threshold), + // so the bus force-pushes a `slow_client_warning`; event 3 then + // trips the eviction path and appends a `client_evicted` // terminal frame. bus.publish({ type: 'foo', data: 1 }); bus.publish({ type: 'foo', data: 2 }); @@ -113,14 +115,119 @@ describe('EventBus', () => { for await (const e of iter) { collected.push(e); } - expect(collected).toHaveLength(3); + expect(collected).toHaveLength(4); expect(collected[0]?.data).toBe(1); expect(collected[1]?.data).toBe(2); - expect(collected[2]?.type).toBe('client_evicted'); + expect(collected[2]?.type).toBe('slow_client_warning'); + expect(collected[3]?.type).toBe('client_evicted'); expect(bus.subscriberCount).toBe(0); abort.abort(); }); + it('emits slow_client_warning exactly once per overflow episode', async () => { + // Queue size 8; warn threshold = 75% = 6. Push to 6 → warning + // fires; push to 7 → no additional warning (sub.warned latched). + const bus = new EventBus(); + const abort = new AbortController(); + const iter = bus.subscribe({ maxQueued: 8, signal: abort.signal }); + + for (let i = 1; i <= 7; i++) bus.publish({ type: 'foo', data: i }); + + const collected: BridgeEvent[] = []; + // Drain 8 items (7 publishes + 1 warning). + for (let i = 0; i < 8; i++) { + const { value, done } = await iter[Symbol.asyncIterator]().next(); + if (done) break; + collected.push(value); + } + const warnings = collected.filter((e) => e.type === 'slow_client_warning'); + expect(warnings).toHaveLength(1); + expect(warnings[0]?.data).toMatchObject({ maxQueued: 8 }); + abort.abort(); + }); + + it('slow_client_warning frame has no id (synthetic, no sequence slot)', async () => { + const bus = new EventBus(); + const abort = new AbortController(); + const iter = bus.subscribe({ maxQueued: 2, signal: abort.signal }); + bus.publish({ type: 'foo', data: 1 }); + bus.publish({ type: 'foo', data: 2 }); + bus.publish({ type: 'foo', data: 3 }); + + const collected: BridgeEvent[] = []; + for await (const e of iter) collected.push(e); + const warning = collected.find((e) => e.type === 'slow_client_warning'); + const evicted = collected.find((e) => e.type === 'client_evicted'); + expect(warning).toBeDefined(); + expect(warning!.id).toBeUndefined(); + expect(evicted!.id).toBeUndefined(); + // The two live events that DID make it through must carry + // contiguous ids — synthetic frames must not burn a slot. + const live = collected.filter((e) => e.type === 'foo'); + expect(live.map((e) => e.id)).toEqual([1, 2]); + abort.abort(); + }); + + it('rearms slow_client_warning after queue drains below the hysteresis threshold', async () => { + // Threshold 75%, reset 37.5%. maxQueued=8 → warn at 6, reset at 3. + const bus = new EventBus(); + const abort = new AbortController(); + const iter = bus.subscribe({ maxQueued: 8, signal: abort.signal }); + const it = iter[Symbol.asyncIterator](); + + // Fill to 6 → first warning fires (force-pushed AFTER the 6th + // event, so it sits at the back of the queue behind the 6 live + // events). + for (let i = 1; i <= 6; i++) bus.publish({ type: 'foo', data: i }); + // Drain all 7 items (events 1–6 + warning frame) — leaves the + // queue empty, well below the 3-item reset threshold. + const firstEpisode: BridgeEvent[] = []; + for (let i = 0; i < 7; i++) firstEpisode.push((await it.next()).value); + expect( + firstEpisode.filter((e) => e.type === 'slow_client_warning'), + ).toHaveLength(1); + + // Trigger another publish so the hysteresis check inside publish() + // observes the drained queue and re-arms sub.warned. After this + // publish, live size = 1, well below the 3-item reset threshold. + bus.publish({ type: 'foo', data: 7 }); + expect((await it.next()).value.data).toBe(7); + + // Re-fill back past the threshold — second overflow episode must + // produce a second warning because the flag was re-armed. + for (let i = 8; i <= 13; i++) bus.publish({ type: 'foo', data: i }); + const secondEpisode: BridgeEvent[] = []; + for (let i = 0; i < 7; i++) secondEpisode.push((await it.next()).value); + expect( + secondEpisode.filter((e) => e.type === 'slow_client_warning'), + ).toHaveLength(1); + abort.abort(); + }); + + it('default ring size is 8000 (#3803 §02 target)', () => { + const bus = new EventBus(); + for (let i = 1; i <= 8001; i++) bus.publish({ type: 'foo', data: i }); + // After publishing 8001 frames into the default ring, the resume + // backlog should hold the most recent 8000 (1 through 8001 with + // the oldest dropped). Subscribing with `lastEventId: 0` replays + // exactly 8000 frames from the ring. + const it = bus + .subscribe({ lastEventId: 0, maxQueued: 9000 }) + [Symbol.asyncIterator](); + let count = 0; + const drain = (async () => { + for (let i = 0; i < 8000; i++) { + const { value, done } = await it.next(); + if (done) break; + if (value.type !== 'slow_client_warning' && value.id !== undefined) + count++; + } + })(); + return drain.then(() => { + expect(count).toBe(8000); + }); + }); + it('eviction detaches the abort listener from a stalled consumer (BmJT1)', async () => { // Pre-fix the eviction path only did `this.subs.delete(sub)`, // leaving the AbortSignal abort-listener attached because the diff --git a/packages/cli/src/serve/eventBus.ts b/packages/cli/src/serve/eventBus.ts index f85dc35b3de..911fd012d06 100644 --- a/packages/cli/src/serve/eventBus.ts +++ b/packages/cli/src/serve/eventBus.ts @@ -68,10 +68,23 @@ const DEFAULT_MAX_QUEUED = 256; * turn, real workloads can be 10× that or more once tool-call / * thought streams pile up). 1000 was the original default and could * be exhausted by a moderate turn before the client reconnected; - * 4000 gives ~30× headroom over a typical-but-busy turn at the cost - * of a few hundred KB of RAM per session. + * 8000 matches the target set in #3803 §02 for chatty Stage 1 + * sessions, with ~30–60× headroom over a typical-but-busy turn at + * the cost of a few hundred KB of RAM per session. Operators can + * override per-daemon via `qwen serve --event-ring-size `. */ -const DEFAULT_RING_SIZE = 4000; +export const DEFAULT_RING_SIZE = 8000; +/** + * Fraction of `maxQueued` at which a `slow_client_warning` synthetic + * frame is force-pushed to the at-risk subscriber. The warning fires + * ONCE per overflow episode (tracked via `sub.warned`); the queue + * must drain below `WARN_RESET_RATIO * maxQueued` before another + * warning can fire — small hysteresis prevents flap-near-threshold + * spam when a subscriber oscillates around 75% full. + */ +const WARN_THRESHOLD_RATIO = 0.75; +/** See `WARN_THRESHOLD_RATIO` doc. */ +const WARN_RESET_RATIO = 0.375; /** * Per-bus subscriber cap. With per-subscriber `maxQueued` defaulting to * 256 frames, 64 concurrent subscribers caps the per-session subscriber @@ -86,6 +99,16 @@ const DEFAULT_MAX_SUBSCRIBERS = 64; interface InternalSub { queue: BoundedAsyncQueue; evicted: boolean; + /** Cap remembered per subscriber so the warning ratio + reset can be + * checked without rummaging through the queue's private state. */ + maxQueued: number; + /** + * True once `slow_client_warning` has been force-pushed to this + * subscriber in the current overflow episode. Cleared when the queue + * drains below `WARN_RESET_RATIO * maxQueued` (hysteresis), so a + * subscriber that recovers and then lags again gets a fresh warning. + */ + warned: boolean; /** * BmJT1: cleanup hook for the eviction path (overflow → close queue * → remove from `subs`). Without this, the abort listener registered @@ -219,6 +242,34 @@ export class EventBus { // Under attack (thousands of stalled SSE clients) this // amplified into significant heap retention. sub.dispose(); + continue; + } + // Backpressure warning: synthetic `slow_client_warning` frame to + // the at-risk subscriber when its live backlog crosses + // `WARN_THRESHOLD_RATIO`. Fires ONCE per overflow episode (the + // `warned` flag clears only after `WARN_RESET_RATIO` hysteresis + // drain). Like `client_evicted` the frame carries no `id` — it + // is private to this subscriber and must not burn a sequence + // slot the replay ring would otherwise be missing for other + // healthy subscribers. Force-push so the warning bypasses the + // exact backlog cap that triggered it. + const liveSize = sub.queue.size; + if (!sub.warned && liveSize >= WARN_THRESHOLD_RATIO * sub.maxQueued) { + sub.warned = true; + const warningFrame: BridgeEvent = { + v: EVENT_SCHEMA_VERSION, + type: 'slow_client_warning', + data: { + queueSize: liveSize, + maxQueued: sub.maxQueued, + lastEventId: event.id ?? this.lastEventId, + }, + }; + sub.queue.forcePush(warningFrame); + } else if (sub.warned && liveSize <= WARN_RESET_RATIO * sub.maxQueued) { + // Hysteresis: subscriber recovered well below the warn line, + // re-arm so a future lag spike produces a fresh warning. + sub.warned = false; } } return event; @@ -253,15 +304,20 @@ export class EventBus { if (this.subs.size >= this.maxSubscribers) { throw new SubscriberLimitExceededError(this.maxSubscribers); } - const queue = new BoundedAsyncQueue( - opts.maxQueued ?? DEFAULT_MAX_QUEUED, - ); + const maxQueued = opts.maxQueued ?? DEFAULT_MAX_QUEUED; + const queue = new BoundedAsyncQueue(maxQueued); // `dispose` is assigned below (mutable so the closure can reference // `sub.dispose`); placeholder no-op covers the brief window between // `subs.add(sub)` and the real assignment so an absurdly fast // `publish() → forcePush → close → dispose()` race can't crash. - const sub: InternalSub = { queue, evicted: false, dispose: () => {} }; + const sub: InternalSub = { + queue, + evicted: false, + maxQueued, + warned: false, + dispose: () => {}, + }; this.subs.add(sub); if (opts.lastEventId !== undefined) { @@ -393,6 +449,18 @@ class BoundedAsyncQueue { constructor(private readonly maxSize: number) {} + /** + * Number of LIVE (non-force-pushed) items currently waiting in the + * buffer. Mirrors the cap check in `push()`: replay/eviction frames + * inserted via `forcePush` don't count toward the backpressure + * threshold the bus uses to decide when to emit + * `slow_client_warning`. Returns 0 (not negative) if the buffer + * happens to be all-force-pushed. + */ + get size(): number { + return Math.max(0, this.buf.length - this.forcedInBuf); + } + /** Returns true if accepted, false if dropped due to overflow. */ push(value: T): boolean { if (this.closed) return false; diff --git a/packages/cli/src/serve/httpAcpBridge.test.ts b/packages/cli/src/serve/httpAcpBridge.test.ts index 4c3d140cba7..571480f49cd 100644 --- a/packages/cli/src/serve/httpAcpBridge.test.ts +++ b/packages/cli/src/serve/httpAcpBridge.test.ts @@ -268,6 +268,35 @@ function makeChannel(opts: FakeAgentOpts = {}): ChannelHandle { } describe('createHttpAcpBridge', () => { + it('accepts a valid BridgeOptions.eventRingSize at construction time', () => { + // Smoke: positive finite integers are accepted; the underlying + // EventBus ring-size threading is exercised end-to-end in + // `eventBus.test.ts` ("default ring size is 8000 (#3803 §02 + // target)"). The bridge layer only contributes validation + + // pass-through. + expect(() => makeBridge({ eventRingSize: 1 })).not.toThrow(); + expect(() => makeBridge({ eventRingSize: 8000 })).not.toThrow(); + expect(() => makeBridge({ eventRingSize: 100_000 })).not.toThrow(); + }); + + it('rejects an invalid eventRingSize at construction time', () => { + expect(() => makeBridge({ eventRingSize: 0 })).toThrow( + /Invalid eventRingSize/, + ); + expect(() => makeBridge({ eventRingSize: -1 })).toThrow( + /Invalid eventRingSize/, + ); + expect(() => makeBridge({ eventRingSize: 1.5 })).toThrow( + /Invalid eventRingSize/, + ); + expect(() => makeBridge({ eventRingSize: Number.NaN })).toThrow( + /Invalid eventRingSize/, + ); + expect(() => + makeBridge({ eventRingSize: Number.POSITIVE_INFINITY }), + ).toThrow(/Invalid eventRingSize/); + }); + it('spawns a session and returns the agent-assigned id', async () => { const handles: ChannelHandle[] = []; const factory: ChannelFactory = async () => { diff --git a/packages/cli/src/serve/httpAcpBridge.ts b/packages/cli/src/serve/httpAcpBridge.ts index 9670032a60e..b0e6034f7f2 100644 --- a/packages/cli/src/serve/httpAcpBridge.ts +++ b/packages/cli/src/serve/httpAcpBridge.ts @@ -17,6 +17,7 @@ import { import { writeStderrLine } from '../utils/stdioHelpers.js'; import { EventBus, + DEFAULT_RING_SIZE, type BridgeEvent, type SubscribeOptions, } from './eventBus.js'; @@ -523,6 +524,20 @@ export interface BridgeOptions { * `ServeOptions.maxSessions` for the rationale. */ maxSessions?: number; + /** + * Per-session SSE replay ring depth. Sets `ringSize` on every + * `new EventBus(...)` the bridge constructs (both fresh sessions + * and restored sessions). Defaults to `DEFAULT_RING_SIZE` (8000, + * #3803 §02 target). Must be a positive finite integer; `0` / + * `NaN` / negative throw at boot (fail-CLOSED — same posture as + * `maxSessions`, where silently disabling a backpressure knob on a + * config typo is worse than failing to start). + * + * Operators tune via `qwen serve --event-ring-size `. Cost is + * roughly `ringSize × ~500 B per session` of RAM held until the + * session ends. + */ + eventRingSize?: number; /** * Bd1yh: per-`requestPermission` wall clock. After this many ms with * no client vote, the agent's permission promise resolves as @@ -1198,6 +1213,21 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { `Expected 'single' or 'thread'.`, ); } + // `eventRingSize` follows the same fail-CLOSED posture as + // `maxSessions`: silently disabling SSE backpressure on a config + // typo is worse than failing to start. Unlike `maxSessions` there + // is NO unlimited sentinel — an unbounded ring would grow forever. + const eventRingSize = opts.eventRingSize ?? DEFAULT_RING_SIZE; + if ( + !Number.isFinite(eventRingSize) || + !Number.isInteger(eventRingSize) || + eventRingSize < 1 + ) { + throw new TypeError( + `Invalid eventRingSize: ${opts.eventRingSize}. ` + + `Must be a positive finite integer.`, + ); + } const channelFactory = opts.channelFactory ?? defaultSpawnChannelFactory; const initTimeoutMs = opts.initializeTimeoutMs ?? DEFAULT_INIT_TIMEOUT_MS; if (initTimeoutMs <= 0) { @@ -1918,7 +1948,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { ci: ChannelInfo, sessionId: string, workspaceCwd: string, - events = new EventBus(), + events = new EventBus(eventRingSize), ): SessionEntry => { const entry: SessionEntry = { sessionId, @@ -2055,7 +2085,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { throw new SessionLimitExceededError(maxSessions); } - const restoreEvents = new EventBus(); + const restoreEvents = new EventBus(eventRingSize); let registeredEntry: SessionEntry | undefined; let ci: ChannelInfo | undefined; // Live counter shared with coalesced waiters (see InFlightRestore diff --git a/packages/cli/src/serve/runQwenServe.ts b/packages/cli/src/serve/runQwenServe.ts index 80c2d2e6a23..f9d6c33deca 100644 --- a/packages/cli/src/serve/runQwenServe.ts +++ b/packages/cli/src/serve/runQwenServe.ts @@ -168,6 +168,9 @@ export async function runQwenServe( deps.bridge ?? createHttpAcpBridge({ maxSessions: opts.maxSessions, + ...(opts.eventRingSize !== undefined + ? { eventRingSize: opts.eventRingSize } + : {}), boundWorkspace, }); let actualPort = opts.port; diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 01fa01e94e9..bb05259bc61 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -72,6 +72,7 @@ const EXPECTED_STAGE1_FEATURES = [ 'session_prompt', 'session_cancel', 'session_events', + 'slow_client_warning', 'session_set_model', 'client_identity', 'permission_vote', @@ -2112,6 +2113,90 @@ describe('GET /session/:id/events (SSE)', () => { expect(frames[0]?.id).toBe('42'); }); + it('forwards ?maxQueued=N to the bridge when in [16, 2048]', async () => { + const seen: Array = []; + const bridge = fakeBridge({ + async *subscribeImpl(_sessionId, opts) { + seen.push(opts?.maxQueued); + yield { id: 1, v: 1, type: 'session_update', data: 'x' }; + 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?maxQueued=512`, + ); + await readSseFrames(res.body!, 1); + expect(seen).toEqual([512]); + }); + + it('omits maxQueued from the bridge call when the query param is absent', async () => { + const seen: Array = []; + const bridge = fakeBridge({ + async *subscribeImpl(_sessionId, opts) { + seen.push(opts?.maxQueued); + yield { id: 1, v: 1, type: 'session_update', data: 'x' }; + 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`); + await readSseFrames(res.body!, 1); + // Empty param ≡ missing — bridge sees `undefined` so the bus + // applies its default cap (256). + expect(seen).toEqual([undefined]); + }); + + it('400s a non-decimal ?maxQueued before opening the SSE stream', async () => { + const bridge = fakeBridge({ + subscribeImpl: () => { + throw new Error('bridge must not be touched'); + }, + }); + 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?maxQueued=abc`, + ); + expect(res.status).toBe(400); + expect(await res.json()).toMatchObject({ code: 'invalid_max_queued' }); + }); + + it('400s an out-of-range ?maxQueued before opening the SSE stream', async () => { + const bridge = fakeBridge({ + subscribeImpl: () => { + throw new Error('bridge must not be touched'); + }, + }); + handle = await runQwenServe( + { hostname: '127.0.0.1', port: 0, mode: 'http-bridge' }, + { bridge }, + ); + const port = (handle.server.address() as { port: number }).port; + + for (const bad of ['0', '15', '2049', '9999']) { + const res = await fetch( + `http://127.0.0.1:${port}/session/sess-A/events?maxQueued=${bad}`, + ); + expect(res.status).toBe(400); + expect(await res.json()).toMatchObject({ code: 'invalid_max_queued' }); + } + }); + it('returns 404 when the bridge reports unknown session', async () => { const bridge = fakeBridge({ subscribeImpl: (sessionId) => { diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 1e98e415dcc..cbd30f042af 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -665,6 +665,12 @@ export function createServeApp( app.get('/session/:id/events', (req, res) => { const sessionId = req.params['id']; const lastEventId = parseLastEventId(req.headers['last-event-id']); + const maxQueued = parseMaxQueuedQuery(req.query['maxQueued'], res); + // `parseMaxQueuedQuery` sends its own 400 + JSON body on rejection + // (returns `null`) so the SSE handshake doesn't get half-written. + // `undefined` means "client didn't ask for an override; use bus + // default 256" — proceed as before. + if (maxQueued === null) return; let iter: AsyncIterator | undefined; const abort = new AbortController(); @@ -672,6 +678,7 @@ export function createServeApp( const iterable = bridge.subscribeEvents(sessionId, { signal: abort.signal, lastEventId, + ...(maxQueued !== undefined ? { maxQueued } : {}), }); iter = iterable[Symbol.asyncIterator](); } catch (err) { @@ -1035,6 +1042,60 @@ function isValidOutcome( ); } +/** Range bounds for the `?maxQueued=N` query param on `/session/:id/events`. */ +const MIN_QUERY_MAX_QUEUED = 16; +const MAX_QUERY_MAX_QUEUED = 2048; + +/** + * Parse the optional `?maxQueued=N` query param on + * `GET /session/:id/events`. Returns: + * - `undefined` — param absent, EventBus uses its default cap (256). + * - a positive integer in `[16, 2048]` — caller wants a custom cap. + * - `null` — malformed value; the function ALREADY sent a 400 JSON + * response and the route must short-circuit. (Pre-handshake 400 + * is safer than half-opening an SSE stream and emitting a + * `stream_error` frame the client has to parse — `EventSource` + * auto-reconnects on the latter.) + * + * Cap range rationale: lower bound 16 (smaller is useless for any + * replay backlog); upper bound 2048 (so a single subscriber can't + * pin ~1 MB of queue memory just by asking). + */ +function parseMaxQueuedQuery( + raw: unknown, + res: import('express').Response, +): number | undefined | null { + if (raw === undefined || raw === '') return undefined; + if (typeof raw !== 'string' || !/^\d+$/.test(raw)) { + writeStderrLine( + `qwen serve: rejected ?maxQueued "${String(raw).slice(0, 80)}" ` + + `(not a decimal integer)`, + ); + res.status(400).json({ + error: '`maxQueued` must be a decimal integer', + code: 'invalid_max_queued', + }); + return null; + } + const n = Number.parseInt(raw, 10); + if ( + !Number.isFinite(n) || + n < MIN_QUERY_MAX_QUEUED || + n > MAX_QUERY_MAX_QUEUED + ) { + writeStderrLine( + `qwen serve: rejected ?maxQueued "${raw.slice(0, 80)}" ` + + `(outside [${MIN_QUERY_MAX_QUEUED}, ${MAX_QUERY_MAX_QUEUED}])`, + ); + res.status(400).json({ + error: `\`maxQueued\` must be in [${MIN_QUERY_MAX_QUEUED}, ${MAX_QUERY_MAX_QUEUED}]`, + code: 'invalid_max_queued', + }); + return null; + } + return n; +} + function parseLastEventId(raw: unknown): number | undefined { // Stricter than Number.parseInt: only accept pure decimal digits to avoid // values like "1abc" or "1.5e10z" silently parsing to 1. diff --git a/packages/cli/src/serve/types.ts b/packages/cli/src/serve/types.ts index 2fa4c72a927..7ebb041f88b 100644 --- a/packages/cli/src/serve/types.ts +++ b/packages/cli/src/serve/types.ts @@ -63,6 +63,16 @@ export interface ServeOptions { * (default cap 64) plus short-lived REST calls. */ maxConnections?: number; + /** + * Per-session SSE replay ring depth. Threaded into the bridge as + * `BridgeOptions.eventRingSize` and used at every `new EventBus(...)` + * construction site. Defaults to 8000 (the target named in + * #3803 §02 for chatty Stage 1 sessions). Must be a positive + * finite integer — `0` / `NaN` / negative fail at boot. Larger + * rings let clients with longer reconnect gaps replay more history + * at the cost of a few hundred KB extra RAM per session. + */ + eventRingSize?: number; /** * Absolute workspace path this daemon binds to. Per #3803 §02 the * daemon is **1 daemon = 1 workspace × N sessions**: one bound diff --git a/packages/sdk-typescript/src/daemon/events.ts b/packages/sdk-typescript/src/daemon/events.ts index 13e555f6a25..1612fdd2aeb 100644 --- a/packages/sdk-typescript/src/daemon/events.ts +++ b/packages/sdk-typescript/src/daemon/events.ts @@ -14,6 +14,7 @@ const DAEMON_KNOWN_EVENT_TYPE_VALUES = [ 'model_switch_failed', 'session_died', 'client_evicted', + 'slow_client_warning', 'stream_error', ] as const; @@ -80,6 +81,20 @@ export interface DaemonClientEvictedData { [key: string]: unknown; } +export interface DaemonSlowClientWarningData { + /** Live (non-replay) items currently queued for this subscriber. */ + queueSize: number; + /** Per-subscriber backlog cap that triggered the warning. */ + maxQueued: number; + /** + * Most recent monotonic event id observed by the bus at warning + * time. Lets the client decide whether to reconnect with a + * `Last-Event-ID` or detach + drain. + */ + lastEventId: number; + [key: string]: unknown; +} + export interface DaemonStreamErrorData { error: string; [key: string]: unknown; @@ -113,6 +128,10 @@ export type DaemonClientEvictedEvent = DaemonEventEnvelope< 'client_evicted', DaemonClientEvictedData >; +export type DaemonSlowClientWarningEvent = DaemonEventEnvelope< + 'slow_client_warning', + DaemonSlowClientWarningData +>; export type DaemonStreamErrorEvent = DaemonEventEnvelope< 'stream_error', DaemonStreamErrorData @@ -130,6 +149,7 @@ export type DaemonControlEvent = export type DaemonStreamLifecycleEvent = | DaemonClientEvictedEvent + | DaemonSlowClientWarningEvent | DaemonStreamErrorEvent; export type KnownDaemonEvent = @@ -161,6 +181,14 @@ export interface DaemonSessionViewState { lastDroppedPermissionRequestId?: string; unmatchedPermissionResolutionCount: number; lastUnmatchedPermissionResolutionId?: string; + /** + * Count of `slow_client_warning` frames this stream has observed. + * Non-terminal — warnings precede eviction but don't themselves + * close the stream. Adapters tap this counter to surface "your + * stream is lagging" UI before `client_evicted` arrives. + */ + slowClientWarningCount: number; + lastSlowClientWarning?: DaemonSlowClientWarningData; } export function createDaemonSessionViewState( @@ -184,6 +212,8 @@ export function createDaemonSessionViewState( seed.unmatchedPermissionResolutionCount ?? 0, lastUnmatchedPermissionResolutionId: seed.lastUnmatchedPermissionResolutionId, + slowClientWarningCount: seed.slowClientWarningCount ?? 0, + lastSlowClientWarning: seed.lastSlowClientWarning, }; } @@ -233,6 +263,10 @@ export function asKnownDaemonEvent( return isClientEvictedData(event.data) ? (event as DaemonClientEvictedEvent) : undefined; + case 'slow_client_warning': + return isSlowClientWarningData(event.data) + ? (event as DaemonSlowClientWarningEvent) + : undefined; case 'stream_error': return isStreamErrorData(event.data) ? (event as DaemonStreamErrorEvent) @@ -328,6 +362,16 @@ export function reduceDaemonSessionEvent( terminalEvent: chooseTerminalEvent(base.terminalEvent, event), pendingPermissions: {}, }; + case 'slow_client_warning': + // Non-terminal: warning precedes eviction but doesn't close + // the stream on its own. Count + capture the latest snapshot + // so adapters can render lag UI (or pre-emptively detach). + // `alive` and `pendingPermissions` are unchanged. + return { + ...base, + slowClientWarningCount: base.slowClientWarningCount + 1, + lastSlowClientWarning: event.data, + }; case 'stream_error': return { ...base, @@ -438,6 +482,17 @@ function isClientEvictedData(value: unknown): value is DaemonClientEvictedData { ); } +function isSlowClientWarningData( + value: unknown, +): value is DaemonSlowClientWarningData { + return ( + isRecord(value) && + typeof value['queueSize'] === 'number' && + typeof value['maxQueued'] === 'number' && + typeof value['lastEventId'] === 'number' + ); +} + function isStreamErrorData(value: unknown): value is DaemonStreamErrorData { return isRecord(value) && isNonEmptyString(value['error']); } diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index d6c1c12f2a4..547fc89dc78 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -49,6 +49,8 @@ export type { DaemonSessionUpdateData, DaemonSessionUpdateEvent, DaemonSessionViewState, + DaemonSlowClientWarningData, + DaemonSlowClientWarningEvent, DaemonStreamErrorData, DaemonStreamErrorEvent, DaemonStreamLifecycleEvent, diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index cd15e0b010e..cd4315ef3c0 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -50,6 +50,8 @@ export { type DaemonSessionUpdateData, type DaemonSessionUpdateEvent, type DaemonSessionViewState, + type DaemonSlowClientWarningData, + type DaemonSlowClientWarningEvent, type DaemonStreamErrorData, type DaemonStreamErrorEvent, type DaemonStreamLifecycleEvent, diff --git a/packages/sdk-typescript/test/unit/daemonEvents.test.ts b/packages/sdk-typescript/test/unit/daemonEvents.test.ts index 6a65e4f2996..5430fc4ad87 100644 --- a/packages/sdk-typescript/test/unit/daemonEvents.test.ts +++ b/packages/sdk-typescript/test/unit/daemonEvents.test.ts @@ -455,4 +455,70 @@ describe('daemon event schema', () => { expect(upgradedToDeath.terminalEvent?.type).toBe('session_died'); expect(upgradedToDeath.lastEventId).toBe(3); }); + + it('recognizes slow_client_warning frames as known events', () => { + const warning = { + // No `id` on synthetic frames (matches the daemon's emit shape). + v: 1, + type: 'slow_client_warning', + data: { queueSize: 192, maxQueued: 256, lastEventId: 42 }, + }; + const known = asKnownDaemonEvent(warning); + expect(known?.type).toBe('slow_client_warning'); + + // Schema validation: required numeric fields. Missing or wrongly + // typed payloads must NOT be recognized as known events. + expect( + asKnownDaemonEvent({ + v: 1, + type: 'slow_client_warning', + data: { queueSize: 'lots', maxQueued: 256, lastEventId: 42 }, + }), + ).toBeUndefined(); + expect( + asKnownDaemonEvent({ + v: 1, + type: 'slow_client_warning', + data: { queueSize: 192, lastEventId: 42 }, + }), + ).toBeUndefined(); + }); + + it('reduces slow_client_warning into the view state without ending the stream', () => { + const state = reduceDaemonSessionEvents([ + { + id: 1, + v: 1, + type: 'session_update', + data: { sessionId: 's-1', phase: 'prompting' }, + }, + // Warning #1. + { + v: 1, + type: 'slow_client_warning', + data: { queueSize: 200, maxQueued: 256, lastEventId: 1 }, + }, + // Warning #2 (e.g. after a drain + refill on the daemon side). + { + v: 1, + type: 'slow_client_warning', + data: { queueSize: 220, maxQueued: 256, lastEventId: 5 }, + }, + ]); + + // Counter increments + most recent snapshot wins. + expect(state.slowClientWarningCount).toBe(2); + expect(state.lastSlowClientWarning).toEqual({ + queueSize: 220, + maxQueued: 256, + lastEventId: 5, + }); + // Warning is non-terminal — stream is still alive, no + // terminalEvent recorded. + expect(state.alive).toBe(true); + expect(state.terminalEvent).toBeUndefined(); + // Warnings carry no `id`, so `lastEventId` stays at the highest + // id observed (the original session_update at id=1). + expect(state.lastEventId).toBe(1); + }); }); From a6ccde5b1883cbb7952f1b81ce12b2bbe7327a3e Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Sun, 17 May 2026 17:20:28 +0800 Subject: [PATCH 2/5] refactor(serve): adopt PR #4237 review feedback (eventBus polish) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address the actionable items from the Qwen Code review bot's pass on PR #4237: - Pre-compute `warnThreshold` / `warnResetThreshold` per `InternalSub` at `subscribe()` time so `publish()`'s per-event hot path is one integer compare per subscriber instead of a multiply + compare. The `!warned` short-circuit still collapses the steady state to a single boolean read; this just shaves a multiply when the threshold check actually fires. - Document the back-of-queue ordering choice for the synthetic `slow_client_warning` frame in `EventBus.publish()`: front-push was considered but mid-stream front-insertion would mis-count `forcedInBuf` in `BoundedAsyncQueue.next()`, and `forcePush` already short-circuits via `resolvers.shift()` for the active-consumer case — the back-of-queue path only matters for stalled consumers, who can't drain regardless of warning position. - Reuse the existing `collect()` helper in the "default ring size 8000" test for consistency with the rest of the file; the new test also tightens the assertion by checking that the first retained event id is 2 (id=1 dropped by the ring) and the last is 8001. - Soften the "~500 B per session" magic number in `BridgeOptions.eventRingSize`'s JSDoc to a qualitative description (each retained `BridgeEvent` is a reference plus its serialized payload; ceiling scales as `ringSize × average-event-size`). Rejected: - Bot's claim that the error JSON contains `\`...\`` escape sequences — bot misread the JS template-literal source as the wire output; `JSON.stringify` does not escape backticks, and the existing `cwd` error messages use the same style. - Bot's "use `Record` instead of `[key: string]: unknown`" suggestion on `DaemonSlowClientWarningData` — every other event-data type in `sdk-typescript/src/daemon/events.ts` carries the same index signature for additive-field compatibility. - Bot's "features list breaks alphabetical order" — the capability list is grouped by protocol lifecycle (health → capabilities → session lifecycle → events → permissions), not alphabetical. Tests: 139 focused tests across eventBus + httpAcpBridge + SDK daemon events — all passing. Behavior unchanged; this is hot-path micro-opt + comment polish only. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/cli/src/serve/eventBus.test.ts | 38 ++++++++++++------------ packages/cli/src/serve/eventBus.ts | 39 ++++++++++++++++++++++--- packages/cli/src/serve/httpAcpBridge.ts | 8 +++-- 3 files changed, 59 insertions(+), 26 deletions(-) diff --git a/packages/cli/src/serve/eventBus.test.ts b/packages/cli/src/serve/eventBus.test.ts index c3d02d3dd46..601b007afd3 100644 --- a/packages/cli/src/serve/eventBus.test.ts +++ b/packages/cli/src/serve/eventBus.test.ts @@ -204,28 +204,28 @@ describe('EventBus', () => { abort.abort(); }); - it('default ring size is 8000 (#3803 §02 target)', () => { + it('default ring size is 8000 (#3803 §02 target)', async () => { const bus = new EventBus(); for (let i = 1; i <= 8001; i++) bus.publish({ type: 'foo', data: i }); - // After publishing 8001 frames into the default ring, the resume - // backlog should hold the most recent 8000 (1 through 8001 with - // the oldest dropped). Subscribing with `lastEventId: 0` replays - // exactly 8000 frames from the ring. - const it = bus - .subscribe({ lastEventId: 0, maxQueued: 9000 }) - [Symbol.asyncIterator](); - let count = 0; - const drain = (async () => { - for (let i = 0; i < 8000; i++) { - const { value, done } = await it.next(); - if (done) break; - if (value.type !== 'slow_client_warning' && value.id !== undefined) - count++; - } - })(); - return drain.then(() => { - expect(count).toBe(8000); + // After publishing 8001 frames into the default ring, the replay + // backlog should hold the most recent 8000 (oldest dropped). + // A `lastEventId: 0` resume with a queue cap larger than the ring + // collects exactly 8000 live frames; ids start at 2 because id=1 + // was the one shifted out of the ring. + const abort = new AbortController(); + const iter = bus.subscribe({ + lastEventId: 0, + maxQueued: 9000, + signal: abort.signal, }); + const events = await collect(iter, 8000); + abort.abort(); + const liveIds = events + .filter((e) => e.id !== undefined) + .map((e) => e.id as number); + expect(liveIds).toHaveLength(8000); + expect(liveIds[0]).toBe(2); + expect(liveIds[liveIds.length - 1]).toBe(8001); }); it('eviction detaches the abort listener from a stalled consumer (BmJT1)', async () => { diff --git a/packages/cli/src/serve/eventBus.ts b/packages/cli/src/serve/eventBus.ts index 911fd012d06..6f094332639 100644 --- a/packages/cli/src/serve/eventBus.ts +++ b/packages/cli/src/serve/eventBus.ts @@ -102,11 +102,21 @@ interface InternalSub { /** Cap remembered per subscriber so the warning ratio + reset can be * checked without rummaging through the queue's private state. */ maxQueued: number; + /** + * Pre-computed `WARN_THRESHOLD_RATIO * maxQueued` so `publish()` + * does one integer compare per subscriber instead of a multiply + + * compare. `publish()` is on the per-event hot path; per-sub + * caching here collapses to a single field read in the steady + * state (after the `!warned` short-circuit). + */ + warnThreshold: number; + /** Pre-computed `WARN_RESET_RATIO * maxQueued` — see `warnThreshold`. */ + warnResetThreshold: number; /** * True once `slow_client_warning` has been force-pushed to this * subscriber in the current overflow episode. Cleared when the queue - * drains below `WARN_RESET_RATIO * maxQueued` (hysteresis), so a - * subscriber that recovers and then lags again gets a fresh warning. + * drains below `warnResetThreshold` (hysteresis), so a subscriber + * that recovers and then lags again gets a fresh warning. */ warned: boolean; /** @@ -253,8 +263,27 @@ export class EventBus { // slot the replay ring would otherwise be missing for other // healthy subscribers. Force-push so the warning bypasses the // exact backlog cap that triggered it. + // + // Ordering: `forcePush` appends to the queue's back. Pushing to + // the FRONT was considered to maximize lead-time, but (a) the + // forward-position invariant in `BoundedAsyncQueue.next()`'s + // `forcedInBuf` accounting is sized for "replay at front, live + // at back" — mid-stream front-insertion would mis-count the + // live backlog cap; and (b) when a consumer is actively + // `await`ing `next()`, `forcePush`'s `resolvers.shift()` + // shortcut delivers the warning immediately without ever + // touching `buf`. The back-of-queue case only matters for + // stalled consumers — and a stalled consumer can't drain + // regardless of warning position, so the ordering is + // informational by the time they finally pull it. + // + // The `warnThreshold` / `warnResetThreshold` are pre-computed + // at `subscribe()` time so the per-publish hot path is one + // integer compare per subscriber (after the `!warned` + // short-circuit collapses warm-state checks to a single + // boolean read). const liveSize = sub.queue.size; - if (!sub.warned && liveSize >= WARN_THRESHOLD_RATIO * sub.maxQueued) { + if (!sub.warned && liveSize >= sub.warnThreshold) { sub.warned = true; const warningFrame: BridgeEvent = { v: EVENT_SCHEMA_VERSION, @@ -266,7 +295,7 @@ export class EventBus { }, }; sub.queue.forcePush(warningFrame); - } else if (sub.warned && liveSize <= WARN_RESET_RATIO * sub.maxQueued) { + } else if (sub.warned && liveSize <= sub.warnResetThreshold) { // Hysteresis: subscriber recovered well below the warn line, // re-arm so a future lag spike produces a fresh warning. sub.warned = false; @@ -315,6 +344,8 @@ export class EventBus { queue, evicted: false, maxQueued, + warnThreshold: WARN_THRESHOLD_RATIO * maxQueued, + warnResetThreshold: WARN_RESET_RATIO * maxQueued, warned: false, dispose: () => {}, }; diff --git a/packages/cli/src/serve/httpAcpBridge.ts b/packages/cli/src/serve/httpAcpBridge.ts index b0e6034f7f2..159fc69806f 100644 --- a/packages/cli/src/serve/httpAcpBridge.ts +++ b/packages/cli/src/serve/httpAcpBridge.ts @@ -533,9 +533,11 @@ export interface BridgeOptions { * `maxSessions`, where silently disabling a backpressure knob on a * config typo is worse than failing to start). * - * Operators tune via `qwen serve --event-ring-size `. Cost is - * roughly `ringSize × ~500 B per session` of RAM held until the - * session ends. + * Operators tune via `qwen serve --event-ring-size `. Cost + * scales linearly with `ringSize`; each retained `BridgeEvent` is + * an object reference plus its serialized payload (text chunks / + * tool-call args / etc.), so the per-session memory ceiling is + * `ringSize × average-event-size` held until the session ends. */ eventRingSize?: number; /** From bae42c88bc8bdd542bb53928fe54ef16b7f7701d Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Sun, 17 May 2026 17:45:13 +0800 Subject: [PATCH 3/5] fix(serve): correct queue tagging + plumb maxQueued through SDK MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address both P2 findings from the Codex review pass on PR #4237. **Bug 1: `BoundedAsyncQueue.forcedInBuf` position-invariant break** The previous `forcedInBuf` counter only tracked LIVE-vs-FORCED correctly when all forced entries lived at the FRONT of the buffer (subscribe-time `Last-Event-ID` replay). The new mid-stream `slow_client_warning` path force-pushes to the BACK of the queue while the queue is still open, which the existing accounting was not designed for: - publish 6 events at maxQueued=8 → 75% threshold trips → force-push warning at the back → buf=[1..6, warning], forcedInBuf=1. - consumer shifts `1` → forcedInBuf decremented to 0 (incorrect: `1` was a live frame, not the forced one). - consumer drains 2..6 + warning → buf=[], forcedInBuf=0, true live count = 0, but `size` getter and `push()` cap check then use `buf.length - forcedInBuf` which drifts over subsequent refills, causing premature warn / eviction before the cap is actually reached. Replace the position-dependent counter with a per-entry `{value, forced}` tag. `liveCount` is incremented in `push()` / decremented in `next()` only when the shifted entry was non-forced — position becomes irrelevant. `size` getter returns `liveCount` directly. The class doc comment is rewritten to call out that the new tag is the position-independent replacement for the old "forced frames must stay at the front" invariant. Regression test in `eventBus.test.ts` reproduces the codex trace (warn at 75%, drain past warning, refill to cap) and asserts no premature eviction. **Bug 2: SDK does not expose `?maxQueued`** `docs/users/qwen-serve.md` and `docs/developers/qwen-serve-protocol.md` both document `?maxQueued=N` as something SDK clients can request, but `SubscribeOptions` on `DaemonClient` only declared `lastEventId` + `signal`, and `subscribeEvents()` always fetched `/events` without a query string. Typed-SDK consumers had no way to opt in without hand-crafting URLs. - Add `SubscribeOptions.maxQueued?: number` with JSDoc noting the daemon range `[16, 2048]` and the pre-flight requirement on `caps.features.slow_client_warning`. - `DaemonClient.subscribeEvents` builds the URL with an optional `?maxQueued=` segment. No client-side range validation — the daemon's `parseMaxQueuedQuery` is the source of truth and returns structured `400 invalid_max_queued`; duplicating the bounds in two layers would diverge on the next tweak. - `DaemonSessionSubscribeOptions extends SubscribeOptions` so the new field flows through `DaemonSessionClient` automatically. Three new SDK tests: - subscribeEvents appends `?maxQueued=N` when set - omits the query string when absent (existing behavior preserved) - propagates a `400 invalid_max_queued` unchanged Tests: 214 focused tests across eventBus / bridge / SDK DaemonClient / DaemonSessionClient / daemonEvents, plus 111 in the server suite. All green; the new eventBus regression case proves the position-invariant fix. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/cli/src/serve/eventBus.test.ts | 62 +++++++++++++++ packages/cli/src/serve/eventBus.ts | 79 ++++++++++--------- .../sdk-typescript/src/daemon/DaemonClient.ts | 27 ++++++- .../test/unit/DaemonClient.test.ts | 41 ++++++++++ 4 files changed, 166 insertions(+), 43 deletions(-) diff --git a/packages/cli/src/serve/eventBus.test.ts b/packages/cli/src/serve/eventBus.test.ts index 601b007afd3..029526b8f82 100644 --- a/packages/cli/src/serve/eventBus.test.ts +++ b/packages/cli/src/serve/eventBus.test.ts @@ -204,6 +204,68 @@ describe('EventBus', () => { abort.abort(); }); + it('warn-at-back forced frame does NOT skew the live cap for subsequent publishes (codex P2)', async () => { + // Regression for the `forcedInBuf` position-invariant bug Codex + // flagged: a mid-stream slow_client_warning force-pushed to the + // BACK of the queue, then drained past, would previously cause + // `next()` to decrement the forced counter on a LIVE shift, + // making subsequent `push()` cap checks under-count live items + // and warn/evict the client before they actually had `maxQueued` + // live items in queue. + const bus = new EventBus(); + const abort = new AbortController(); + const iter = bus.subscribe({ maxQueued: 8, signal: abort.signal }); + const it = iter[Symbol.asyncIterator](); + + // Episode 1: fill to 6 → warn at 75%. buf = [1..6, warning]. + for (let i = 1; i <= 6; i++) bus.publish({ type: 'foo', data: i }); + + // Drain ALL 7 items (events 1..6 + warning frame). Live cap should + // now be 0 — the warning was a forced frame and must NOT have + // counted as a live drain. + const drained: BridgeEvent[] = []; + for (let i = 0; i < 7; i++) drained.push((await it.next()).value); + expect( + drained.filter((e) => e.type === 'slow_client_warning'), + ).toHaveLength(1); + + // Refill to EXACTLY maxQueued (8). Pre-fix: the post-drain live + // count was wrong, so somewhere between pushes 5 and 7 the 75% + // threshold (live=6) fired a second warning prematurely or the + // push at 7 was even rejected. Post-fix: live count is the truth, + // and the second warning fires exactly at push 8 (live=8, queue + // full → push 8 fills the cap and either succeeds at the cap line + // or trips the warn check first). + let rejected = 0; + for (let i = 7; i <= 14; i++) { + // Stop publishing once the queue refuses — the 8th live publish + // is the maxQueued ceiling. + const ok = bus.publish({ type: 'foo', data: i }) !== undefined; + if (!ok) rejected++; + } + void rejected; // EventBus.publish never returns false; rejection + // happens inside the bus when subscriber queues fill. + + // Drain everything that's still alive in the iter. The exact frame + // shape varies (depending on whether the bus also force-pushed a + // second warning + evicted), but the ASSERTION we need is: the + // sub didn't get evicted on a phantom premature overflow — i.e. + // we received MORE THAN 1 live frame in this episode (pre-fix, + // the live count drift evicted after 0-1 frames). + const episode2: BridgeEvent[] = []; + for (let i = 0; i < 9; i++) { + const { value, done } = await it.next(); + if (done) break; + episode2.push(value); + } + const live2 = episode2.filter((e) => e.id !== undefined && e.id >= 7); + // Pre-fix: live2 would be <8 because the queue evicted prematurely + // after the buggy live count drift. Post-fix: all 8 live frames + // (ids 7..14) get through cleanly. + expect(live2.length).toBeGreaterThanOrEqual(8); + abort.abort(); + }); + it('default ring size is 8000 (#3803 §02 target)', async () => { const bus = new EventBus(); for (let i = 1; i <= 8001; i++) bus.publish({ type: 'foo', data: i }); diff --git a/packages/cli/src/serve/eventBus.ts b/packages/cli/src/serve/eventBus.ts index 6f094332639..5e21b710301 100644 --- a/packages/cli/src/serve/eventBus.ts +++ b/packages/cli/src/serve/eventBus.ts @@ -446,50 +446,53 @@ function emptyAsyncIterable(): AsyncIterable { * that signal to evict slow subscribers. * * The cap (`maxSize`) applies only to LIVE items pushed via `push()`. Items - * inserted via `forcePush()` (the `Last-Event-ID` replay path on subscribe - * and the terminal `client_evicted` frame) are tracked separately and don't + * inserted via `forcePush()` (the `Last-Event-ID` replay path on subscribe, + * the terminal `client_evicted` frame, and the mid-stream + * `slow_client_warning` frame) carry a `forced` tag per entry and never * count toward the cap. Without this split, a reconnect with a large * backlog would force-push ~ringSize entries into `buf`, push `buf.length` * past `maxSize`, and the very next live publish would evict the * just-resumed subscriber — defeating the resume contract. + * + * Previously this class tracked `forcedInBuf` as a count, which was + * correct only when forced frames stayed contiguous at the FRONT of the + * buffer (subscribe-time replay). The `slow_client_warning` path + * force-pushes mid-stream to the BACK of the queue, so the count-based + * approach drifted: a live shift would decrement `forcedInBuf`, then a + * later cap check on a live push would under-count the live backlog and + * warn/evict the client before there were actually `maxSize` live + * items. The per-entry `forced` tag below is the position-independent + * fix. */ +interface BoundedQueueEntry { + value: T; + /** True for replay / eviction / slow_client_warning frames (don't count toward cap). */ + forced: boolean; +} + class BoundedAsyncQueue { - private readonly buf: T[] = []; + private readonly buf: Array> = []; private readonly resolvers: Array<(v: IteratorResult) => void> = []; private closed = false; /** - * Number of force-pushed items still in `buf`. The cap check in - * `push()` only applies to LIVE items; this counter tells us how - * many slots in `buf` are replay-injected and shouldn't count. - * - * Position invariant: under the bus's two callers, - * 1. subscribe-time replay (`Last-Event-ID` resume) — forcePush - * fires BEFORE any live `push()`, so replay items are at the - * front of `buf`; - * 2. eviction terminal frame — forcePush fires AFTER `push()` - * rejection, then `close()` is called immediately, so the - * eviction frame is at the BACK of `buf`. - * - * `next()` decrements `forcedInBuf` whenever the counter is > 0 on - * shift, which is correct for case (1). For case (2) it slightly - * misaccounts (decrements on the first live shift), but that's - * harmless: the queue is closed so no `push()` runs the cap check - * again. The counter only matters for live cap enforcement. + * O(1) snapshot of how many LIVE (non-forced) entries are in `buf`. + * Maintained directly by `push()`/`next()`: any time a forced entry + * is added or removed `liveCount` is untouched; any time a live entry + * is added or removed `liveCount` moves with it. Replaces the + * position-dependent `forcedInBuf` heuristic — `liveCount` is correct + * no matter where in the queue the forced entries are. */ - private forcedInBuf = 0; + private liveCount = 0; constructor(private readonly maxSize: number) {} /** * Number of LIVE (non-force-pushed) items currently waiting in the - * buffer. Mirrors the cap check in `push()`: replay/eviction frames - * inserted via `forcePush` don't count toward the backpressure - * threshold the bus uses to decide when to emit - * `slow_client_warning`. Returns 0 (not negative) if the buffer - * happens to be all-force-pushed. + * buffer. Backpressure decisions in `EventBus.publish()` (the + * `slow_client_warning` threshold) read this value. */ get size(): number { - return Math.max(0, this.buf.length - this.forcedInBuf); + return this.liveCount; } /** Returns true if accepted, false if dropped due to overflow. */ @@ -501,12 +504,14 @@ class BoundedAsyncQueue { return true; } // Cap is on the LIVE backlog only. - if (this.buf.length - this.forcedInBuf >= this.maxSize) return false; - this.buf.push(value); + if (this.liveCount >= this.maxSize) return false; + this.buf.push({ value, forced: false }); + this.liveCount += 1; return true; } - /** Bypasses the size cap. Used for replay frames and terminal eviction. */ + /** Bypasses the size cap. Used for replay frames, eviction terminal, + * and slow-client warnings. */ forcePush(value: T): void { if (this.closed) return; const r = this.resolvers.shift(); @@ -514,8 +519,7 @@ class BoundedAsyncQueue { r({ value, done: false }); return; } - this.buf.push(value); - this.forcedInBuf += 1; + this.buf.push({ value, forced: true }); } /** @@ -539,7 +543,7 @@ class BoundedAsyncQueue { // Truncate the buffer so subsequent `next()` calls see the // closed sentinel immediately. this.buf.length = 0; - this.forcedInBuf = 0; + this.liveCount = 0; } while (this.resolvers.length > 0) { this.resolvers.shift()!({ @@ -554,12 +558,9 @@ class BoundedAsyncQueue { // queue whose element type legitimately includes `undefined`. The bus // never pushes undefined today, but the queue is generic. if (this.buf.length > 0) { - const value = this.buf.shift() as T; - // Force-pushed entries are FIFO at the front of `buf` (forcePush - // only happens at subscribe time, before any live push). So as long - // as `forcedInBuf > 0` the shifted item is a replay frame. - if (this.forcedInBuf > 0) this.forcedInBuf -= 1; - return Promise.resolve({ value, done: false }); + const entry = this.buf.shift() as BoundedQueueEntry; + if (!entry.forced) this.liveCount -= 1; + return Promise.resolve({ value: entry.value, done: false }); } if (this.closed) { return Promise.resolve({ diff --git a/packages/sdk-typescript/src/daemon/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 1ee89b99757..246b248507c 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -139,6 +139,18 @@ export interface SubscribeOptions { lastEventId?: number; /** Aborts the subscription cleanly. */ signal?: AbortSignal; + /** + * Per-subscriber backlog cap requested from the daemon. Forwarded as + * `?maxQueued=N` on `GET /session/:id/events`. Daemon-side range is + * `[16, 2048]` (default 256); out-of-range or non-decimal values get + * a `400 invalid_max_queued` response. Old daemons without the + * `slow_client_warning` capability silently ignore the param — SDK + * clients should pre-flight `caps.features.slow_client_warning` + * before opting in. Useful for cold reconnects with a large + * `Last-Event-ID: 0` replay backlog so the force-pushed replay + * frames don't trip the warn / eviction path on the first publish. + */ + maxQueued?: number; } export class DaemonClient { @@ -510,12 +522,19 @@ export class DaemonClient { const fetchSignal = opts.signal ? composeAbortSignals([opts.signal, connectCtrl.signal]) : connectCtrl.signal; + // Build the SSE URL, optionally with `?maxQueued=N`. We don't + // validate the value client-side — the daemon's + // `parseMaxQueuedQuery` is the source of truth on the range + // `[16, 2048]` and returns a structured `400 invalid_max_queued` + // for anything outside, so duplicating the bounds here would + // diverge if the daemon's range ever shifts. + let url = `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/events`; + if (opts.maxQueued !== undefined) { + url += `?maxQueued=${encodeURIComponent(String(opts.maxQueued))}`; + } let res: Response; try { - res = await this._fetch( - `${this.baseUrl}/session/${encodeURIComponent(sessionId)}/events`, - { headers, signal: fetchSignal }, - ); + res = await this._fetch(url, { headers, signal: fetchSignal }); } finally { if (connectTimer !== undefined) clearTimeout(connectTimer); } diff --git a/packages/sdk-typescript/test/unit/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index aa4b188d665..6642f2c7059 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -546,6 +546,47 @@ describe('DaemonClient', () => { const iter = client.subscribeEvents('missing'); await expect(iter.next()).rejects.toMatchObject({ status: 404 }); }); + + it('appends ?maxQueued=N when SubscribeOptions.maxQueued is set', async () => { + const { fetch, calls } = recordingFetch(() => sseResponse('')); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + for await (const _ of client.subscribeEvents('s-1', { + maxQueued: 512, + })) { + /* unreachable */ + } + expect(calls[0]?.url).toBe( + 'http://daemon/session/s-1/events?maxQueued=512', + ); + }); + + it('omits the query string when maxQueued is undefined', async () => { + const { fetch, calls } = recordingFetch(() => sseResponse('')); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + for await (const _ of client.subscribeEvents('s-1', { + lastEventId: 7, + })) { + /* unreachable */ + } + // Bare events URL — no `?` introduced when the caller didn't ask. + expect(calls[0]?.url).toBe('http://daemon/session/s-1/events'); + expect(calls[0]?.headers['last-event-id']).toBe('7'); + }); + + it('propagates a server 400 invalid_max_queued unchanged', async () => { + const { fetch } = recordingFetch(() => + jsonResponse(400, { + error: '`maxQueued` must be in [16, 2048]', + code: 'invalid_max_queued', + }), + ); + const client = new DaemonClient({ baseUrl: 'http://daemon', fetch }); + const iter = client.subscribeEvents('s-1', { maxQueued: 9999 }); + await expect(iter.next()).rejects.toMatchObject({ + status: 400, + body: { code: 'invalid_max_queued' }, + }); + }); }); describe('listWorkspaceSessions', () => { From 7a0224ab1a6414eef1a9c1be686f3aa93cbcb2ed Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Sun, 17 May 2026 17:55:56 +0800 Subject: [PATCH 4/5] refactor(serve): adopt PR #4237 copilot review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address 6 of 8 copilot-reviewer findings on PR #4237; the other 2 (#1 forcedInBuf live-size corruption, #5 SDK lacks maxQueued) were already fixed in bae42c88b — replied on the threads with the commit hash. - **[2] server.ts:1068** — `?maxQueued=` (present-but-empty) now fails closed with `400 invalid_max_queued` instead of silently falling back to the default queue cap. The API documents fail-closed for any malformed value before opening SSE, so an empty string is unambiguously malformed. New server.test.ts case locks this in. - **[3] commands/serve.ts:93** — CLI help text for `--event-ring-size` no longer mis-shapes `Last-Event-ID` as a query parameter. It is an HTTP header, and the daemon's SSE route does not parse a `?Last-Event-ID=` query. - **[4] docs/developers/qwen-serve-protocol.md:351** — clarify that `?maxQueued=N` controls the LIVE-event backlog cap. Replay frames are force-pushed and exempt from the cap; what consumes it is live events that arrive while the subscriber is still draining a cold-reconnect replay. Bumping for cold reconnects is still the right answer, but for the live tail, not for the replay frames themselves. - **[6] eventBus.ts:214** — stale `ringSize=4000` performance comment updated to the new `ringSize=8000` default with a note about the O(n) `shift()` cost scaling. - **[7] sdk-typescript events.ts:492** — `isSlowClientWarningData` now uses the existing `isFiniteNumber` helper instead of bare `typeof === 'number'`. Mirrors the sibling predicates and rejects `NaN` / `Infinity` payloads as schema garbage. New daemonEvents.test.ts assertions cover both. - **[8] server.ts:127** — `createServeApp`'s default-bridge construction now also forwards `opts.eventRingSize` to `createHttpAcpBridge`, symmetric with the `runQwenServe.ts` path. Direct embeds / tests that called `createServeApp` without supplying their own bridge but did pass `ServeOptions.eventRingSize` were silently getting the default 8000 ring. Tests: 326 focused tests across eventBus / bridge / SDK DaemonClient / DaemonSessionClient / daemonEvents / server. All green; the new server.test.ts case + the extended daemonEvents.test.ts assertions cover the tightened guards. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- docs/developers/qwen-serve-protocol.md | 6 ++--- packages/cli/src/commands/serve.ts | 7 +++--- packages/cli/src/serve/eventBus.ts | 12 +++++----- packages/cli/src/serve/server.test.ts | 22 +++++++++++++++++++ packages/cli/src/serve/server.ts | 14 +++++++++++- packages/sdk-typescript/src/daemon/events.ts | 10 ++++++--- .../test/unit/daemonEvents.test.ts | 22 +++++++++++++++++++ 7 files changed, 78 insertions(+), 15 deletions(-) diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index fb89d28d151..35a780e8776 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -346,9 +346,9 @@ Last-Event-ID: 42 ← optional, replays from after id 42 Query params: -| Param | Required | Notes | -| ----------- | -------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `maxQueued` | no | Per-subscriber live-backlog cap. Range `[16, 2048]`, default 256. Bump for cold reconnects with `Last-Event-ID: 0` against a large replay ring so the force-pushed replay frames don't immediately trip eviction. Out-of-range / non-decimal values return `400 invalid_max_queued` before the SSE handshake opens. Pre-flight `caps.features.slow_client_warning` — old daemons silently ignore the param. | +| Param | Required | Notes | +| ----------- | -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `maxQueued` | no | Per-subscriber **live-backlog** cap. Range `[16, 2048]`, default 256. Replay frames force-pushed at subscribe time are exempt from the cap; what actually consumes it is live events that arrive while the subscriber is still draining a large `Last-Event-ID: 0` replay. Bump for cold reconnects so the live tail doesn't trip the slow-client warning / eviction before the consumer catches up. Out-of-range / non-decimal / present-but-empty values return `400 invalid_max_queued` before the SSE handshake opens. Pre-flight `caps.features.slow_client_warning` — old daemons silently ignore the param. | Frame format. The `data:` line is the **full event envelope**, JSON-stringified on a single line — `{id?, v, type, data, originatorClientId?}`. The ACP-specific payload (`sessionUpdate`, `requestPermission` arguments, etc.) sits under the envelope's `data` field; the envelope's own `type` matches the SSE `event:` line. diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index b0452d59171..d43f910b95e 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -90,9 +90,10 @@ export const serveCommand: CommandModule = { default: 8000, description: 'Per-session SSE replay ring depth (#3803 §02 target). Sets the ' + - 'replay backlog available to `GET /session/:id/events?Last-Event-ID=N` ' + - 'reconnects. Larger = more reconnect headroom at the cost of a few ' + - 'hundred KB extra RAM per session. Must be a positive finite integer.', + 'replay backlog available to `GET /session/:id/events` reconnects ' + + 'that send a `Last-Event-ID: N` header. Larger = more reconnect ' + + 'headroom at the cost of a few hundred KB extra RAM per session. ' + + 'Must be a positive finite integer.', }) .option('http-bridge', { type: 'boolean', diff --git a/packages/cli/src/serve/eventBus.ts b/packages/cli/src/serve/eventBus.ts index 5e21b710301..4de3adb6284 100644 --- a/packages/cli/src/serve/eventBus.ts +++ b/packages/cli/src/serve/eventBus.ts @@ -211,11 +211,13 @@ export class EventBus { ...input, }; this.ring.push(event); - // Eviction-by-shift is O(n) once the ring is full. With ringSize=4000 - // and per-publish work measured in hundreds of microseconds even on - // chatty sessions, this isn't a real hotspot today. A circular-buffer - // refactor would push it to O(1) but adds index bookkeeping; deferred - // until profiling actually flags it. + // Eviction-by-shift is O(n) once the ring is full. At the current + // default `ringSize=8000` (#3803 §02) the per-publish shift work + // measures in low milliseconds on chatty sessions — still well + // below per-frame latency budgets. A circular-buffer refactor + // would push it to O(1) but adds index bookkeeping; deferred until + // profiling actually flags it, or the operator bumps + // `--event-ring-size` to an order of magnitude larger. if (this.ring.length > this.ringSize) this.ring.shift(); // Snapshot the subscribers so an in-loop `this.subs.delete(sub)` // (the new immediate-eviction cleanup below) doesn't mutate the diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index bb05259bc61..346e8946da6 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -2157,6 +2157,28 @@ describe('GET /session/:id/events (SSE)', () => { expect(seen).toEqual([undefined]); }); + it('400s a present-but-empty ?maxQueued= before opening the SSE stream', async () => { + // `?maxQueued=` (typed explicitly without a value) is malformed + // and must fail-CLOSED, not silently fall back to the default + // queue cap. Symmetric to non-decimal / out-of-range rejection. + const bridge = fakeBridge({ + subscribeImpl: () => { + throw new Error('bridge must not be touched'); + }, + }); + 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?maxQueued=`, + ); + expect(res.status).toBe(400); + expect(await res.json()).toMatchObject({ code: 'invalid_max_queued' }); + }); + it('400s a non-decimal ?maxQueued before opening the SSE stream', async () => { const bridge = fakeBridge({ subscribeImpl: () => { diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index cbd30f042af..7b93f7f01a1 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -127,6 +127,13 @@ export function createServeApp( deps.bridge ?? createHttpAcpBridge({ maxSessions: opts.maxSessions, + // Symmetric with `runQwenServe.ts` — direct embeds / tests that + // call `createServeApp` without supplying their own bridge and + // pass `ServeOptions.eventRingSize` would otherwise silently + // get the default 8000 ring instead of their configured value. + ...(opts.eventRingSize !== undefined + ? { eventRingSize: opts.eventRingSize } + : {}), boundWorkspace, }); @@ -1065,7 +1072,12 @@ function parseMaxQueuedQuery( raw: unknown, res: import('express').Response, ): number | undefined | null { - if (raw === undefined || raw === '') return undefined; + // Absent param → undefined (use bus default). Present-but-empty + // (`?maxQueued=` typed explicitly) → fail-CLOSED 400 — the API + // documents fail-closed for any malformed value before opening + // SSE, and an empty string is unambiguously malformed (real values + // are positive integers in [16, 2048]). + if (raw === undefined) return undefined; if (typeof raw !== 'string' || !/^\d+$/.test(raw)) { writeStderrLine( `qwen serve: rejected ?maxQueued "${String(raw).slice(0, 80)}" ` + diff --git a/packages/sdk-typescript/src/daemon/events.ts b/packages/sdk-typescript/src/daemon/events.ts index 1612fdd2aeb..506bf2ff570 100644 --- a/packages/sdk-typescript/src/daemon/events.ts +++ b/packages/sdk-typescript/src/daemon/events.ts @@ -485,11 +485,15 @@ function isClientEvictedData(value: unknown): value is DaemonClientEvictedData { function isSlowClientWarningData( value: unknown, ): value is DaemonSlowClientWarningData { + // Mirror the sibling predicates' finite-number guard + // (`isOptionalNumber` → `isFiniteNumber`): `typeof NaN === 'number'` + // and `typeof Infinity === 'number'` both pass a bare `typeof` + // check but would be schema garbage for a queue-size measurement. return ( isRecord(value) && - typeof value['queueSize'] === 'number' && - typeof value['maxQueued'] === 'number' && - typeof value['lastEventId'] === 'number' + isFiniteNumber(value['queueSize']) && + isFiniteNumber(value['maxQueued']) && + isFiniteNumber(value['lastEventId']) ); } diff --git a/packages/sdk-typescript/test/unit/daemonEvents.test.ts b/packages/sdk-typescript/test/unit/daemonEvents.test.ts index 5430fc4ad87..23a72f950a2 100644 --- a/packages/sdk-typescript/test/unit/daemonEvents.test.ts +++ b/packages/sdk-typescript/test/unit/daemonEvents.test.ts @@ -482,6 +482,28 @@ describe('daemon event schema', () => { data: { queueSize: 192, lastEventId: 42 }, }), ).toBeUndefined(); + + // NaN / Infinity pass a bare `typeof === 'number'` check but are + // schema garbage for a queue-size measurement — finite-number + // validation must reject them (sibling predicates do the same). + expect( + asKnownDaemonEvent({ + v: 1, + type: 'slow_client_warning', + data: { queueSize: Number.NaN, maxQueued: 256, lastEventId: 42 }, + }), + ).toBeUndefined(); + expect( + asKnownDaemonEvent({ + v: 1, + type: 'slow_client_warning', + data: { + queueSize: 192, + maxQueued: Number.POSITIVE_INFINITY, + lastEventId: 42, + }, + }), + ).toBeUndefined(); }); it('reduces slow_client_warning into the view state without ending the stream', () => { From b51e04f337300b0656e8fe927c290fa6263c6ae3 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Sun, 17 May 2026 18:33:19 +0800 Subject: [PATCH 5/5] refactor(serve): adopt PR #4237 wenshao round-2 review feedback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Six adopted findings from @wenshao's second review pass on PR #4237. The seventh ([10] forcedInBuf 3rd case invariant) was already fixed in bae42c88b — replied on that thread. - **[9] + [14] server.ts** — Sanitize attacker-controlled values before stderr interpolation in both `parseMaxQueuedQuery` and `parseLastEventId`. New `safeLogValue()` helper uses `JSON.stringify` to escape control characters (`\n`/`\r`/…) so a URL-encoded newline in `?maxQueued=%0a` can't inject extra log lines into journald/Loki/Splunk pipelines. Matches the `workspace_mismatch` sanitization style in `sendBridgeError`. Fixed in both helpers (the sibling pre-existing `parseLastEventId` had the same shape) so the file stays consistent. - **[11] httpAcpBridge.ts** — `!Number.isFinite(eventRingSize)` was redundant: `Number.isInteger(NaN)` and `Number.isInteger(Infinity)` both return `false`, so the sibling `!Number.isInteger` already catches both. Drop the dead guard. - **[12] httpAcpBridge.ts** — Add soft upper bound `MAX_EVENT_RING_SIZE = 1_000_000` on `eventRingSize` to catch operator typos (`--event-ring-size 80000000` vs `8000000`). At ~500 B per `BridgeEvent` an 1M-frame ring already pins ~500 MB per session — well past any realistic workload. Not a security boundary (operator-controlled flag), pure typo defense. Existing bridge construction test extended with an `80_000_000` case. - **[13] commands/serve.ts** — CLI `--event-ring-size` flag now sources its default from `DEFAULT_RING_SIZE` (imported from `serve/eventBus.js`) instead of the hardcoded literal `8000`. Without this, a future bump of the bus default would silently not take effect for daemons launched through the CLI because the flag always overrides — single source of truth fixes that. - **[15] eventBus.ts** — Drop unreachable `event.id ?? this.lastEventId` fallback in the `slow_client_warning` frame. `event` is locally constructed at the top of `publish()` with `id: this.nextId++` and is guaranteed defined. Use `event.id as number` directly + an inline note about the invariant. Tests: 197 (eventBus 20 / bridge 107 / SDK DaemonClient 57 / SDK daemonEvents 14) + 112 server. All green; the new upper-bound bridge case + the existing log assertions pin the changed behaviors. 🤖 Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/cli/src/commands/serve.ts | 7 +++++- packages/cli/src/serve/eventBus.ts | 5 +++- packages/cli/src/serve/httpAcpBridge.test.ts | 5 ++++ packages/cli/src/serve/httpAcpBridge.ts | 20 ++++++++++++--- packages/cli/src/serve/server.ts | 26 +++++++++++++++++--- 5 files changed, 54 insertions(+), 9 deletions(-) diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index d43f910b95e..8ec5c48bc63 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -11,6 +11,7 @@ import type { Argv, CommandModule } from 'yargs'; // with ~50ms of cold ESM resolution. The runtime import is deferred to the // handler below so it only loads when the user actually runs `qwen serve`. import { writeStderrLine } from '../utils/stdioHelpers.js'; +import { DEFAULT_RING_SIZE } from '../serve/eventBus.js'; /** * Pause the current async function indefinitely. Used after the daemon @@ -87,7 +88,11 @@ export const serveCommand: CommandModule = { }) .option('event-ring-size', { type: 'number', - default: 8000, + // Single source of truth — `DEFAULT_RING_SIZE` (currently 8000, + // #3803 §02) is also what the bridge falls back to when the + // option is undefined. Importing here keeps a future bump in + // one place rather than drifting between CLI and bus. + default: DEFAULT_RING_SIZE, description: 'Per-session SSE replay ring depth (#3803 §02 target). Sets the ' + 'replay backlog available to `GET /session/:id/events` reconnects ' + diff --git a/packages/cli/src/serve/eventBus.ts b/packages/cli/src/serve/eventBus.ts index 4de3adb6284..861e02fbc1c 100644 --- a/packages/cli/src/serve/eventBus.ts +++ b/packages/cli/src/serve/eventBus.ts @@ -293,7 +293,10 @@ export class EventBus { data: { queueSize: liveSize, maxQueued: sub.maxQueued, - lastEventId: event.id ?? this.lastEventId, + // `event.id` is always defined here — the just-published + // `event` is constructed at the top of `publish()` with + // `id: this.nextId++`. No `??` fallback needed. + lastEventId: event.id as number, }, }; sub.queue.forcePush(warningFrame); diff --git a/packages/cli/src/serve/httpAcpBridge.test.ts b/packages/cli/src/serve/httpAcpBridge.test.ts index 571480f49cd..12203ebf7c5 100644 --- a/packages/cli/src/serve/httpAcpBridge.test.ts +++ b/packages/cli/src/serve/httpAcpBridge.test.ts @@ -295,6 +295,11 @@ describe('createHttpAcpBridge', () => { expect(() => makeBridge({ eventRingSize: Number.POSITIVE_INFINITY }), ).toThrow(/Invalid eventRingSize/); + // Upper-bound typo defense (1M cap). `80_000_000` here mimics the + // common shell typo `--event-ring-size 80000000` vs `8000000`. + expect(() => makeBridge({ eventRingSize: 80_000_000 })).toThrow( + /Invalid eventRingSize/, + ); }); it('spawns a session and returns the agent-assigned id', async () => { diff --git a/packages/cli/src/serve/httpAcpBridge.ts b/packages/cli/src/serve/httpAcpBridge.ts index 159fc69806f..9dd29e6058c 100644 --- a/packages/cli/src/serve/httpAcpBridge.ts +++ b/packages/cli/src/serve/httpAcpBridge.ts @@ -1162,6 +1162,14 @@ class BridgeClient implements Client { const DEFAULT_INIT_TIMEOUT_MS = 10_000; const DEFAULT_MAX_SESSIONS = 20; +/** + * Soft upper bound on `BridgeOptions.eventRingSize` to catch operator + * typos before they OOM the daemon. At ~500 B per `BridgeEvent` an + * 1 000 000-frame ring already pins ~500 MB per session — well past + * any realistic workload. Not a security boundary (the flag is + * operator-controlled), just typo defense. + */ +const MAX_EVENT_RING_SIZE = 1_000_000; // Bd1yh: per-permission-request wall clock. Without this, an agent // calling `requestPermission` while no SSE subscriber is connected // would hang the per-session FIFO promptQueue forever (the prompt @@ -1219,15 +1227,21 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // `maxSessions`: silently disabling SSE backpressure on a config // typo is worse than failing to start. Unlike `maxSessions` there // is NO unlimited sentinel — an unbounded ring would grow forever. + // Soft upper bound MAX_EVENT_RING_SIZE catches operator typos + // (`--event-ring-size 80000000` instead of `8000000`); at 1M + // frames × ~500 B/frame the per-session ceiling is already + // ~500 MB, well past any legitimate use. const eventRingSize = opts.eventRingSize ?? DEFAULT_RING_SIZE; + // `Number.isInteger` already rejects NaN / Infinity / non-finite + // — no separate `Number.isFinite` guard needed. if ( - !Number.isFinite(eventRingSize) || !Number.isInteger(eventRingSize) || - eventRingSize < 1 + eventRingSize < 1 || + eventRingSize > MAX_EVENT_RING_SIZE ) { throw new TypeError( `Invalid eventRingSize: ${opts.eventRingSize}. ` + - `Must be a positive finite integer.`, + `Must be a positive integer in [1, ${MAX_EVENT_RING_SIZE}].`, ); } const channelFactory = opts.channelFactory ?? defaultSpawnChannelFactory; diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index 7b93f7f01a1..b0dbaa2cb63 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -1079,8 +1079,14 @@ function parseMaxQueuedQuery( // are positive integers in [16, 2048]). if (raw === undefined) return undefined; if (typeof raw !== 'string' || !/^\d+$/.test(raw)) { + // Sanitize via JSON.stringify so an attacker-controlled value + // containing `\n` / `\r` / other control chars can't inject extra + // log lines into stderr (line-based shipper like + // journald/Loki/Splunk would otherwise treat the injected line as + // a fresh entry). Matches the `workspace_mismatch` log style in + // `sendBridgeError`. writeStderrLine( - `qwen serve: rejected ?maxQueued "${String(raw).slice(0, 80)}" ` + + `qwen serve: rejected ?maxQueued ${safeLogValue(raw)} ` + `(not a decimal integer)`, ); res.status(400).json({ @@ -1096,7 +1102,7 @@ function parseMaxQueuedQuery( n > MAX_QUERY_MAX_QUEUED ) { writeStderrLine( - `qwen serve: rejected ?maxQueued "${raw.slice(0, 80)}" ` + + `qwen serve: rejected ?maxQueued ${safeLogValue(raw)} ` + `(outside [${MIN_QUERY_MAX_QUEUED}, ${MAX_QUERY_MAX_QUEUED}])`, ); res.status(400).json({ @@ -1108,6 +1114,18 @@ function parseMaxQueuedQuery( return n; } +/** + * Wrap an attacker-controllable string for safe interpolation into a + * stderr log line. `JSON.stringify` escapes control characters + * (`\n`, `\r`, etc.) and wraps the result in quotes — any injection + * attempt surfaces as visible-as-quoted-noise rather than a + * forged log line. Truncated AFTER stringify to keep the budget + * predictable even for control-heavy inputs. + */ +function safeLogValue(raw: unknown): string { + return JSON.stringify(String(raw)).slice(0, 82); +} + function parseLastEventId(raw: unknown): number | undefined { // Stricter than Number.parseInt: only accept pure decimal digits to avoid // values like "1abc" or "1.5e10z" silently parsing to 1. @@ -1120,7 +1138,7 @@ function parseLastEventId(raw: unknown): number | undefined { // "first connect, no resume"). if (typeof raw === 'string' && raw.length > 0) { writeStderrLine( - `qwen serve: rejected Last-Event-ID "${raw.slice(0, 80)}" ` + + `qwen serve: rejected Last-Event-ID ${safeLogValue(raw)} ` + `(not a decimal integer)`, ); } @@ -1132,7 +1150,7 @@ function parseLastEventId(raw: unknown): number | undefined { // tries to resume from beyond that is either malicious or broken. if (!Number.isFinite(n) || n > Number.MAX_SAFE_INTEGER) { writeStderrLine( - `qwen serve: rejected Last-Event-ID "${raw.slice(0, 80)}" ` + + `qwen serve: rejected Last-Event-ID ${safeLogValue(raw)} ` + `(exceeds Number.MAX_SAFE_INTEGER)`, ); return undefined;