diff --git a/docs/design/2026-07-07-bounded-replay-snapshot-window.md b/docs/design/2026-07-07-bounded-replay-snapshot-window.md new file mode 100644 index 00000000000..57ed0123429 --- /dev/null +++ b/docs/design/2026-07-07-bounded-replay-snapshot-window.md @@ -0,0 +1,73 @@ +# Bounded Replay Snapshot Window + +## Problem + +Live daemon sessions currently retain replay history in memory so `POST /session/:id/load` can inject replay for clients that attach after the session already exists. That replay retention must be bounded independently from the SSE ring: response-mode restore can seed large historical updates in bulk, and completed live turns can accumulate indefinitely in long-running sessions. + +Disk session history remains the authoritative full transcript source. PR-1 only bounds the daemon's live in-memory replay window; it does not add a full-transcript endpoint. + +## Goals + +- Cap retained replay events by serialized bytes per live session, defaulting to 4 MiB and rejecting invalid configuration at boot. +- Apply the cap to both completed live-turn replay segments and response-mode or stream-mode restored historical replay. +- Preserve the existing snapshot wire shape: `compactedReplay`, `liveJournal`, and `lastEventId`. +- Keep at least one real replay event or one completed live-turn segment even when that single unit exceeds the cap. +- Surface truncation with an id-less `history_truncated` marker at the start of `compactedReplay`. +- Treat `history_truncated` as status only. It must not trigger `state_resync_required`, reload loops, or persistence back into the replay window. + +## Non-Goals + +- No cap on a single in-flight live turn in PR-1; `liveJournal` continues to hold the active turn until a boundary. +- No turn-count cap. Turn counts are diagnostic only when the engine can count dropped completed turn segments exactly. +- No `/capabilities` feature tag for this additive event. The resolved limit is exposed in daemon status. +- No complete transcript endpoint. PR-2 must design paginated or streaming transcript reads and must not expose a one-shot full array response. + +## Design + +`TurnBoundaryCompactionEngine` stores retained replay as ordered segments instead of an unbounded flat array. A completed live turn is one segment. Restore/bulk seed replay is stored as event-level segments so the oldest restore events can be discarded independently when the byte cap is exceeded. + +Sizing reuses the EventBus safe JSON sizing semantics. Sizing failure logs diagnostics and counts that event as zero bytes so publish and seed paths keep their never-throws contract. + +When `replayBytes > maxReplayBytes`, the engine drops oldest segments while more than one segment remains. It increments `truncatedEvents`, and increments `truncatedTurns` only for dropped live-turn segments. `snapshot()` flattens retained segments and prepends: + +```json +{ + "type": "history_truncated", + "data": { + "reason": "replay_window_exceeded", + "truncatedEvents": 12, + "retainedEvents": 8, + "maxBytes": 4194304, + "truncatedTurns": 3, + "fullTranscriptAvailable": false + } +} +``` + +The marker is synthetic and id-less. It is excluded from byte accounting and from transient replay retention. `ingest()`, `seed(snapshot)`, and `seedReplayEvents()` all filter it out so loading a bounded snapshot cannot compound markers. + +`EventBus.seedReplayEvents()` assigns ids and timestamps to restore replay events, calls the compaction engine's dedicated seed method, and clears the SSE ring as before. This prevents bulk restore replay from being appended to `liveJournal`. + +The CLI wiring passes one resolved cap through yargs, the fast-path parser, `ServeOptions`, server wiring, `BridgeOptions`, bridge status, and daemon status rendering. Invalid values (`0`, negative, non-integer, `NaN`, `Infinity`, or values above 256 MiB) fail closed. + +SDK and WebUI know `history_truncated`, validate its payload, project it to view-state counters and transcript status, and render a terminal status line. The event is not an unknown/debug event and is not part of resync gating. + +## Audit Notes + +Round 1: A cap only on completed live turns is insufficient because response-mode restore can seed large historical replay without live boundaries. The design therefore adds `seedReplayEvents()` and event-level historical segments. + +Round 2: Reusing `state_resync_required` for truncation would create reload loops because `/load` would keep returning the same bounded window. The design uses a separate status marker that never sets `awaitingResync`. + +Round 3: A turn-count cap does not bound memory when one turn contains large tool output. PR-1 uses byte-only enforcement and leaves active-turn capping out of scope. + +Round 4: Returning the full transcript as an array would recreate the same peak memory problem at request time. PR-2 is explicitly constrained to pagination or streaming. + +Round 5: Empty replay after truncation would make clients lose all visible state. The engine preserves the newest segment even when oversized. + +## Verification Plan + +- Unit-test live turn trimming, restore seed trimming, marker placement, transient marker filtering, oversized latest retention, safe sizing failure, and EventBus never-throws behavior. +- Unit-test bridge response-mode restore and live-session load behavior with the bounded window. +- Unit-test CLI parsing, fast-path parsing, runQwenServe validation, server bridge wiring, and daemon status limits. +- Unit-test SDK known-event validation, reducer state, UI normalizer, transcript status, terminal rendering, and WebUI replay injection. +- Keep final verification on `npm run build`, `npm run typecheck`, and `npm run lint`. diff --git a/docs/developers/daemon/01-architecture.md b/docs/developers/daemon/01-architecture.md index dbeaf3fc186..aad97e62a5f 100644 --- a/docs/developers/daemon/01-architecture.md +++ b/docs/developers/daemon/01-architecture.md @@ -196,7 +196,7 @@ sequenceDiagram Note over EB,SR: If subscriber queue >= maxQueued,
EventBus emits client_evicted terminal frame
and closes subscriber. ``` -The ring buffer is bounded (`eventRingSize`, default 8000). A reconnecting client whose `Last-Event-ID` is older than the ring's head receives a synthetic catch-up signal and must call `loadSession` / `resumeSession` to rebuild deeper state. Slow clients trigger `slow_client_warning` at 75% queue fill and `client_evicted` at the cap. +The ring buffer is bounded (`eventRingSize`, default 8000). A reconnecting client whose `Last-Event-ID` is older than the ring's head receives `state_resync_required` and must rebuild from `loadSession`'s bounded replay snapshot window or use `resumeSession` when it already has local history. Slow clients trigger `slow_client_warning` at 75% queue fill and `client_evicted` at the cap. ## Workflow 3: Multi-client permission mediation @@ -327,7 +327,7 @@ The two-phase shutdown matters because in-flight HTTP requests, in-flight SSE su | Concern | File | | -------------------- | ----------------------------------------------------------- | -| Bootstrap | `packages/cli/src/serve/run-qwen-serve.ts` | +| Bootstrap | `packages/cli/src/serve/run-qwen-serve.ts` | | Express app | `packages/cli/src/serve/server.ts` | | Capability registry | `packages/cli/src/serve/capabilities.ts` | | Auth middleware | `packages/cli/src/serve/auth.ts` | diff --git a/docs/developers/daemon/03-acp-bridge.md b/docs/developers/daemon/03-acp-bridge.md index 173a71d7364..7ba78b1fa7e 100644 --- a/docs/developers/daemon/03-acp-bridge.md +++ b/docs/developers/daemon/03-acp-bridge.md @@ -242,9 +242,13 @@ In addition to the core `spawnOrAttach`, `sendPrompt`, `cancelSession`, `BridgeSpawnRequest.sessionScope` was renamed from `'per-client'` to `'thread'`. `BridgeRestoredSession` now carries `compactedReplay`, -`liveJournal`, and `lastEventId`. `BridgeClientRequestContext` is the request -context threaded through bridge calls; it carries `clientId`, -`fromLoopback: boolean`, and `promptId`. +`liveJournal`, and `lastEventId`. Those replay fields are a bounded in-memory +window for live sessions, capped by `BridgeOptions.compactedReplayMaxBytes` +(default 4 MiB, hard ceiling 256 MiB). If older retained replay was dropped, +`compactedReplay[0]` is the id-less `history_truncated` marker. The full +persisted transcript remains on disk and is not exposed by this bridge response. +`BridgeClientRequestContext` is the request context threaded through bridge +calls; it carries `clientId`, `fromLoopback: boolean`, and `promptId`. ## Caveats & Known Limits diff --git a/docs/developers/daemon/08-session-lifecycle.md b/docs/developers/daemon/08-session-lifecycle.md index 9538fbbb351..290e2a42edc 100644 --- a/docs/developers/daemon/08-session-lifecycle.md +++ b/docs/developers/daemon/08-session-lifecycle.md @@ -100,7 +100,7 @@ sequenceDiagram ### Load / resume -`POST /session/:id/load` — replays full ACP history (`session/load` notifications fire before the response returns). +`POST /session/:id/load` — restores a persisted session and returns the current bounded replay snapshot window (`session/load` notifications or response-mode replay are seeded before the response returns). `POST /session/:id/resume` — restores without replay (`connection.unstable_resumeSession`, exposed under the stable `session_resume` daemon capability; `unstable_session_resume` remains a deprecated alias). Both: @@ -259,11 +259,19 @@ not as a transport error. `POST /session/:id/load` now returns a `BridgeRestoredSession` that can include `compactedReplay?: BridgeEvent[]`, `liveJournal?: BridgeEvent[]`, and -`lastEventId?: number`. `compactedReplay` is produced by +`lastEventId?: number`. These fields are the daemon's bounded in-memory replay +window for a live session, not a full transcript API. The default window cap is +4 MiB per live session (`--compacted-replay-max-bytes`), and boot rejects +invalid caps; the hard ceiling is 256 MiB. `compactedReplay` is produced by `TurnBoundaryCompactionEngine`: at turn boundaries it folds consecutive text / thought blocks, collapses tool-call sequences to their final state, discards transient signals, and produces O(turns) replay logs instead of O(tokens) logs -(typically a 25-30x reduction). +(typically a 25-30x reduction). When older replay entries have been dropped +from that byte window, `compactedReplay[0]` is a synthetic id-less +`history_truncated` marker with `{reason: 'replay_window_exceeded', +truncatedEvents, retainedEvents, maxBytes, truncatedTurns?, +fullTranscriptAvailable: false}`. Clients should render it as status and apply +the retained replay normally; it must not trigger a resync loop. ### ACP Child Preheat diff --git a/docs/developers/daemon/09-event-schema.md b/docs/developers/daemon/09-event-schema.md index 5c4108ecccd..e52f27ef3e9 100644 --- a/docs/developers/daemon/09-event-schema.md +++ b/docs/developers/daemon/09-event-schema.md @@ -2,7 +2,7 @@ ## Overview -Every SSE frame emitted by the daemon on `GET /session/:id/events` has the shape `{ id, v, type, data, originatorClientId?, _meta? }`. `v: 1` is the current `EVENT_SCHEMA_VERSION`. `type` comes from the closed, version-pinned `DAEMON_KNOWN_EVENT_TYPE_VALUES` set in `packages/sdk-typescript/src/daemon/events.ts`; the current set has 47 known event types. The envelope `_meta` field is stamped at the SSE write boundary by `formatSseFrame()` in `packages/cli/src/serve/routes/sse-events.ts`; see [Envelope-level metadata](#envelope-level-metadata). +Every SSE frame emitted by the daemon on `GET /session/:id/events` has the shape `{ id, v, type, data, originatorClientId?, _meta? }`. `v: 1` is the current `EVENT_SCHEMA_VERSION`. `type` comes from the closed, version-pinned `DAEMON_KNOWN_EVENT_TYPE_VALUES` set in `packages/sdk-typescript/src/daemon/events.ts`. The envelope `_meta` field is stamped at the SSE write boundary by `formatSseFrame()` in `packages/cli/src/serve/routes/sse-events.ts`; see [Envelope-level metadata](#envelope-level-metadata). The SDK exposes `asKnownDaemonEvent(evt)`. It returns a discriminated `KnownDaemonEvent` for known event types and `undefined` for other types. SDK consumers can therefore handle forward compatibility without requiring a lockstep SDK upgrade when a newer daemon adds an event type; the session reducer records those as `unrecognizedKnownEventCount`. @@ -15,7 +15,7 @@ The wire format lives in [`../qwen-serve-protocol.md`](../qwen-serve-protocol.md - Provide pure reducers (`reduceDaemonSessionEvent`, `reduceDaemonAuthEvent`) that project an event stream into SDK view state. - Broadcast the `typed_event_schema` capability tag as an informational signal. If the tag is absent, `asKnownDaemonEvent` still falls back to `unknown`. -## Event vocabulary (47 known types) +## Event vocabulary Grouped by domain. @@ -37,6 +37,7 @@ Grouped by domain. | `slow_client_warning` | Live frame backlog or live serialized-byte backlog >= 75%; force-pushed and **has no `id`** | `queueSize, maxQueued, lastEventId, queuedBytes?, maxQueuedBytes?, threshold?: 'frames' \| 'bytes' \| 'frames_and_bytes'`; re-armed after both frame and byte measurements drop below 37.5%. | | `stream_error` | `SubscriberLimitExceededError` or another route stream error | `error: string`; terminal for the subscription. | | `state_resync_required` | `subscribe({lastEventId})` detects that the daemon ring no longer holds `[lastEventId+1, earliestInRing-1]`, or the client cursor is from a previous bus epoch. Force-pushed **before** remaining replay frames and **has no `id`**. | `reason: 'ring_evicted' \| 'epoch_reset' \| string`, `lastDeliveredId: number`, `earliestAvailableId: number`. This is a recovery signal, not terminal: the SSE stream stays open and replay + live frames continue. The SDK reducer sets `awaitingResync = true` and skips deltas until the caller resets with `loadSession`. | +| `history_truncated` | `POST /session/:id/load` returns a bounded replay snapshot after older in-memory replay entries were dropped. Prepended to `compactedReplay` and **has no `id`**. | `reason: 'replay_window_exceeded'`, `truncatedEvents: number`, `retainedEvents: number`, `maxBytes: number`, `truncatedTurns?: number`, `fullTranscriptAvailable: false`. This is a status marker, not a resync request; clients render it and continue applying retained replay. | | `replay_complete` | Id-less sentinel emitted after the `Last-Event-ID` replay loop finishes, for both clean replay and ring-evicted paths, even when `data.replayedCount === 0`. **No `id`** | `replayedCount: number`; lets consumers remove catch-up UI deterministically without a timeout. | ### Permissions (F3 + base) diff --git a/docs/developers/daemon/10-event-bus.md b/docs/developers/daemon/10-event-bus.md index 656525af08f..251e2bde6bf 100644 --- a/docs/developers/daemon/10-event-bus.md +++ b/docs/developers/daemon/10-event-bus.md @@ -239,13 +239,13 @@ The daemon's `EventBus` replays all events from the ring buffer whose `id > Last ### Replay Behavior -| Scenario | Behavior | -| -------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `Last-Event-ID` absent | Live-only stream; no replay. Backward-compatible with pre-resume clients. | -| `Last-Event-ID: 0` | Replay entire ring buffer from the beginning (bounded by `--event-ring-size`, default 8000). | -| `Last-Event-ID: N` where `ring[0].id <= N+1` | Contiguous replay of events `id > N`, then live. | -| `Last-Event-ID: N` where `ring[0].id > N+1` | Gap detected — `state_resync_required` (`reason: 'ring_evicted'`) emitted before replay of surviving suffix. SDK must call `loadSession` to recover full state. | -| `Last-Event-ID: N` where `N >= nextId` | Epoch reset (daemon restart) — `state_resync_required` (`reason: 'epoch_reset'`) emitted, then full ring replay. | +| Scenario | Behavior | +| -------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `Last-Event-ID` absent | Live-only stream; no replay. Backward-compatible with pre-resume clients. | +| `Last-Event-ID: 0` | Replay entire ring buffer from the beginning (bounded by `--event-ring-size`, default 8000). | +| `Last-Event-ID: N` where `ring[0].id <= N+1` | Contiguous replay of events `id > N`, then live. | +| `Last-Event-ID: N` where `ring[0].id > N+1` | Gap detected — `state_resync_required` (`reason: 'ring_evicted'`) emitted before replay of surviving suffix. SDK must call `loadSession` to recover a bounded replay snapshot window; the returned `compactedReplay` may begin with `history_truncated` if older in-memory replay entries were dropped. | +| `Last-Event-ID: N` where `N >= nextId` | Epoch reset (daemon restart) — `state_resync_required` (`reason: 'epoch_reset'`) emitted, then full ring replay. | ### Validation Rules diff --git a/docs/developers/daemon/13-sdk-daemon-client.md b/docs/developers/daemon/13-sdk-daemon-client.md index 46430fffdf4..7e70252f720 100644 --- a/docs/developers/daemon/13-sdk-daemon-client.md +++ b/docs/developers/daemon/13-sdk-daemon-client.md @@ -302,10 +302,14 @@ async function* subscribe(sessionId: string, signal: AbortSignal) { } // Handle ring-eviction gap. if (event.type === 'state_resync_required') { - // State is stale — reload full session state. + // State is stale — reload the daemon's bounded replay snapshot window. await client.loadSession(sessionId); continue; } + if (event.type === 'history_truncated') { + // Informational only. Render a status notice, then continue applying + // the retained replay events; do not trigger another reload. + } yield event; } } @@ -336,7 +340,7 @@ async function resilientSubscribe(session: DaemonSessionClient) { } ``` -On reconnect the daemon replays events with `id > lastSeenEventId` from its bounded ring (default 8000 events). If the gap exceeds the ring, a `state_resync_required` frame signals the client to call `loadSession` for a full state rebuild. +On reconnect the daemon replays events with `id > lastSeenEventId` from its bounded ring (default 8000 events). If the gap exceeds the ring, a `state_resync_required` frame signals the client to call `loadSession` and rebuild from the current bounded replay snapshot window. That snapshot may begin with `history_truncated`; treat it as an operator-visible status marker, not as another resync request. ### Seeding `lastEventId` at Construction diff --git a/docs/developers/daemon/14-cli-tui-adapter.md b/docs/developers/daemon/14-cli-tui-adapter.md index 9d5e05aa69b..819724e4e31 100644 --- a/docs/developers/daemon/14-cli-tui-adapter.md +++ b/docs/developers/daemon/14-cli-tui-adapter.md @@ -128,7 +128,7 @@ Hosts can stop at `(E)` and implement their own reducer, or consume `(G)` and th ### `state_resync_required` -`session.state_resync_required` maps to a transcript "missed range" marker. UI code can call `formatMissedRange(state)` to render text such as "missed events X-Y". The reducer **continues applying later events**, but marks affected blocks with `resyncRecovery: true` so renderers can add visual context. See [`10-event-bus.md`](./10-event-bus.md) for ring-eviction and `state_resync_required` semantics. +`session.state_resync_required` maps to a transcript "missed range" marker. UI code can call `formatMissedRange(state)` to render text such as "missed events X-Y". The reducer sets `awaitingResync` and skips ordinary delta events until consumer code reloads the session's bounded replay snapshot window and clears the latch. A loaded snapshot may start with `history_truncated`; that marker renders as status only and must not start another resync loop. See [`10-event-bus.md`](./10-event-bus.md) for ring-eviction and `state_resync_required` semantics. ## Consumers diff --git a/docs/developers/daemon/17-configuration.md b/docs/developers/daemon/17-configuration.md index fb1c6c8dce3..4a3c5bfd395 100644 --- a/docs/developers/daemon/17-configuration.md +++ b/docs/developers/daemon/17-configuration.md @@ -18,6 +18,7 @@ This page collects every setting that affects the `qwen serve` daemon and its ad | `--max-connections ` | number | `256` | HTTP listener `server.maxConnections`; `0` / `Infinity` means unlimited. | | `--enable-session-shell` | boolean | `false` | Enables direct `POST /session/:id/shell` execution. Requires bearer token, and every call must carry a session-bound `X-Qwen-Client-Id`. | | `--event-ring-size ` | number | `8000` | Per-session SSE replay ring; soft cap is `1_000_000`. | +| `--compacted-replay-max-bytes ` | positive integer | `4194304` | Byte cap for the bounded in-memory replay snapshot returned by `POST /session/:id/load`; hard cap is `268435456`. | | `--http-bridge` | boolean | `true` | Stage 1 bridge mode. `--no-http-bridge` still falls back to http-bridge and prints to stderr. | | `--mcp-client-budget ` | positive integer | unset | Sets `WorkspaceMcpBudget.clientBudget` and forwards it to the ACP child through `childEnvOverrides`. | | `--mcp-budget-mode ` | `off` / `warn` / `enforce` | `warn` when budget is set, otherwise `off` | Sets `WorkspaceMcpBudget.mode`; `enforce` requires `--mcp-client-budget`. | diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 793792c9e3b..0b710a7e379 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -300,6 +300,7 @@ Response shape: "maxPendingPromptsPerSession": 5, "listenerMaxConnections": 256, "eventRingSize": 8000, + "compactedReplayMaxBytes": 4194304, "promptDeadlineMs": null, "writerIdleTimeoutMs": null, "channelIdleTimeoutMs": 0, @@ -1259,7 +1260,7 @@ Response: `attached: true` means the session was already live (either from a prior `session/load`/`session/resume`, or because a coalesced concurrent caller raced just ahead). -**History replay over SSE.** While `loadSession` is in flight on the agent side, the agent emits `session_update` notifications for every persisted turn. The daemon buffers them onto the session's event-bus before the route response returns, so subscribers that immediately call `GET /session/:id/events` with `Last-Event-ID: 0` see the full replay. **The replay ring is bounded** (default 8000 frames per session). Long histories with many tool-call / thought-stream turns can exceed that — the oldest frames are dropped silently. Clients that need full history should subscribe immediately after `load` returns; alternatively they can persist the SSE event ids and use `Last-Event-ID` to resume from a later turn boundary. +**History replay over SSE.** While `loadSession` is in flight on the agent side, the agent may emit `session_update` notifications for persisted turns, or return bulk replay updates in the response metadata. The daemon seeds those events into the session's bounded replay snapshot window before the route response returns. For live sessions, `POST /session/:id/load` only promises that bounded window (`compactedReplay`, `liveJournal`, `lastEventId`), not the full transcript. The window is byte-capped by `--compacted-replay-max-bytes` (default 4 MiB, maximum 256 MiB); if older replay entries were dropped, `compactedReplay[0]` is an id-less `history_truncated` marker. Clients should render that marker as status and continue applying retained events. Full transcript access must use a future paginated or streaming endpoint rather than a single array response. **Errors:** @@ -1857,8 +1858,8 @@ The SSE-level `id:` / `event:` lines duplicate `envelope.id` / `envelope.type` f Reconnect semantics: -- 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. +- 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:** if `` predates the oldest event still in the ring, the daemon emits an id-less `state_resync_required` frame before replaying the surviving suffix. The SDK latches `awaitingResync`; clients should call `POST /session/:id/load` and rebuild from the current bounded replay snapshot window. That snapshot may itself start with `history_truncated` when older in-memory replay entries were dropped; this marker is informational and must not start another resync loop. - IDs are monotonic per session, starting at 1 - Synthetic frames (`client_evicted`, `slow_client_warning`, `stream_error`) intentionally omit `id` so they don't burn a sequence slot for other subscribers diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index 9f249e46ecc..a8d42092de2 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -309,6 +309,7 @@ Notes: | `--channel ` | — | Experimental daemon-managed channel worker. Repeat the flag to select multiple configured channels, or pass `all` to start every configured channel. `all` cannot be combined with named channels. Selected channel `cwd` values must resolve to the daemon workspace. The worker is owned by `qwen serve`; stop the daemon to stop serve-managed channels. | | `--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`. | +| `--compacted-replay-max-bytes ` | `4194304` | Per-live-session byte cap for the retained replay events in the bounded snapshot returned by `POST /session/:id/load`. The cap applies to `compactedReplay`; the current in-flight `liveJournal` remains uncapped. Values must be positive safe integers; invalid values fail at boot, and the hard ceiling is 256 MiB. When older retained replay is dropped, the snapshot begins with `history_truncated`. This does not limit the on-disk transcript. | | `--mcp-client-budget ` | — | Positive integer cap on live MCP clients **per ACP session** (issue [#4175](https://github.com/QwenLM/qwen-code/issues/4175) PR 14 v1; PR 23 graduates this to per-workspace via the shared MCP pool). Combine with `--mcp-budget-mode`. When unset, no accounting-driven enforcement (but `GET /workspace/mcp` still reports `clientCount`). Distinct from claude-code's `MCP_SERVER_CONNECTION_BATCH_SIZE` which gates startup concurrency, not the total client count. Pre-flight `caps.features.mcp_guardrails`. | | `--mcp-budget-mode ` | `warn` / `off` | How `--mcp-client-budget` is enforced. `warn` (default when budget set): no refusal, snapshot's `budgets[0].status` flips to `warning` at ≥75% of budget. `enforce`: connects past the cap are refused, per-server cell shows `disabledReason: 'budget'`, deterministic by `mcpServers` declaration order. `off` (default when budget unset): pure observability. Boot rejects `enforce` without a budget. | | `--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. | @@ -421,10 +422,10 @@ To handle multiple **users** (each with their own quota, audit log, sandbox) or The daemon exposes ACP's `session/load` and resume flow over HTTP via two routes: -| Route | Use when | -| -------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `POST /session/:id/load` | The client has **no** history rendered (cold reconnect, picker-then-open). The daemon replays every persisted turn through SSE so subscribers see the full transcript. Capability tag: `session_load`. | -| `POST /session/:id/resume` | The client already has the turns on screen and only needs the daemon-side handle back. Model context is restored on the agent side without UI replay — the SSE stream stays clean. Capability tag: `session_resume` (`unstable_session_resume` remains a deprecated alias for older clients). | +| Route | Use when | +| -------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| `POST /session/:id/load` | The client has **no** useful local history rendered (cold reconnect, picker-then-open). For a live session, the daemon returns and injects the current bounded replay snapshot window; if older replay was dropped, the snapshot begins with `history_truncated`. Capability tag: `session_load`. | +| `POST /session/:id/resume` | The client already has the turns on screen and only needs the daemon-side handle back. Model context is restored on the agent side without UI replay — the SSE stream stays clean. Capability tag: `session_resume` (`unstable_session_resume` remains a deprecated alias for older clients). | The TypeScript SDK exposes both as static factories on `DaemonSessionClient`: @@ -433,7 +434,7 @@ import { DaemonClient, DaemonSessionClient } from '@qwen-code/sdk'; const client = new DaemonClient({ baseUrl: 'http://127.0.0.1:4170' }); -// Cold reconnect — daemon will replay history through SSE. +// Cold reconnect — daemon will replay the bounded snapshot window through SSE. const session = await DaemonSessionClient.load(client, 'persisted-id'); // Or, if your UI already has the history, skip the replay: @@ -447,7 +448,7 @@ for await (const event of session.events()) { Pre-flight `caps.features.session_load` / `caps.features.session_resume` before calling — older daemons return `404`. `unstable_session_resume` is still advertised as a deprecated compatibility alias. Concurrent same-action requests for the same id coalesce; cross-action races (a `load` racing a `resume`) get `409 restore_in_progress` with `Retry-After: 5`. See the [protocol reference](../developers/qwen-serve-protocol.md) for the full error envelope. -Note: history replay is bounded by the SSE ring (default 8000 frames). Long histories with chatty turns can exceed that — earliest frames are dropped silently. For very long sessions, prefer `resume` and rely on the client's local persisted UI. +Note: live-session history replay is bounded twice: by the SSE ring for `Last-Event-ID` reconnects and by `--compacted-replay-max-bytes` for the snapshot returned by `POST /session/:id/load`. Long histories with chatty turns can exceed either bound. The daemon surfaces snapshot truncation with `history_truncated`; full transcript access is intentionally not a one-shot array response in this API. For very long sessions, prefer `resume` and rely on the client's local persisted UI until a paginated or streaming transcript endpoint is available. ## Durability model @@ -455,7 +456,7 @@ Note: history replay is bounded by the SSE ring (default 8000 frames). Long hist - A child process crash publishes `session_died` and removes the live session from the daemon's maps. The persisted on-disk session **can** be reloaded via `POST /session/:id/load` if a fresh agent child is spawnable. - A daemon restart loses every in-flight live session. The persisted sessions remain on disk and can be loaded against a new daemon process, subject to the same workspace binding rules. -- Long client disconnects (>5 min on a chatty turn) can outrun the SSE replay ring (default 8000 frames) — `Last-Event-ID` reconnect succeeds but state may be incoherent. For mobile / flaky-network clients, plan to re-open SSE on long drops or call `POST /session/:id/load` to replay from disk. +- Long client disconnects (>5 min on a chatty turn) can outrun the SSE replay ring (default 8000 frames) — `Last-Event-ID` reconnect triggers `state_resync_required`. For mobile / flaky-network clients, plan to re-open SSE on long drops or call `POST /session/:id/load` to recover the current bounded replay snapshot; do not assume that route returns the full transcript. - File operations (`writeTextFile`) are atomic across crashes (write-then-rename); they aren't atomic across daemon restarts in the sense of replaying — the file write either landed or it didn't. If your integration needs server-side cross-restart durability beyond what `session/load` covers (e.g. server-managed retry queues), you still need application-level state recovery. Don't hold long-running, restart-sensitive state inside the daemon's session. diff --git a/packages/acp-bridge/package.json b/packages/acp-bridge/package.json index 4dee541dce9..13d3a59812f 100644 --- a/packages/acp-bridge/package.json +++ b/packages/acp-bridge/package.json @@ -59,6 +59,10 @@ "types": "./dist/bridgeOptions.d.ts", "import": "./dist/bridgeOptions.js" }, + "./replayWindowLimits": { + "types": "./dist/replayWindowLimits.d.ts", + "import": "./dist/replayWindowLimits.js" + }, "./spawnChannel": { "types": "./dist/spawnChannel.d.ts", "import": "./dist/spawnChannel.js" diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 7e211489847..7b0c93afaa0 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -111,6 +111,28 @@ describe('createAcpSessionBridge', () => { ); }); + it('accepts and rejects BridgeOptions.compactedReplayMaxBytes at construction time', () => { + expect(() => makeBridge({ compactedReplayMaxBytes: 1 })).not.toThrow(); + expect(() => + makeBridge({ compactedReplayMaxBytes: 4 * 1024 * 1024 }), + ).not.toThrow(); + expect(() => makeBridge({ compactedReplayMaxBytes: 0 })).toThrow( + /compactedReplayMaxBytes/, + ); + expect(() => makeBridge({ compactedReplayMaxBytes: -1 })).toThrow( + /compactedReplayMaxBytes/, + ); + expect(() => makeBridge({ compactedReplayMaxBytes: 1.5 })).toThrow( + /compactedReplayMaxBytes/, + ); + expect(() => + makeBridge({ compactedReplayMaxBytes: Number.POSITIVE_INFINITY }), + ).toThrow(/compactedReplayMaxBytes/); + expect(() => + makeBridge({ compactedReplayMaxBytes: 512 * 1024 * 1024 }), + ).toThrow(/compactedReplayMaxBytes/); + }); + it('sanitizes client artifact provenance fields', async () => { const bridge = makeBridge({ channelFactory: async () => makeChannel().channel, @@ -1298,9 +1320,9 @@ describe('createAcpSessionBridge', () => { expect(loaded.partial).toBe(true); expect(loaded.replayError).toBe('replay boom'); expect(loaded.lastEventId).toBe(2); - expect(loaded.compactedReplay).toEqual([]); - expect(loaded.liveJournal).toHaveLength(2); - expect(loaded.liveJournal?.[0]?._meta?.['serverTimestamp']).toBe( + expect(loaded.compactedReplay).toHaveLength(2); + expect(loaded.liveJournal).toEqual([]); + expect(loaded.compactedReplay?.[0]?._meta?.['serverTimestamp']).toBe( 1_700_000_000_000, ); @@ -1316,6 +1338,56 @@ describe('createAcpSessionBridge', () => { await bridge.shutdown(); }); + it('bounds response-mode load replay and emits a history_truncated marker', async () => { + const factory: ChannelFactory = async () => + makeChannel({ + loadSessionImpl: () => ({ + _meta: { + 'qwen.session.loadReplay': { + v: 1, + updates: [ + { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: `old-${'x'.repeat(600)}` }, + }, + { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: `new-${'y'.repeat(600)}` }, + }, + ], + }, + }, + }), + }).channel; + const bridge = makeBridge({ + channelFactory: factory, + compactedReplayMaxBytes: 512, + }); + + const loaded = await bridge.loadSession({ + sessionId: 'persisted-bounded-history', + workspaceCwd: WS_A, + historyReplay: 'response', + }); + + expect(loaded.lastEventId).toBe(2); + expect(loaded.liveJournal).toEqual([]); + expect(loaded.compactedReplay?.[0]?.type).toBe('history_truncated'); + expect(loaded.compactedReplay?.[0]?.data).toMatchObject({ + reason: 'replay_window_exceeded', + truncatedEvents: 1, + retainedEvents: 1, + maxBytes: 512, + fullTranscriptAvailable: false, + }); + const retained = loaded.compactedReplay?.[1]?.data as { + update?: { content?: { text?: string } }; + }; + expect(retained.update?.content?.text).toBe(`new-${'y'.repeat(600)}`); + + await bridge.shutdown(); + }); + it('rejects oversized response-mode load replay payloads', async () => { const factory: ChannelFactory = async () => makeChannel({ @@ -1485,12 +1557,14 @@ describe('createAcpSessionBridge', () => { historyReplay: 'response', }); - expect(loaded.liveJournal?.map((event) => event.type)).toEqual([ + expect(loaded.compactedReplay?.map((event) => event.type)).toEqual([ 'session_update', + ]); + expect(loaded.liveJournal?.map((event) => event.type)).toEqual([ 'mcp_budget_warning', ]); - expect(loaded.liveJournal?.[0]?.id).toBe(1); - expect(loaded.liveJournal?.[1]?.id).toBe(2); + expect(loaded.compactedReplay?.[0]?.id).toBe(1); + expect(loaded.liveJournal?.[0]?.id).toBe(2); await bridge.shutdown(); }); diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 41c9890a2dd..b18ddd8c7eb 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -33,7 +33,10 @@ import { EVENT_SCHEMA_VERSION, type BridgeEvent, } from './eventBus.js'; -import { TurnBoundaryCompactionEngine } from './compactionEngine.js'; +import { + normalizeCompactedReplayMaxBytes, + TurnBoundaryCompactionEngine, +} from './compactionEngine.js'; import { BridgeChannelClosedError, BridgeTimeoutError, @@ -1135,6 +1138,9 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { `Must be a positive integer in [1, ${MAX_EVENT_RING_SIZE}].`, ); } + const compactedReplayMaxBytes = normalizeCompactedReplayMaxBytes( + opts.compactedReplayMaxBytes, + ); const channelFactory = opts.channelFactory ?? defaultSpawnChannelFactory; // Close over a per-handle env-override snapshot. Calls to // `channelFactory` at spawn time receive this as the 2nd arg, so @@ -2456,7 +2462,18 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { }; const createSessionEventBus = (): EventBus => - new EventBus(eventRingSize, undefined, new TurnBoundaryCompactionEngine()); + new EventBus( + eventRingSize, + undefined, + new TurnBoundaryCompactionEngine({ + maxReplayBytes: compactedReplayMaxBytes, + onReplayWindowEviction: (eviction) => { + teeServeDebugLine( + `replay window evicted ${JSON.stringify(eviction)}`, + ); + }, + }), + ); // §2.3 publish helpers — centralise cache + generation + bus publish so // every `model_switched` / `approval_mode_changed` site stays atomic. @@ -3250,6 +3267,7 @@ export function createAcpSessionBridge(opts: BridgeOptions): AcpSessionBridge { ? null : maxPendingPromptsPerSession, eventRingSize, + compactedReplayMaxBytes, channelIdleTimeoutMs: resolvedChannelIdleTimeoutMs(), sessionIdleTimeoutMs, }, diff --git a/packages/acp-bridge/src/bridgeOptions.ts b/packages/acp-bridge/src/bridgeOptions.ts index a58571a612c..816a4b942aa 100644 --- a/packages/acp-bridge/src/bridgeOptions.ts +++ b/packages/acp-bridge/src/bridgeOptions.ts @@ -184,6 +184,14 @@ export interface BridgeOptions { * `ringSize × average-event-size` held until the session ends. */ eventRingSize?: number; + /** + * Per-session cap, in serialized bytes, for the in-memory compacted replay + * snapshot returned by `session/load` late attach. This bounds daemon heap + * retained for historical replay; the current unfinished live turn remains in + * `liveJournal` until its turn boundary. Defaults to 4 MiB. Must be a + * positive safe integer; there is no unlimited sentinel. + */ + compactedReplayMaxBytes?: number; /** * Per-`requestPermission` wall clock. After this many ms with * no client vote, the agent's permission promise resolves as diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 70da34dc1a8..f34397ccae1 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -371,6 +371,7 @@ export interface BridgeDaemonStatusLimits { maxSessions: number | null; maxPendingPromptsPerSession: number | null; eventRingSize: number; + compactedReplayMaxBytes: number; channelIdleTimeoutMs: number; sessionIdleTimeoutMs: number; } diff --git a/packages/acp-bridge/src/compactionEngine.test.ts b/packages/acp-bridge/src/compactionEngine.test.ts index e69121ebfa8..5f316fd4101 100644 --- a/packages/acp-bridge/src/compactionEngine.test.ts +++ b/packages/acp-bridge/src/compactionEngine.test.ts @@ -4,7 +4,7 @@ * SPDX-License-Identifier: Apache-2.0 */ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import { TurnBoundaryCompactionEngine } from './compactionEngine.js'; import { EventBus } from './eventBus.js'; import type { BridgeEvent } from './eventBus.js'; @@ -354,6 +354,134 @@ describe('TurnBoundaryCompactionEngine', () => { expect(snap.compactedTurns).toHaveLength(2); // text + turn_complete expect(snap.liveJournal).toHaveLength(0); }); + + it('does not persist history_truncated markers through ingest or seed', () => { + const marker: BridgeEvent = { + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: 2, + retainedEvents: 1, + maxBytes: 128, + fullTranscriptAvailable: false, + }, + }; + const engine = new TurnBoundaryCompactionEngine(); + + engine.ingest(marker); + engine.ingest(makeTextChunk(1, 'Hello')); + engine.ingest(makeTurnComplete(2)); + + expect(engine.snapshot().compactedTurns.map((e) => e.type)).toEqual([ + 'session_update', + 'turn_complete', + ]); + + const seeded = new TurnBoundaryCompactionEngine(); + seeded.seed({ + compactedTurns: [ + marker, + makeTextChunk(1, 'Loaded'), + makeTurnComplete(2), + ], + lastEventId: 2, + }); + + expect(seeded.snapshot().compactedTurns.map((e) => e.type)).toEqual([ + 'session_update', + 'turn_complete', + ]); + }); + }); + + describe('bounded replay window', () => { + it('drops oldest completed live turn segments when max replay bytes is exceeded', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 512 }); + + engine.ingest(makeTextChunk(1, `first-${'x'.repeat(600)}`)); + engine.ingest(makeTurnComplete(2)); + engine.ingest(makeTextChunk(3, `second-${'y'.repeat(600)}`)); + engine.ingest(makeTurnComplete(4)); + engine.ingest(makeTextChunk(5, `third-${'z'.repeat(600)}`)); + engine.ingest(makeTurnComplete(6)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns[0]?.type).toBe('history_truncated'); + expect(extractTexts(snap.compactedTurns)).toEqual([ + `third-${'z'.repeat(600)}`, + ]); + expect(snap.compactedTurns.at(-1)?.id).toBe(6); + expect(snap.liveJournal).toHaveLength(0); + + expect(snap.compactedTurns[0]?.data).toMatchObject({ + reason: 'replay_window_exceeded', + truncatedEvents: 4, + truncatedTurns: 2, + retainedEvents: 2, + maxBytes: 512, + fullTranscriptAvailable: false, + }); + }); + + it('retains the newest oversized live turn without a truncation marker', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 128 }); + + engine.ingest(makeTextChunk(1, `oversized-${'x'.repeat(600)}`)); + engine.ingest(makeTurnComplete(2)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns[0]?.type).not.toBe('history_truncated'); + expect(extractTexts(snap.compactedTurns)).toEqual([ + `oversized-${'x'.repeat(600)}`, + ]); + expect(snap.compactedTurns.at(-1)?.id).toBe(2); + }); + + it('notifies the eviction diagnostic hook when replay is dropped', () => { + const onReplayWindowEviction = vi.fn(); + const engine = new TurnBoundaryCompactionEngine({ + maxReplayBytes: 512, + onReplayWindowEviction, + }); + + engine.ingest(makeTextChunk(1, `first-${'x'.repeat(600)}`)); + engine.ingest(makeTurnComplete(2)); + engine.ingest(makeTextChunk(3, `second-${'y'.repeat(600)}`)); + engine.ingest(makeTurnComplete(4)); + + expect(onReplayWindowEviction).toHaveBeenCalledWith( + expect.objectContaining({ + droppedEvents: 2, + droppedSegments: 1, + droppedTurns: 1, + maxBytes: 512, + retainedEvents: 2, + }), + ); + }); + + it('keeps replay working when the eviction diagnostic hook throws', () => { + const engine = new TurnBoundaryCompactionEngine({ + maxReplayBytes: 512, + onReplayWindowEviction: () => { + throw new Error('diagnostic failed'); + }, + }); + + expect(() => { + engine.ingest(makeTextChunk(1, `first-${'x'.repeat(600)}`)); + engine.ingest(makeTurnComplete(2)); + engine.ingest(makeTextChunk(3, `second-${'y'.repeat(600)}`)); + engine.ingest(makeTurnComplete(4)); + }).not.toThrow(); + + const snap = engine.snapshot(); + expect(snap.compactedTurns[0]?.type).toBe('history_truncated'); + expect(extractTexts(snap.compactedTurns)).toEqual([ + `second-${'y'.repeat(600)}`, + ]); + }); }); describe('latest-wins events', () => { @@ -544,6 +672,65 @@ describe('TurnBoundaryCompactionEngine', () => { expect(texts).toEqual(['seeded', 'fresh']); expect(snap.compactedTurns).toHaveLength(4); // seeded text + seeded tc + fresh text + fresh tc }); + + it('applies the replay byte cap to seeded compacted turns', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 512 }); + + engine.seed({ + compactedTurns: [ + makeTextChunk(10, `old-${'x'.repeat(600)}`), + makeTextChunk(11, `new-${'y'.repeat(600)}`), + ], + lastEventId: 11, + }); + + const snap = engine.snapshot(); + expect(snap.lastEventId).toBe(11); + expect(snap.liveJournal).toHaveLength(0); + expect(snap.compactedTurns[0]?.type).toBe('history_truncated'); + expect(extractTexts(snap.compactedTurns)).toEqual([ + `new-${'y'.repeat(600)}`, + ]); + expect(snap.compactedTurns[0]?.data).toMatchObject({ + reason: 'replay_window_exceeded', + truncatedEvents: 1, + retainedEvents: 1, + maxBytes: 512, + fullTranscriptAvailable: false, + }); + }); + + it('evicts seeded replay segments when later live turns exceed the byte cap', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 512 }); + + engine.seed({ + compactedTurns: [makeTextChunk(10, `seed-${'x'.repeat(600)}`)], + lastEventId: 10, + }); + engine.ingest(makeTextChunk(11, `live-${'y'.repeat(600)}`)); + engine.ingest(makeTurnComplete(12)); + + const snap = engine.snapshot(); + expect(snap.lastEventId).toBe(12); + expect(snap.liveJournal).toHaveLength(0); + expect(snap.compactedTurns[0]?.type).toBe('history_truncated'); + expect(extractTexts(snap.compactedTurns)).toEqual([ + `live-${'y'.repeat(600)}`, + ]); + expect(snap.compactedTurns.at(-1)?.id).toBe(12); + expect(snap.compactedTurns[0]?.data).toMatchObject({ + reason: 'replay_window_exceeded', + truncatedEvents: 1, + retainedEvents: 2, + maxBytes: 512, + fullTranscriptAvailable: false, + }); + expect( + (snap.compactedTurns[0]?.data as Record)[ + 'truncatedTurns' + ], + ).toBeUndefined(); + }); }); describe('close', () => { @@ -652,7 +839,7 @@ describe('TurnBoundaryCompactionEngine', () => { }); describe('EventBus + CompactionEngine integration', () => { - it('seedReplayEvents advances replay state without populating the ring', async () => { + it('seedReplayEvents advances replay state without populating the ring or liveJournal', async () => { const engine = new TurnBoundaryCompactionEngine(); const bus = new EventBus(100, undefined, engine); @@ -680,8 +867,9 @@ describe('EventBus + CompactionEngine integration', () => { const snapshot = bus.snapshotReplay()!; expect(snapshot.lastEventId).toBe(2); - expect(snapshot.liveJournal).toHaveLength(2); - expect(snapshot.liveJournal[0]!._meta?.['serverTimestamp']).toBe( + expect(snapshot.compactedTurns).toHaveLength(2); + expect(snapshot.liveJournal).toHaveLength(0); + expect(snapshot.compactedTurns[0]!._meta?.['serverTimestamp']).toBe( 1_700_000_000_000, ); @@ -698,6 +886,163 @@ describe('EventBus + CompactionEngine integration', () => { await iterator.return?.(); }); + it('seedReplayEvents emits a bounded compacted replay window with a truncation marker', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 512 }); + const bus = new EventBus(100, undefined, engine); + + bus.seedReplayEvents([ + { + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text: `old-${'x'.repeat(600)}` }, + }, + }, + }, + { + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: `new-${'y'.repeat(600)}` }, + }, + }, + }, + ]); + + const snapshot = bus.snapshotReplay()!; + expect(snapshot.lastEventId).toBe(2); + expect(snapshot.liveJournal).toHaveLength(0); + expect(snapshot.compactedTurns[0]?.type).toBe('history_truncated'); + expect(extractTexts(snapshot.compactedTurns)).toEqual([ + `new-${'y'.repeat(600)}`, + ]); + expect(snapshot.compactedTurns[0]?.data).toMatchObject({ + reason: 'replay_window_exceeded', + truncatedEvents: 1, + retainedEvents: 1, + maxBytes: 512, + fullTranscriptAvailable: false, + }); + expect( + (snapshot.compactedTurns[0]?.data as Record)[ + 'truncatedTurns' + ], + ).toBeUndefined(); + }); + + it('seedReplayEvents replaces prior replay and truncation state', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 512 }); + const bus = new EventBus(100, undefined, engine); + + bus.seedReplayEvents([ + { + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: `old-${'x'.repeat(600)}` }, + }, + }, + }, + { + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: `drop-${'y'.repeat(600)}` }, + }, + }, + }, + ]); + expect(bus.snapshotReplay()!.compactedTurns[0]?.type).toBe( + 'history_truncated', + ); + + bus.seedReplayEvents([ + { + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'fresh' }, + }, + }, + }, + ]); + + const snapshot = bus.snapshotReplay()!; + expect(snapshot.lastEventId).toBe(3); + expect(snapshot.liveJournal).toHaveLength(0); + expect(snapshot.compactedTurns).toHaveLength(1); + expect(snapshot.compactedTurns[0]?.type).toBe('session_update'); + expect(snapshot.compactedTurns[0]?.id).toBe(3); + expect(extractTexts(snapshot.compactedTurns)).toEqual(['fresh']); + }); + + it('seedReplayEvents treats event sizing failures as zero bytes', () => { + const engine = new TurnBoundaryCompactionEngine({ maxReplayBytes: 1 }); + const bus = new EventBus(100, undefined, engine); + const circular: Record = {}; + circular['self'] = circular; + const stderrSpy = vi.spyOn(process.stderr, 'write').mockReturnValue(true); + + try { + expect(() => + bus.seedReplayEvents([ + { type: 'seeded_misc', data: circular }, + { + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'tail' }, + }, + }, + }, + ]), + ).not.toThrow(); + + const snapshot = bus.snapshotReplay()!; + expect(snapshot.compactedTurns[0]?.type).toBe('history_truncated'); + expect(extractTexts(snapshot.compactedTurns)).toEqual(['tail']); + expect(snapshot.compactedTurns[0]?.data).toMatchObject({ + truncatedEvents: 1, + retainedEvents: 1, + maxBytes: 1, + }); + expect(stderrSpy).toHaveBeenCalledWith( + expect.stringContaining( + 'qwen serve: EventBus event sizing failed {"type":"seeded_misc"}', + ), + ); + } finally { + stderrSpy.mockRestore(); + } + }); + + it('seedReplayEvents keeps its never-throws contract when the compaction seed path fails', () => { + const engine = { + ingest: vi.fn(), + seedReplayEvents: vi.fn(() => { + throw new Error('seed boom'); + }), + snapshot: vi.fn(() => ({ + compactedTurns: [], + liveJournal: [], + lastEventId: 0, + })), + close: vi.fn(), + }; + const bus = new EventBus(100, undefined, engine); + + expect(() => + bus.seedReplayEvents([{ type: 'session_update', data: {} }]), + ).not.toThrow(); + expect(bus.lastEventId).toBe(1); + }); + it('snapshotReplay returns compacted state after publish + turn_complete', () => { const engine = new TurnBoundaryCompactionEngine(); const bus = new EventBus(100, undefined, engine); diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts index 284fd9efc1a..97cdfb24d55 100644 --- a/packages/acp-bridge/src/compactionEngine.ts +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -6,12 +6,19 @@ import { EVENT_SCHEMA_VERSION, + serializedBridgeEventByteLength, type BridgeEvent, type CompactionEngine, type SessionReplaySnapshot, } from './eventBus.js'; +import { normalizeCompactedReplayMaxBytes } from './replayWindowLimits.js'; export type { CompactionEngine, SessionReplaySnapshot }; +export { + DEFAULT_COMPACTED_REPLAY_MAX_BYTES, + MAX_COMPACTED_REPLAY_MAX_BYTES, + normalizeCompactedReplayMaxBytes, +} from './replayWindowLimits.js'; interface SessionUpdateData { update?: { @@ -27,6 +34,7 @@ interface SessionUpdateData { const TURN_BOUNDARY_TYPES = new Set(['turn_complete', 'turn_error']); const TRANSIENT_TYPES = new Set([ + 'history_truncated', 'slow_client_warning', 'client_evicted', 'replay_complete', @@ -36,6 +44,7 @@ const LATEST_WINS_UPDATES = new Set([ 'available_commands_update', 'current_mode_update', ]); +const REPLAY_SEGMENT_COMPACT_THRESHOLD = 64; type CompactedSlot = | { @@ -50,6 +59,27 @@ type CompactedSlot = | { kind: 'misc'; event: BridgeEvent } | { kind: 'latestWins'; key: string; event: BridgeEvent }; +interface ReplaySegment { + events: BridgeEvent[]; + bytes: number; + turnCount: number; +} + +export interface ReplayWindowEviction { + droppedBytes: number; + droppedEvents: number; + droppedSegments: number; + droppedTurns: number; + maxBytes: number; + retainedBytes: number; + retainedEvents: number; +} + +export interface TurnBoundaryCompactionEngineOptions { + maxReplayBytes?: number; + onReplayWindowEviction?: (eviction: ReplayWindowEviction) => void; +} + /** * Compaction engine that merges events at turn boundaries. * @@ -62,15 +92,28 @@ type CompactedSlot = * O(streaming_tokens). Typical compression: 25-30x for chatty sessions. */ export class TurnBoundaryCompactionEngine implements CompactionEngine { - private compactedTurns: BridgeEvent[] = []; + private readonly maxReplayBytes: number; + private readonly onReplayWindowEviction: + | ((eviction: ReplayWindowEviction) => void) + | undefined; + private replaySegments: ReplaySegment[] = []; + private replaySegmentStart = 0; + private replayBytes = 0; private liveJournal: BridgeEvent[] = []; private lastEventId = 0; private closed = false; + private truncatedEvents = 0; + private truncatedTurns = 0; private slots: CompactedSlot[] = []; private toolSlotIndex: Map = new Map(); private textSlotIndex: Map = new Map(); + constructor(opts: TurnBoundaryCompactionEngineOptions = {}) { + this.maxReplayBytes = normalizeCompactedReplayMaxBytes(opts.maxReplayBytes); + this.onReplayWindowEviction = opts.onReplayWindowEviction; + } + ingest(event: BridgeEvent): void { if (this.closed) return; if (event.id !== undefined) { @@ -95,8 +138,14 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { } snapshot(): SessionReplaySnapshot { + const compactedTurns = this.flattenReplaySegments(); + if (this.truncatedEvents > 0) { + compactedTurns.unshift( + this.makeHistoryTruncatedEvent(compactedTurns.length), + ); + } return { - compactedTurns: this.compactedTurns.slice(), + compactedTurns, liveJournal: this.liveJournal.slice(), lastEventId: this.lastEventId, }; @@ -104,8 +153,26 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { seed(snapshot: { compactedTurns: BridgeEvent[]; lastEventId: number }): void { if (this.closed) return; - this.compactedTurns = snapshot.compactedTurns.slice(); + this.resetReplayWindow(); this.lastEventId = snapshot.lastEventId; + for (const event of snapshot.compactedTurns) { + if (TRANSIENT_TYPES.has(event.type)) continue; + this.addReplaySegment([event], 0); + } + this.liveJournal = []; + this.slots = []; + this.toolSlotIndex.clear(); + this.textSlotIndex.clear(); + } + + seedReplayEvents(events: BridgeEvent[]): void { + if (this.closed) return; + this.resetReplayWindow(); + for (const event of events) { + this.recordLastEventId(event); + if (TRANSIENT_TYPES.has(event.type)) continue; + this.addReplaySegment([event], 0); + } this.liveJournal = []; this.slots = []; this.toolSlotIndex.clear(); @@ -115,7 +182,7 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { close(): void { if (this.closed) return; this.closed = true; - this.compactedTurns = []; + this.resetReplayWindow(); this.liveJournal = []; this.slots = []; this.toolSlotIndex.clear(); @@ -290,12 +357,113 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { } compacted.push(boundaryEvent); - this.compactedTurns.push(...compacted); + this.addReplaySegment(compacted, 1); this.liveJournal = []; this.slots = []; this.toolSlotIndex.clear(); this.textSlotIndex.clear(); } + + private recordLastEventId(event: BridgeEvent): void { + if (event.id !== undefined) { + this.lastEventId = event.id; + } + } + + private addReplaySegment(events: BridgeEvent[], turnCount: number): void { + if (events.length === 0) return; + const bytes = events.reduce( + (sum, event) => sum + serializedBridgeEventByteLength(event), + 0, + ); + this.replaySegments.push({ events: events.slice(), bytes, turnCount }); + this.replayBytes += bytes; + this.enforceReplayWindow(); + } + + private enforceReplayWindow(): void { + let droppedSegmentCount = 0; + let droppedBytes = 0; + let droppedEvents = 0; + let droppedTurns = 0; + + while ( + this.replayBytes > this.maxReplayBytes && + this.activeReplaySegmentCount() > 1 + ) { + const dropped = this.replaySegments[this.replaySegmentStart]!; + this.replaySegmentStart += 1; + droppedSegmentCount += 1; + droppedBytes += dropped.bytes; + droppedEvents += dropped.events.length; + droppedTurns += dropped.turnCount; + this.replayBytes -= dropped.bytes; + this.truncatedEvents += dropped.events.length; + this.truncatedTurns += dropped.turnCount; + } + + if (droppedSegmentCount > 0) { + this.compactReplaySegmentQueueIfNeeded(); + this.notifyReplayWindowEviction({ + droppedBytes, + droppedEvents, + droppedSegments: droppedSegmentCount, + droppedTurns, + maxBytes: this.maxReplayBytes, + retainedBytes: this.replayBytes, + retainedEvents: this.flattenReplaySegments().length, + }); + } + } + + private flattenReplaySegments(): BridgeEvent[] { + return this.replaySegments + .slice(this.replaySegmentStart) + .flatMap((segment) => segment.events); + } + + private activeReplaySegmentCount(): number { + return this.replaySegments.length - this.replaySegmentStart; + } + + private compactReplaySegmentQueueIfNeeded(): void { + if (this.replaySegmentStart < REPLAY_SEGMENT_COMPACT_THRESHOLD) return; + this.replaySegments.splice(0, this.replaySegmentStart); + this.replaySegmentStart = 0; + } + + private notifyReplayWindowEviction(eviction: ReplayWindowEviction): void { + try { + this.onReplayWindowEviction?.(eviction); + } catch { + // Best-effort diagnostic; eviction accounting must not break replay. + } + } + + private makeHistoryTruncatedEvent(retainedEvents: number): BridgeEvent { + return { + v: EVENT_SCHEMA_VERSION, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: this.truncatedEvents, + retainedEvents, + maxBytes: this.maxReplayBytes, + ...(this.truncatedTurns > 0 + ? { truncatedTurns: this.truncatedTurns } + : {}), + fullTranscriptAvailable: false, + }, + }; + } + + private resetReplayWindow(): void { + this.replaySegments = []; + this.replaySegmentStart = 0; + this.replayBytes = 0; + this.truncatedEvents = 0; + this.truncatedTurns = 0; + } } function makeMergedSessionUpdateEvent( diff --git a/packages/acp-bridge/src/eventBus.ts b/packages/acp-bridge/src/eventBus.ts index 33dd75eb6ca..e836a3772f1 100644 --- a/packages/acp-bridge/src/eventBus.ts +++ b/packages/acp-bridge/src/eventBus.ts @@ -27,6 +27,7 @@ export interface SessionReplaySnapshot { export interface CompactionEngine { ingest(event: BridgeEvent): void; + seedReplayEvents(events: BridgeEvent[]): void; snapshot(): SessionReplaySnapshot; close(): void; } @@ -132,7 +133,7 @@ function normalizeMaxQueuedBytes(value: number | undefined): number { return value; } -function serializedByteLength(event: BridgeEvent): number { +export function serializedBridgeEventByteLength(event: BridgeEvent): number { try { const serialized = JSON.stringify(event); if (serialized === undefined) return 0; @@ -291,12 +292,12 @@ export class EventBus { }, }; events.push(event); - try { - this.compactionEngine?.ingest(event); - } catch { - // CompactionEngine is best-effort; mirror publish()'s never-throws - // contract for bulk replay seeding. - } + } + try { + this.compactionEngine?.seedReplayEvents(events); + } catch { + // CompactionEngine is best-effort; mirror publish()'s never-throws + // contract for bulk replay seeding. } // Seeded replay frames intentionally do not enter the reconnect ring. A @@ -361,7 +362,7 @@ export class EventBus { if (this.ring.length > this.ringSize) this.ring.shift(); let eventBytes: number | undefined; const getEventBytes = () => { - eventBytes ??= serializedByteLength(event); + eventBytes ??= serializedBridgeEventByteLength(event); return eventBytes; }; // Snapshot the subscribers so an in-loop `this.subs.delete(sub)` diff --git a/packages/acp-bridge/src/index.ts b/packages/acp-bridge/src/index.ts index 051831b8b1a..b926e1b653b 100644 --- a/packages/acp-bridge/src/index.ts +++ b/packages/acp-bridge/src/index.ts @@ -15,6 +15,7 @@ export * from './bridgeErrors.js'; export * from './sessionArtifacts.js'; export * from './bridgeTypes.js'; export * from './bridgeOptions.js'; +export * from './replayWindowLimits.js'; export * from './spawnChannel.js'; export * from './ndJsonStream.js'; export * from './bridgeClient.js'; diff --git a/packages/acp-bridge/src/replayWindowLimits.ts b/packages/acp-bridge/src/replayWindowLimits.ts new file mode 100644 index 00000000000..230e7455f2a --- /dev/null +++ b/packages/acp-bridge/src/replayWindowLimits.ts @@ -0,0 +1,25 @@ +/** + * @license + * Copyright 2026 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +export const DEFAULT_COMPACTED_REPLAY_MAX_BYTES = 4 * 1024 * 1024; +export const MAX_COMPACTED_REPLAY_MAX_BYTES = 256 * 1024 * 1024; + +export function normalizeCompactedReplayMaxBytes( + value: number | undefined, +): number { + if (value === undefined) return DEFAULT_COMPACTED_REPLAY_MAX_BYTES; + if ( + !Number.isSafeInteger(value) || + value < 1 || + value > MAX_COMPACTED_REPLAY_MAX_BYTES + ) { + throw new TypeError( + `Invalid compactedReplayMaxBytes: ${value}. ` + + `Must be a positive safe integer in [1, ${MAX_COMPACTED_REPLAY_MAX_BYTES}].`, + ); + } + return value; +} diff --git a/packages/cli/src/commands/serve.test.ts b/packages/cli/src/commands/serve.test.ts index 3b32e5e3e14..374f8ad2aab 100644 --- a/packages/cli/src/commands/serve.test.ts +++ b/packages/cli/src/commands/serve.test.ts @@ -57,6 +57,13 @@ describe('serve command args', () => { expect(parsed['permission-response-timeout-ms']).toBe(60000); }); + it('parses --compacted-replay-max-bytes as a number', () => { + const parsed = buildParser().parseSync( + '--compacted-replay-max-bytes 4194304', + ); + expect(parsed['compacted-replay-max-bytes']).toBe(4 * 1024 * 1024); + }); + it('parses --max-total-sessions as a number', () => { const parsed = buildParser().parseSync('--max-total-sessions 42'); expect(parsed['max-total-sessions']).toBe(42); @@ -240,6 +247,23 @@ describe('serve rate limit env parsing', () => { ); }); + it('passes compacted replay byte cap to runQwenServe', async () => { + mockRunQwenServe.mockResolvedValueOnce({ + url: 'http://127.0.0.1:4170/', + webShellMounted: false, + }); + + await startServeHandlerWithArgs( + '--no-web --compacted-replay-max-bytes 1048576', + ); + + expect(mockRunQwenServe).toHaveBeenCalledWith( + expect.objectContaining({ + compactedReplayMaxBytes: 1024 * 1024, + }), + ); + }); + it('passes --max-total-sessions to runQwenServe', async () => { mockRunQwenServe.mockResolvedValueOnce({ url: 'http://127.0.0.1:4170/', diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index 9e1efede414..c48a553ff17 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -14,6 +14,7 @@ import { normalizeServeChannelSelection } from '../serve/channel-selection.js'; // handler below so it only loads when the user actually runs `qwen serve`. import { writeStderrLine } from '../utils/stdioHelpers.js'; import { DEFAULT_RING_SIZE } from '@qwen-code/acp-bridge/eventBus'; +import { DEFAULT_COMPACTED_REPLAY_MAX_BYTES } from '@qwen-code/acp-bridge/replayWindowLimits'; import { ApprovalMode, MCP_BUDGET_WARN_FRACTION, @@ -100,6 +101,7 @@ interface ServeArgs { 'max-pending-prompts-per-session': number; 'max-connections': number; 'event-ring-size': number; + 'compacted-replay-max-bytes': number; workspace?: string | string[]; 'require-auth': boolean; 'enable-session-shell': boolean; @@ -266,6 +268,15 @@ export const serveCommand: CommandModule = { 'headroom at the cost of a few hundred KB extra RAM per session. ' + 'Must be a positive finite integer.', }) + .option('compacted-replay-max-bytes', { + type: 'number', + default: DEFAULT_COMPACTED_REPLAY_MAX_BYTES, + description: + 'Per-session in-memory compacted replay snapshot byte cap for ' + + '`POST /session/:id/load` late attaches. Larger = more recent ' + + 'history in load snapshots at higher heap cost. Must be a positive ' + + 'safe integer no larger than 256 MiB.', + }) .option('http-bridge', { type: 'boolean', default: true, @@ -555,6 +566,7 @@ export const serveCommand: CommandModule = { maxPendingPromptsPerSession, maxConnections: argv['max-connections'], eventRingSize: argv['event-ring-size'], + compactedReplayMaxBytes: argv['compacted-replay-max-bytes'], workspace: argv.workspace, requireAuth: argv['require-auth'], enableSessionShell: argv['enable-session-shell'], diff --git a/packages/cli/src/serve/daemon-status.test.ts b/packages/cli/src/serve/daemon-status.test.ts index 06da92908a4..e03233a2008 100644 --- a/packages/cli/src/serve/daemon-status.test.ts +++ b/packages/cli/src/serve/daemon-status.test.ts @@ -28,6 +28,7 @@ const BASE_BRIDGE_SNAPSHOT: BridgeDaemonStatusSnapshot = { maxSessions: 20, maxPendingPromptsPerSession: 5, eventRingSize: 8000, + compactedReplayMaxBytes: 4 * 1024 * 1024, channelIdleTimeoutMs: 0, sessionIdleTimeoutMs: 1_800_000, }, diff --git a/packages/cli/src/serve/daemon-status.ts b/packages/cli/src/serve/daemon-status.ts index 66df50c0f56..253b446aa9e 100644 --- a/packages/cli/src/serve/daemon-status.ts +++ b/packages/cli/src/serve/daemon-status.ts @@ -147,6 +147,7 @@ interface DaemonStatusLimits { maxPendingPromptsPerSession: number | null; listenerMaxConnections: number | null; eventRingSize: number; + compactedReplayMaxBytes: number; promptDeadlineMs: number | null; writerIdleTimeoutMs: number | null; channelIdleTimeoutMs: number; @@ -346,6 +347,7 @@ export async function buildDaemonStatusResponse( bridgeSnapshot.limits.maxPendingPromptsPerSession, listenerMaxConnections: listenerMaxConnections(input.opts.maxConnections), eventRingSize: bridgeSnapshot.limits.eventRingSize, + compactedReplayMaxBytes: bridgeSnapshot.limits.compactedReplayMaxBytes, promptDeadlineMs: positiveFiniteOrNull(input.opts.promptDeadlineMs), writerIdleTimeoutMs: positiveFiniteOrNull(input.opts.writerIdleTimeoutMs), channelIdleTimeoutMs: bridgeSnapshot.limits.channelIdleTimeoutMs, diff --git a/packages/cli/src/serve/fast-path.test.ts b/packages/cli/src/serve/fast-path.test.ts index 9a2207dc0b8..c9f8ab0d466 100644 --- a/packages/cli/src/serve/fast-path.test.ts +++ b/packages/cli/src/serve/fast-path.test.ts @@ -620,6 +620,10 @@ describe('serve fast path argument parsing', () => { ], ['max-connections', ['--max-connections', '256']], ['event-ring-size', ['--event-ring-size', '8000']], + [ + 'compacted-replay-max-bytes', + ['--compacted-replay-max-bytes', '4194304'], + ], ['workspace', ['--workspace', process.cwd()]], ['require-auth', ['--require-auth']], ['enable-session-shell', ['--enable-session-shell']], @@ -682,11 +686,27 @@ describe('serve fast path argument parsing', () => { expect(fastPathParsed).not.toHaveProperty('options.maxTotalSessions'); expect(fastPathParsed).not.toHaveProperty('options.maxConnections'); expect(fastPathParsed).not.toHaveProperty('options.eventRingSize'); + expect(fastPathParsed).not.toHaveProperty( + 'options.compactedReplayMaxBytes', + ); expect(fastPathParsed).not.toHaveProperty( 'options.maxPendingPromptsPerSession', ); }); + it('parses --compacted-replay-max-bytes on the fast path', () => { + const parsed = parseServeFastPathArgs([ + 'serve', + '--compacted-replay-max-bytes', + '1048576', + ]); + + expect(parsed).toMatchObject({ + kind: 'serve', + options: { compactedReplayMaxBytes: 1024 * 1024 }, + }); + }); + it('keeps --experimental-lsp on the fast path', () => { const parsed = parseServeFastPathArgs(['serve', '--experimental-lsp']); @@ -734,6 +754,10 @@ describe('serve fast path argument parsing', () => { ['serve', '--max-pending-prompts-per-session=-1'], 'qwen serve: --max-pending-prompts-per-session must be a non-negative integer (0 / Infinity = unlimited).', ], + [ + ['serve', '--compacted-replay-max-bytes=0'], + 'qwen serve: --compacted-replay-max-bytes must be a positive safe integer in [1, 268435456].', + ], [ ['serve', '--rate-limit', '--rate-limit-prompt=0'], 'qwen serve: --rate-limit-prompt must be a positive integer.', diff --git a/packages/cli/src/serve/fast-path.ts b/packages/cli/src/serve/fast-path.ts index 96e9d79ec0c..cd13dcf0904 100644 --- a/packages/cli/src/serve/fast-path.ts +++ b/packages/cli/src/serve/fast-path.ts @@ -5,6 +5,7 @@ */ import type { RunHandle } from './run-qwen-serve.js'; +import { MAX_COMPACTED_REPLAY_MAX_BYTES } from '@qwen-code/acp-bridge/replayWindowLimits'; import { normalizeServeFastPathArgv } from './fast-path-argv.js'; import type { ServeFastPathSettings } from './fast-path-settings.js'; import { RUNTIME_STARTUP_CANCELLED_MESSAGE } from './runtime-startup-errors.js'; @@ -39,6 +40,7 @@ const NUMBER_OPTIONS = new Map< ['maxPendingPromptsPerSession', 'max-pending-prompts-per-session'], ['maxConnections', 'max-connections'], ['eventRingSize', 'event-ring-size'], + ['compactedReplayMaxBytes', 'compacted-replay-max-bytes'], ['mcp-client-budget', 'mcp-client-budget'], ['promptDeadlineMs', 'prompt-deadline-ms'], ['writerIdleTimeoutMs', 'writer-idle-timeout-ms'], @@ -193,6 +195,19 @@ function getServeFastPathValidationError( return 'qwen serve: --max-pending-prompts-per-session must be a non-negative integer (0 / Infinity = unlimited).'; } + const compactedReplayMaxBytes = parsed.options.compactedReplayMaxBytes; + if ( + compactedReplayMaxBytes !== undefined && + (!Number.isSafeInteger(compactedReplayMaxBytes) || + compactedReplayMaxBytes < 1 || + compactedReplayMaxBytes > MAX_COMPACTED_REPLAY_MAX_BYTES) + ) { + return ( + 'qwen serve: --compacted-replay-max-bytes must be a positive ' + + `safe integer in [1, ${MAX_COMPACTED_REPLAY_MAX_BYTES}].` + ); + } + return null; } diff --git a/packages/cli/src/serve/run-qwen-serve.test.ts b/packages/cli/src/serve/run-qwen-serve.test.ts index 85c9658f5d3..60ca942d623 100644 --- a/packages/cli/src/serve/run-qwen-serve.test.ts +++ b/packages/cli/src/serve/run-qwen-serve.test.ts @@ -48,6 +48,7 @@ const BASE_BRIDGE_SNAPSHOT: BridgeDaemonStatusSnapshot = { maxSessions: 20, maxPendingPromptsPerSession: 5, eventRingSize: 8000, + compactedReplayMaxBytes: 4 * 1024 * 1024, channelIdleTimeoutMs: 0, sessionIdleTimeoutMs: 1_800_000, }, @@ -972,6 +973,13 @@ describe('runQwenServe pre-listen bridge option validation', () => { ['eventRingSize', 0, /eventRingSize/], ['eventRingSize', 1.5, /eventRingSize/], ['eventRingSize', Number.POSITIVE_INFINITY, /eventRingSize/], + ['compactedReplayMaxBytes', 0, /compactedReplayMaxBytes/], + ['compactedReplayMaxBytes', 1.5, /compactedReplayMaxBytes/], + [ + 'compactedReplayMaxBytes', + Number.POSITIVE_INFINITY, + /compactedReplayMaxBytes/, + ], ] as const)( 'rejects invalid %s=%s before printing the listening line', async (optionName, value, message) => { @@ -2985,6 +2993,7 @@ describe('runQwenServe runtime startup failures', () => { maxPendingPromptsPerSession: 5, listenerMaxConnections: 256, eventRingSize: 8000, + compactedReplayMaxBytes: 4 * 1024 * 1024, promptDeadlineMs: null, writerIdleTimeoutMs: null, channelIdleTimeoutMs: 0, diff --git a/packages/cli/src/serve/run-qwen-serve.ts b/packages/cli/src/serve/run-qwen-serve.ts index 22f0a87e043..5901655a7ac 100644 --- a/packages/cli/src/serve/run-qwen-serve.ts +++ b/packages/cli/src/serve/run-qwen-serve.ts @@ -19,6 +19,10 @@ import express, { type Response, } from 'express'; import { writeStderrLine, writeStdoutLine } from '../utils/stdioHelpers.js'; +import { + DEFAULT_COMPACTED_REPLAY_MAX_BYTES, + normalizeCompactedReplayMaxBytes, +} from '@qwen-code/acp-bridge/replayWindowLimits'; import type { BridgeEvent } from '@qwen-code/acp-bridge/eventBus'; import type { NdJsonMessageObservation } from '@qwen-code/acp-bridge/ndJsonStream'; import { getDeviceFlowRegistry } from './auth/device-flow.js'; @@ -1153,6 +1157,8 @@ function createBootstrapServeApp(input: { ), listenerMaxConnections: listenerMaxConnections(opts.maxConnections), eventRingSize: opts.eventRingSize ?? DEFAULT_EVENT_RING_SIZE, + compactedReplayMaxBytes: + opts.compactedReplayMaxBytes ?? DEFAULT_COMPACTED_REPLAY_MAX_BYTES, promptDeadlineMs: positiveFiniteOrNull(opts.promptDeadlineMs), writerIdleTimeoutMs: positiveFiniteOrNull(opts.writerIdleTimeoutMs), channelIdleTimeoutMs: channelIdleTimeoutMs(opts.channelIdleTimeoutMs), @@ -1810,6 +1816,9 @@ export async function runQwenServe( ); } } + if (opts.compactedReplayMaxBytes !== undefined) { + normalizeCompactedReplayMaxBytes(opts.compactedReplayMaxBytes); + } if (opts.writerIdleTimeoutMs !== undefined) { if (!isPositiveIntegerMs(opts.writerIdleTimeoutMs)) { throw new TypeError( @@ -2405,6 +2414,9 @@ export async function runQwenServe( ...(opts.eventRingSize !== undefined ? { eventRingSize: opts.eventRingSize } : {}), + ...(opts.compactedReplayMaxBytes !== undefined + ? { compactedReplayMaxBytes: opts.compactedReplayMaxBytes } + : {}), ...(opts.channelIdleTimeoutMs !== undefined ? { channelIdleTimeoutMs: opts.channelIdleTimeoutMs } : {}), diff --git a/packages/cli/src/serve/server.test.ts b/packages/cli/src/serve/server.test.ts index 4dfb3e23cd2..f33e765ee73 100644 --- a/packages/cli/src/serve/server.test.ts +++ b/packages/cli/src/serve/server.test.ts @@ -1276,6 +1276,7 @@ function fakeBridge(opts: FakeBridgeOpts = {}): FakeBridge { maxSessions: 20, maxPendingPromptsPerSession: 5, eventRingSize: 8000, + compactedReplayMaxBytes: 4 * 1024 * 1024, channelIdleTimeoutMs: 0, sessionIdleTimeoutMs: 1_800_000, }, @@ -11956,6 +11957,7 @@ describe('createServeApp', () => { maxSessions: 20, maxPendingPromptsPerSession: 5, eventRingSize: 8000, + compactedReplayMaxBytes: 4 * 1024 * 1024, channelIdleTimeoutMs: 0, sessionIdleTimeoutMs: 1_800_000, }, diff --git a/packages/cli/src/serve/server.ts b/packages/cli/src/serve/server.ts index e26a8e63e0e..ac9d197493d 100644 --- a/packages/cli/src/serve/server.ts +++ b/packages/cli/src/serve/server.ts @@ -528,6 +528,7 @@ export function createServeApp( : {}), maxPendingPromptsPerSession: opts.maxPendingPromptsPerSession, eventRingSize: opts.eventRingSize, + compactedReplayMaxBytes: opts.compactedReplayMaxBytes, permissionResponseTimeoutMs: opts.permissionResponseTimeoutMs, boundWorkspace, sessionShellCommandEnabled, diff --git a/packages/cli/src/serve/types.ts b/packages/cli/src/serve/types.ts index af17435bf44..eb40b9528d0 100644 --- a/packages/cli/src/serve/types.ts +++ b/packages/cli/src/serve/types.ts @@ -92,6 +92,12 @@ export interface ServeOptions { * at the cost of a few hundred KB extra RAM per session. */ eventRingSize?: number; + /** + * Per-session in-memory compacted replay snapshot byte cap. Threaded into + * `BridgeOptions.compactedReplayMaxBytes`. Defaults to 4 MiB. Must be a + * positive safe integer; there is no unlimited sentinel. + */ + compactedReplayMaxBytes?: number; /** * Absolute workspace path this daemon binds to. The daemon is * **1 daemon = 1 workspace × N sessions**: one bound diff --git a/packages/cli/vitest.config.ts b/packages/cli/vitest.config.ts index fe51cd9b18c..59ea254ebbc 100644 --- a/packages/cli/vitest.config.ts +++ b/packages/cli/vitest.config.ts @@ -69,6 +69,10 @@ export default defineConfig({ __dirname, '../acp-bridge/src/eventBus.ts', ), + '@qwen-code/acp-bridge/replayWindowLimits': path.resolve( + __dirname, + '../acp-bridge/src/replayWindowLimits.ts', + ), '@qwen-code/acp-bridge/workspacePaths': path.resolve( __dirname, '../acp-bridge/src/workspacePaths.ts', diff --git a/packages/sdk-typescript/scripts/build.js b/packages/sdk-typescript/scripts/build.js index 91a2dc967ff..9c4ac20d956 100755 --- a/packages/sdk-typescript/scripts/build.js +++ b/packages/sdk-typescript/scripts/build.js @@ -42,7 +42,9 @@ const rootDir = join(__dirname, '..'); // Bumped from 133KB to 136KB after merging session artifacts plus sessionless // workspace memory forget/dream APIs and event validation. // Bumped from 136KB to 137KB for EventBus byte-backlog telemetry validation. -const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 137 * 1024; +// Bumped from 137KB to 138KB for history_truncated event validation and +// transcript status projection. +const MAX_DAEMON_BROWSER_BUNDLE_BYTES = 138 * 1024; // The opt-in `daemon/transports` browser bundle legitimately ships the concrete // ACP transports (AcpHttpTransport/AcpWsTransport/AutoReconnect + negotiate), so // it's larger than the default barrel — but still budgeted so a future PR can't diff --git a/packages/sdk-typescript/src/daemon/events.ts b/packages/sdk-typescript/src/daemon/events.ts index 250ba0a48e8..98c33e17f7b 100644 --- a/packages/sdk-typescript/src/daemon/events.ts +++ b/packages/sdk-typescript/src/daemon/events.ts @@ -60,6 +60,11 @@ export const DAEMON_KNOWN_EVENT_TYPE_VALUES = [ // reseeds state. Synthetic (no `id`) so it doesn't burn a slot // in the per-session monotonic sequence. 'state_resync_required', + // Synthetic marker prepended to a bounded `/session/:id/load` replay + // snapshot when older replay history was dropped from the daemon's + // in-memory window. This is NOT a resync request: consumers should render + // it as transcript status and continue applying the retained snapshot. + 'history_truncated', // MCP guardrail push events. See `mcp_guardrail_events` capability // tag. Both fire on the per-session SSE bus; consumers should // pre-flight `caps.features.includes('mcp_guardrail_events')` @@ -398,6 +403,16 @@ export interface DaemonStateResyncRequiredData { [key: string]: unknown; } +export interface DaemonHistoryTruncatedData { + reason: 'replay_window_exceeded'; + truncatedEvents: number; + retainedEvents: number; + maxBytes: number; + truncatedTurns?: number; + fullTranscriptAvailable: false; + [key: string]: unknown; +} + /** * Payload for the `mcp_budget_warning` SSE frame. Fired on the upward * 75% crossing of `reservedSlots.size / clientBudget`. Re-arms only @@ -952,6 +967,10 @@ export type DaemonStateResyncRequiredEvent = DaemonEventEnvelope< 'state_resync_required', DaemonStateResyncRequiredData >; +export type DaemonHistoryTruncatedEvent = DaemonEventEnvelope< + 'history_truncated', + DaemonHistoryTruncatedData +>; export type DaemonMcpBudgetWarningEvent = DaemonEventEnvelope< 'mcp_budget_warning', DaemonMcpBudgetWarningData @@ -1103,7 +1122,8 @@ export type DaemonStreamLifecycleEvent = | DaemonClientEvictedEvent | DaemonSlowClientWarningEvent | DaemonStreamErrorEvent - | DaemonStateResyncRequiredEvent; + | DaemonStateResyncRequiredEvent + | DaemonHistoryTruncatedEvent; /** * MCP guardrail push events. Grouped as their own union member (rather @@ -1320,6 +1340,14 @@ export interface DaemonSessionViewState { resyncRequiredCount: number; /** Most recent resync payload (reason + gap range). */ lastResyncRequired?: DaemonStateResyncRequiredData; + /** + * Count of `history_truncated` markers observed from bounded replay + * snapshots. This is informational only and does not imply stale local state + * or trigger resync recovery. + */ + historyTruncatedCount: number; + /** Most recent bounded replay-window marker. */ + lastHistoryTruncated?: DaemonHistoryTruncatedData; /** * Daemon assist push: most recent `followup_suggestion` observed on * this session. Adapters render it as ghost-text in the input @@ -1370,6 +1398,7 @@ const MAX_FORBIDDEN_VOTES_PER_SESSION = 32; */ const RESYNC_PASSTHROUGH_TYPES = new Set([ 'state_resync_required', + 'history_truncated', 'session_died', 'session_closed', 'client_evicted', @@ -1432,6 +1461,8 @@ export function createDaemonSessionViewState( awaitingResync: seed.awaitingResync ?? false, resyncRequiredCount: seed.resyncRequiredCount ?? 0, lastResyncRequired: seed.lastResyncRequired, + historyTruncatedCount: seed.historyTruncatedCount ?? 0, + lastHistoryTruncated: seed.lastHistoryTruncated, lastFollowupSuggestion: seed.lastFollowupSuggestion, rewindCount: seed.rewindCount ?? 0, lastRewind: seed.lastRewind, @@ -1563,6 +1594,10 @@ export function asKnownDaemonEvent( return isStateResyncRequiredData(event.data) ? (event as DaemonStateResyncRequiredEvent) : undefined; + case 'history_truncated': + return isHistoryTruncatedData(event.data) + ? (event as DaemonHistoryTruncatedEvent) + : undefined; case 'mcp_budget_warning': return isMcpBudgetWarningData(event.data) ? (event as DaemonMcpBudgetWarningEvent) @@ -1928,6 +1963,12 @@ export function reduceDaemonSessionEvent( resyncRequiredCount: base.resyncRequiredCount + 1, lastResyncRequired: event.data, }; + case 'history_truncated': + return { + ...base, + historyTruncatedCount: base.historyTruncatedCount + 1, + lastHistoryTruncated: event.data, + }; case 'mcp_budget_warning': // Non-terminal: budget pressure is a status signal, not a stream // close. Count + capture latest so adapters can render @@ -2571,6 +2612,28 @@ function isStateResyncRequiredData( ); } +function isHistoryTruncatedData( + value: unknown, +): value is DaemonHistoryTruncatedData { + if ( + !isRecord(value) || + value['reason'] !== 'replay_window_exceeded' || + !isFiniteNumber(value['truncatedEvents']) || + !isFiniteNumber(value['retainedEvents']) || + !isFiniteNumber(value['maxBytes']) || + value['fullTranscriptAvailable'] !== false + ) { + return false; + } + const truncatedTurns = value['truncatedTurns']; + return ( + isNonNegativeInteger(value['truncatedEvents']) && + isNonNegativeInteger(value['retainedEvents']) && + isNonNegativeInteger(value['maxBytes']) && + (truncatedTurns === undefined || isNonNegativeInteger(truncatedTurns)) + ); +} + function isSlowClientWarningData( value: unknown, ): value is DaemonSlowClientWarningData { @@ -3038,6 +3101,10 @@ function isOptionalNumberOrNull(value: unknown): boolean { return value === undefined || value === null || isFiniteNumber(value); } +function isNonNegativeInteger(value: unknown): boolean { + return isFiniteNumber(value) && Number.isInteger(value) && value >= 0; +} + function isOptionalStringOrNull(value: unknown): boolean { return value === undefined || value === null || typeof value === 'string'; } diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index 76e2bfc7d16..7e98acbf852 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -197,6 +197,8 @@ export type { DaemonApprovalModeChangedEvent, DaemonClientEvictedData, DaemonClientEvictedEvent, + DaemonHistoryTruncatedData, + DaemonHistoryTruncatedEvent, DaemonControlEvent, // Daemon-emitted resync // signal for SSE reconnects past the ring eviction boundary. diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index fb4255f651c..3ab4b2c0f7d 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -293,6 +293,7 @@ export interface DaemonStatusReport { channelIdleTimeoutMs: number; sessionIdleTimeoutMs: number; acpConnectionCap: number | null; + compactedReplayMaxBytes: number; }; capabilities: { protocolVersions: DaemonProtocolVersions; diff --git a/packages/sdk-typescript/src/daemon/ui/normalizer.ts b/packages/sdk-typescript/src/daemon/ui/normalizer.ts index 37d697ea60e..682eb78047c 100644 --- a/packages/sdk-typescript/src/daemon/ui/normalizer.ts +++ b/packages/sdk-typescript/src/daemon/ui/normalizer.ts @@ -179,6 +179,9 @@ export function normalizeDaemonEvent( case 'state_resync_required': return normalizeStateResyncRequired(event, base); + case 'history_truncated': + return normalizeHistoryTruncated(event, base); + case 'session_rewound': return normalizeSessionRewound(event, base); @@ -360,6 +363,33 @@ function normalizeStateResyncRequired( ]; } +function normalizeHistoryTruncated( + event: DaemonEvent, + base: NormalizedEventBase, +): DaemonUiEvent[] { + const reason = getString(event.data, 'reason'); + const truncatedEvents = numberField(event.data, 'truncatedEvents'); + const retainedEvents = numberField(event.data, 'retainedEvents'); + const maxBytes = numberField(event.data, 'maxBytes'); + if ( + reason !== 'replay_window_exceeded' || + truncatedEvents === undefined || + retainedEvents === undefined || + maxBytes === undefined || + (isRecord(event.data) && event.data['fullTranscriptAvailable'] !== false) + ) { + return fallbackDebug(event, base, 'malformed history_truncated payload'); + } + return [ + { + ...base, + type: 'status', + text: `History truncated: retained ${retainedEvents}, dropped ${truncatedEvents} (window ${maxBytes} bytes).`, + source: 'history_truncated', + }, + ]; +} + function normalizeSessionRewound( event: DaemonEvent, base: NormalizedEventBase, diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index abcba9395f7..1433cd43740 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -70,6 +70,8 @@ export { type DaemonErrorKind, type DaemonClientEvictedData, type DaemonClientEvictedEvent, + type DaemonHistoryTruncatedData, + type DaemonHistoryTruncatedEvent, type DaemonPendingPromptAddedData, type DaemonPendingPromptAddedEvent, type DaemonPendingPromptStartedData, diff --git a/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts b/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts index e7345c6abe4..6e547e42e01 100644 --- a/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts +++ b/packages/sdk-typescript/test/unit/daemon-public-surface.test.ts @@ -30,6 +30,8 @@ import type { DaemonGithubSetupRequest, DaemonGithubSetupResult, DaemonGithubSetupWorkflowResult, + DaemonHistoryTruncatedData, + DaemonHistoryTruncatedEvent, DaemonKnownEventType, DaemonModelSwitchedData, DaemonModelSwitchedEvent, @@ -171,6 +173,7 @@ describe('public SDK entry — typed daemon event surface (#4217)', () => { expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); + expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); @@ -185,6 +188,7 @@ describe('public SDK entry — typed daemon event surface (#4217)', () => { expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); + expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); @@ -236,6 +240,9 @@ describe('public SDK entry — typed daemon event surface (#4217)', () => { // `GET /daemon/status` report surface (PR 5174 client coverage): the // envelope plus the sub-shapes UI dashboards need to type against. expectTypeOf().not.toBeNever(); + expectTypeOf().toMatchTypeOf<{ + compactedReplayMaxBytes: number; + }>(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); expectTypeOf().not.toBeNever(); diff --git a/packages/sdk-typescript/test/unit/daemonEvents.test.ts b/packages/sdk-typescript/test/unit/daemonEvents.test.ts index db44962d5cb..bf9e07b9bde 100644 --- a/packages/sdk-typescript/test/unit/daemonEvents.test.ts +++ b/packages/sdk-typescript/test/unit/daemonEvents.test.ts @@ -2650,6 +2650,100 @@ describe('PR 21 — auth device-flow events', () => { }); describe('state_resync_required (#4175 F4 prereq, Ilya0527 issue #15)', () => { + it('recognizes history_truncated and records it without entering resync', () => { + const event = { + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: 4, + retainedEvents: 2, + maxBytes: 512, + truncatedTurns: 2, + fullTranscriptAvailable: false, + }, + } satisfies DaemonEvent; + + const known = asKnownDaemonEvent(event); + expect(known?.type).toBe('history_truncated'); + + const state = reduceDaemonSessionEvent( + createDaemonSessionViewState(), + event, + ); + expect(state.awaitingResync).toBe(false); + expect(state.historyTruncatedCount).toBe(1); + expect(state.lastHistoryTruncated).toEqual(event.data); + }); + + it('rejects malformed history_truncated payloads', () => { + expect( + asKnownDaemonEvent({ + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + retainedEvents: 2, + maxBytes: 512, + fullTranscriptAvailable: false, + }, + }), + ).toBeUndefined(); + expect( + asKnownDaemonEvent({ + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: -1, + retainedEvents: 2, + maxBytes: 512, + fullTranscriptAvailable: false, + }, + }), + ).toBeUndefined(); + expect( + asKnownDaemonEvent({ + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: 4, + retainedEvents: 2.5, + maxBytes: 512, + fullTranscriptAvailable: false, + }, + }), + ).toBeUndefined(); + expect( + asKnownDaemonEvent({ + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: 4, + retainedEvents: 2, + maxBytes: 512, + truncatedTurns: -1, + fullTranscriptAvailable: false, + }, + }), + ).toBeUndefined(); + expect( + asKnownDaemonEvent({ + v: 1, + type: 'history_truncated', + data: { + reason: 'wrong_reason', + truncatedEvents: 4, + retainedEvents: 2, + maxBytes: 512, + fullTranscriptAvailable: false, + }, + }), + ).toBeUndefined(); + }); + it('sets awaitingResync + records the resync data when daemon emits state_resync_required', () => { const state = reduceDaemonSessionEvent(createDaemonSessionViewState(), { v: 1, diff --git a/packages/sdk-typescript/test/unit/daemonUi.test.ts b/packages/sdk-typescript/test/unit/daemonUi.test.ts index b45cca008cc..e8d14154011 100644 --- a/packages/sdk-typescript/test/unit/daemonUi.test.ts +++ b/packages/sdk-typescript/test/unit/daemonUi.test.ts @@ -3265,6 +3265,68 @@ describe('daemon UI reducer state machine (PR-E)', () => { expect(JSON.stringify(state.blocks)).not.toContain('stale delta'); }); + it('projects history truncation as status without entering resync', () => { + const events = normalizeDaemonEvent({ + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: 4, + retainedEvents: 2, + maxBytes: 512, + truncatedTurns: 2, + fullTranscriptAvailable: false, + }, + } as never); + + expect(events).toEqual([ + expect.objectContaining({ + type: 'status', + source: 'history_truncated', + text: expect.stringContaining('History truncated') as string, + }), + ]); + + const state = reduceDaemonTranscriptEvents( + createDaemonTranscriptState({ now: 1 }), + events, + { now: 2 }, + ); + + expect(state.awaitingResync).toBe(false); + expect(state.resyncRequiredCount).toBe(0); + expect(state.blocks).toMatchObject([ + { + kind: 'status', + text: expect.stringContaining('History truncated') as string, + }, + ]); + expect(daemonUiEventToTerminalText(events[0])).toContain( + 'History truncated', + ); + }); + + it('routes malformed history truncation payloads to debug', () => { + const events = normalizeDaemonEvent({ + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: '4', + retainedEvents: 2, + maxBytes: 512, + fullTranscriptAvailable: false, + }, + } as never); + + expect(events).toEqual([ + expect.objectContaining({ + type: 'debug', + text: 'history_truncated: malformed history_truncated payload', + }), + ]); + }); + it('mirrors approval mode from session.approval_mode.changed event', async () => { const { selectApprovalMode } = await import('../../src/daemon/ui/index.js'); let state = createDaemonTranscriptState({ now: 1 }); diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx index 1c27b801240..7187590ea0c 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.test.tsx @@ -2065,7 +2065,9 @@ describe('DaemonSessionProvider', () => { }); it('logs settings reloads without inserting daemon debug blocks', async () => { - const debug = vi.spyOn(console, 'debug').mockImplementation(() => undefined); + const debug = vi + .spyOn(console, 'debug') + .mockImplementation(() => undefined); const session = createMockSession({ events: async function* settingsReloadEvents() { yield { @@ -2879,6 +2881,70 @@ describe('DaemonSessionProvider', () => { ]); }); + it('renders bounded replay truncation from the loaded snapshot without resync', async () => { + const session = createMockSession({ + replaySnapshot: { + compactedReplay: [ + { + v: 1, + type: 'history_truncated', + data: { + reason: 'replay_window_exceeded', + truncatedEvents: 4, + retainedEvents: 2, + maxBytes: 512, + fullTranscriptAvailable: false, + }, + }, + { + id: 5, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'retained replay' }, + }, + }, + }, + ], + liveJournal: [], + }, + }); + sdkMocks.sessions.push(session); + let blocks: readonly DaemonTranscriptBlock[] = []; + let awaitingResync = false; + + function Harness() { + blocks = useDaemonTranscriptBlocks(); + awaitingResync = useDaemonTranscriptState().awaitingResync; + return null; + } + + await renderWithProvider(, { + autoConnect: true, + reconnectDelayMs: 1, + maxReconnectDelayMs: 1, + }); + await act(async () => { + await flushPromises(); + }); + + expect(awaitingResync).toBe(false); + expect(blocks).toEqual( + expect.arrayContaining([ + expect.objectContaining({ + kind: 'status', + text: expect.stringContaining('History truncated'), + }), + expect.objectContaining({ + kind: 'assistant', + text: 'retained replay', + }), + ]), + ); + }); + it('keeps replayed non-turn events from marking a prompt as waiting', async () => { const session = createMockSession({ replaySnapshot: { diff --git a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx index a4230dc7f44..bf99824312c 100644 --- a/packages/webui/src/daemon/session/DaemonSessionProvider.tsx +++ b/packages/webui/src/daemon/session/DaemonSessionProvider.tsx @@ -379,12 +379,12 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { // store.dispatch() appends to existing blocks. No reset, no // load(), minimal re-render. // - // PATH B — Full reload (session cleared, terminal/auth errors, + // PATH B — Snapshot reload (session cleared, terminal/auth errors, // ring eviction): // `session` is null → enter this block → DaemonSessionClient // .load() fetches compactedReplay + liveJournal → deferred // store.reset() + store.dispatch(replayEvents) rebuilds the - // full transcript in a single synchronous batch. + // current bounded replay window in a single synchronous batch. // // The `needsStoreReset` flag defers store.reset() to avoid an // intermediate empty-blocks state that causes virtualizer @@ -1142,7 +1142,8 @@ export function DaemonSessionProvider(props: DaemonSessionProviderProps) { // Ring eviction means the SSE replay window has a real gap. // Resetting and continuing on the same stream can only replay // the surviving tail; reload the session snapshot instead so - // compactedReplay/liveJournal rebuild the full transcript. + // compactedReplay/liveJournal rebuild the bounded replay + // window. console.warn( '[DaemonSessionProvider] ring eviction detected, reloading session (sessionId=%s)', activeSession.sessionId,