diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 755b982fdac..d2d4d4bfbac 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -91,6 +91,7 @@ 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', + 'slow_client_warning', 'typed_event_schema', 'session_set_model', 'client_identity', 'client_heartbeat', 'session_permission_vote', 'permission_vote'] ``` @@ -99,6 +100,10 @@ registry. Clients **must** gate UI off `features`, not off `mode` (per design `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. + +`typed_event_schema` advertises daemon event payloads that match the SDK's `KnownDaemonEvent` schema. Older daemons may still stream compatible frames, but SDK clients should pre-flight this tag before assuming typed event coverage. + `client_heartbeat` advertises `POST /session/:id/heartbeat`. Older daemons return `404`; pre-flight this tag before issuing periodic heartbeats. ## Routes @@ -375,6 +380,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. 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. ``` @@ -394,28 +405,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 50b644d7bba..623f2464ccd 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..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 @@ -30,6 +31,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 +86,20 @@ 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', + // 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 ' + + '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', default: true, @@ -122,6 +138,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 a61c1c3a917..6a2330a3d39 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' }, // SDK consumers can detect `KnownDaemonEvent` schema support without // pinning against this SDK release — `narrowDaemonEvent` falls back // to `kind: 'unknown'` for daemons that don't advertise the tag, diff --git a/packages/cli/src/serve/eventBus.test.ts b/packages/cli/src/serve/eventBus.test.ts index 063ceed97f4..029526b8f82 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,181 @@ 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('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 }); + // 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 () => { // 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..861e02fbc1c 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,26 @@ 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; + /** + * 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 `warnResetThreshold` (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 @@ -178,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 @@ -219,6 +254,56 @@ 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. + // + // 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 >= sub.warnThreshold) { + sub.warned = true; + const warningFrame: BridgeEvent = { + v: EVENT_SCHEMA_VERSION, + type: 'slow_client_warning', + data: { + queueSize: liveSize, + maxQueued: sub.maxQueued, + // `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); + } 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; } } return event; @@ -253,15 +338,22 @@ 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, + warnThreshold: WARN_THRESHOLD_RATIO * maxQueued, + warnResetThreshold: WARN_RESET_RATIO * maxQueued, + warned: false, + dispose: () => {}, + }; this.subs.add(sub); if (opts.lastEventId !== undefined) { @@ -359,40 +451,55 @@ 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. Backpressure decisions in `EventBus.publish()` (the + * `slow_client_warning` threshold) read this value. + */ + get size(): number { + return this.liveCount; + } + /** Returns true if accepted, false if dropped due to overflow. */ push(value: T): boolean { if (this.closed) return false; @@ -402,12 +509,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(); @@ -415,8 +524,7 @@ class BoundedAsyncQueue { r({ value, done: false }); return; } - this.buf.push(value); - this.forcedInBuf += 1; + this.buf.push({ value, forced: true }); } /** @@ -440,7 +548,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()!({ @@ -455,12 +563,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/cli/src/serve/httpAcpBridge.test.ts b/packages/cli/src/serve/httpAcpBridge.test.ts index 20639d1d043..7196f2a1098 100644 --- a/packages/cli/src/serve/httpAcpBridge.test.ts +++ b/packages/cli/src/serve/httpAcpBridge.test.ts @@ -268,6 +268,40 @@ 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/); + // 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 () => { const handles: ChannelHandle[] = []; const factory: ChannelFactory = async () => { diff --git a/packages/cli/src/serve/httpAcpBridge.ts b/packages/cli/src/serve/httpAcpBridge.ts index c904788320b..c18b5ee7ade 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'; @@ -590,6 +591,22 @@ 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 + * 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; /** * Bd1yh: per-`requestPermission` wall clock. After this many ms with * no client vote, the agent's permission promise resolves as @@ -1248,6 +1265,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 @@ -1301,6 +1326,27 @@ 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. + // 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.isInteger(eventRingSize) || + eventRingSize < 1 || + eventRingSize > MAX_EVENT_RING_SIZE + ) { + throw new TypeError( + `Invalid eventRingSize: ${opts.eventRingSize}. ` + + `Must be a positive integer in [1, ${MAX_EVENT_RING_SIZE}].`, + ); + } const channelFactory = opts.channelFactory ?? defaultSpawnChannelFactory; const initTimeoutMs = opts.initializeTimeoutMs ?? DEFAULT_INIT_TIMEOUT_MS; if (initTimeoutMs <= 0) { @@ -2073,7 +2119,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, @@ -2211,7 +2257,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 c734d0f57f9..2ca97eee836 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -74,6 +74,7 @@ const EXPECTED_STAGE1_FEATURES = [ 'session_prompt', 'session_cancel', 'session_events', + 'slow_client_warning', 'typed_event_schema', 'session_set_model', 'client_identity', @@ -2399,6 +2400,112 @@ 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 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: () => { + 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 33879b17df4..2ffae9c6a8e 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -129,6 +129,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, }); @@ -710,6 +717,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(); @@ -717,6 +730,7 @@ export function createServeApp( const iterable = bridge.subscribeEvents(sessionId, { signal: abort.signal, lastEventId, + ...(maxQueued !== undefined ? { maxQueued } : {}), }); iter = iterable[Symbol.asyncIterator](); } catch (err) { @@ -1102,6 +1116,83 @@ 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 { + // 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)) { + // 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 ${safeLogValue(raw)} ` + + `(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 ${safeLogValue(raw)} ` + + `(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; +} + +/** + * 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. @@ -1114,7 +1205,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)`, ); } @@ -1126,7 +1217,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; 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/DaemonClient.ts b/packages/sdk-typescript/src/daemon/DaemonClient.ts index 75ffd84c987..b7f3d92e8bc 100644 --- a/packages/sdk-typescript/src/daemon/DaemonClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonClient.ts @@ -140,6 +140,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 { @@ -539,12 +551,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/src/daemon/events.ts b/packages/sdk-typescript/src/daemon/events.ts index ad81ff40ed4..828065936a0 100644 --- a/packages/sdk-typescript/src/daemon/events.ts +++ b/packages/sdk-typescript/src/daemon/events.ts @@ -15,6 +15,7 @@ const DAEMON_KNOWN_EVENT_TYPE_VALUES = [ 'model_switch_failed', 'session_died', 'client_evicted', + 'slow_client_warning', 'stream_error', ] as const; @@ -88,6 +89,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; @@ -125,6 +140,10 @@ export type DaemonClientEvictedEvent = DaemonEventEnvelope< 'client_evicted', DaemonClientEvictedData >; +export type DaemonSlowClientWarningEvent = DaemonEventEnvelope< + 'slow_client_warning', + DaemonSlowClientWarningData +>; export type DaemonStreamErrorEvent = DaemonEventEnvelope< 'stream_error', DaemonStreamErrorData @@ -143,6 +162,7 @@ export type DaemonControlEvent = export type DaemonStreamLifecycleEvent = | DaemonClientEvictedEvent + | DaemonSlowClientWarningEvent | DaemonStreamErrorEvent; export type KnownDaemonEvent = @@ -174,6 +194,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( @@ -197,6 +225,8 @@ export function createDaemonSessionViewState( seed.unmatchedPermissionResolutionCount ?? 0, lastUnmatchedPermissionResolutionId: seed.lastUnmatchedPermissionResolutionId, + slowClientWarningCount: seed.slowClientWarningCount ?? 0, + lastSlowClientWarning: seed.lastSlowClientWarning, }; } @@ -250,6 +280,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) @@ -358,6 +392,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, @@ -479,6 +523,21 @@ 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) && + isFiniteNumber(value['queueSize']) && + isFiniteNumber(value['maxQueued']) && + isFiniteNumber(value['lastEventId']) + ); +} + 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 1701fcfd53e..694fd1b63c4 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -51,6 +51,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 07977fa80c7..d9a12933a38 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -52,6 +52,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/DaemonClient.test.ts b/packages/sdk-typescript/test/unit/DaemonClient.test.ts index d7c707d24fc..a17892fe510 100644 --- a/packages/sdk-typescript/test/unit/DaemonClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonClient.test.ts @@ -649,6 +649,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', () => { diff --git a/packages/sdk-typescript/test/unit/daemonEvents.test.ts b/packages/sdk-typescript/test/unit/daemonEvents.test.ts index 28a35fe7307..306e2b5bc48 100644 --- a/packages/sdk-typescript/test/unit/daemonEvents.test.ts +++ b/packages/sdk-typescript/test/unit/daemonEvents.test.ts @@ -518,4 +518,92 @@ 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(); + + // 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', () => { + 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); + }); });