diff --git a/docs/design/live-journal-truncation-recovery.md b/docs/design/live-journal-truncation-recovery.md index 8bf36837cf5..63470f1302d 100644 --- a/docs/design/live-journal-truncation-recovery.md +++ b/docs/design/live-journal-truncation-recovery.md @@ -2,7 +2,7 @@ ## Context -The daemon keeps a bounded in-memory live journal for an unfinished turn. When the journal exceeds 10,000 events or 8 MiB, it discards the oldest replay events and prepends a `history_truncated` marker. The persisted transcript and turn-boundary compaction remain authoritative, so the complete turn becomes available again after a formal terminal event. +The daemon keeps a bounded in-memory live journal for an unfinished turn. Consecutive compatible text or thought chunks share bounded replay entries, with at most 256 source events per entry. When the journal exceeds 10,000 replay entries or 8 MiB of serialized source events, it discards the oldest entries and prepends a `history_truncated` marker whose retained and truncated counts still describe source events. The persisted transcript and turn-boundary compaction remain authoritative, so the complete turn becomes available again after a formal terminal event. The marker previously had no prompt ownership, the SDK rendered a generic message, and WebUI either hid the marker behind history pagination or left the retained tail permanently visible. This design keeps the existing resource limits and eviction policy while making the loss precise and repairing the visible tail without another model request. diff --git a/docs/developers/daemon/03-acp-bridge.md b/docs/developers/daemon/03-acp-bridge.md index 7914bc08427..5915e1ba290 100644 --- a/docs/developers/daemon/03-acp-bridge.md +++ b/docs/developers/daemon/03-acp-bridge.md @@ -250,11 +250,14 @@ In addition to the core `spawnOrAttach`, `sendPrompt`, `cancelSession`, `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). The in-flight `liveJournal` is -separately capped by `BridgeOptions.maxJournalEvents` (default 10 000) and -`BridgeOptions.maxJournalBytes` (default 8 MiB). If older retained replay was -dropped, `compactedReplay[0]` is the id-less `history_truncated` marker; if -journal entries were dropped, `liveJournal[0]` carries a `history_truncated` -marker with `scope: 'live_journal'`. The full persisted transcript remains on +separately capped by `BridgeOptions.maxJournalEvents` (default 10 000 replay +entries) and `BridgeOptions.maxJournalBytes` (default 8 MiB of serialized +source events). Consecutive compatible text or thought chunks share a replay +entry, with at most 256 source events per entry; other event and attribution +boundaries remain intact. If older retained replay was dropped, +`compactedReplay[0]` is the id-less `history_truncated` marker; if journal +entries were dropped, `liveJournal[0]` carries a `history_truncated` marker +with `scope: 'live_journal'`. Its retained and truncated counts describe source events, not replay entries. 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`. diff --git a/docs/developers/qwen-serve-protocol.md b/docs/developers/qwen-serve-protocol.md index 589ffe12ec9..5430bcb48b3 100644 --- a/docs/developers/qwen-serve-protocol.md +++ b/docs/developers/qwen-serve-protocol.md @@ -2082,7 +2082,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 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. The in-flight `liveJournal` is separately capped by `--max-journal-events` (default 10 000) and `--max-journal-bytes` (default 8 MiB); when exceeded, the oldest journal entries are dropped and a `history_truncated` marker with `scope: 'live_journal'` is prepended. Clients should render that marker as status and continue applying retained events. Full persisted transcript access is exposed separately through `GET /session/:id/transcript`. +**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. The in-flight `liveJournal` is separately capped by `--max-journal-events` (default 10 000 replay entries) and `--max-journal-bytes` (default 8 MiB of serialized source events). Consecutive compatible `agent_message_chunk` or `agent_thought_chunk` source events share a replay entry, up to 256 source events per entry, while tool, attribution, provenance, and discrete-message boundaries remain intact. When either cap is exceeded, the oldest entries are dropped whole (so the retained tail can be much smaller than the byte cap) and a `history_truncated` marker with `scope: 'live_journal'` is prepended; its `truncatedEvents` and `retainedEvents` fields count source events, not replay entries. Clients should render that marker as status and continue applying retained events. Full persisted transcript access is exposed separately through `GET /session/:id/transcript`. The replay-window byte caps apply after the child has reconstructed the persisted transcript; they do not cap the on-disk JSONL read. A restore that exceeds the daemon budget returns `504` with a `Retry-After` derived from the restore budget (clamped to 5-120s) and `{code: "session_restore_timeout", errorKind: "restore_timeout", retryable: true, sessionId, action, timeoutMs}`. The daemon fences the still-running ACP request and cleans up any late session instead of registering it. A retry for the same id returns `409 restore_in_progress` with `reason: "awaiting_abandoned_cleanup"` and a `Retry-After` of the restore budget (clamped to 5-120s) until that cleanup settles. If late cleanup is uncertain, or the abandoned restore has still not settled a full restore budget after its deadline, new sessions on that workspace return `503 acp_channel_unavailable` with `reason: "restore_cleanup_failed"` or `"restore_settlement_overdue"`; already-live sessions remain usable while the channel drains. diff --git a/docs/users/qwen-serve.md b/docs/users/qwen-serve.md index 32726428d99..2e3fcf60d23 100644 --- a/docs/users/qwen-serve.md +++ b/docs/users/qwen-serve.md @@ -401,8 +401,8 @@ Notes: | `--child-heap-mode ` | `observe` | Whether the daemon models a per-child heap partition of `--memory-budget-mb`. `observe` (default) reports what it would apply — `limits.memory.childHeap.perChildCeilingMb` and `maxConcurrentChildren` — and counts spawns that would have exceeded the limit. **Nothing is applied**: no child is sized from the budget and no spawn is refused. `off` models nothing, and says so on the wire: `maxConcurrentChildren` and `perChildCeilingMb` are both `null` rather than carrying a partition you switched off. A refusal count of 0 does **not** mean the partition would be safe to apply: children still run on the much larger host-derived ceiling, so a workload needing more old space than the modeled ceiling looks perfectly healthy here. Applying the partition ships with the measurement that can answer that. | | `--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` is separately capped by `--max-journal-events` and `--max-journal-bytes`. 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. | -| `--max-journal-events ` | `10000` | Per-session cap on the number of raw events retained in the in-flight live journal (the current unfinished turn). When exceeded, the oldest journal entries are dropped and a `history_truncated` marker is prepended. Must be a positive safe integer. | -| `--max-journal-bytes ` | `8388608` | Per-session byte cap on the in-flight live journal. When exceeded, the oldest journal entries are dropped (at least one entry is always kept). Must be a positive safe integer. Defaults to 8 MiB. | +| `--max-journal-events ` | `10000` | Per-session cap on replay entries retained in the in-flight `liveJournal` for the current unfinished turn. Consecutive compatible text or thought chunks share an entry, with at most 256 source events per entry; other event boundaries are preserved. When exceeded, the oldest entries are dropped and a `history_truncated` marker is prepended. The marker's `truncatedEvents` and `retainedEvents` counts describe source events. Must be a positive safe integer. | +| `--max-journal-bytes ` | `8388608` | Per-session byte cap on the in-flight `liveJournal`, accounted from the serialized source events even when compatible chunks share a replay entry. When exceeded, the oldest entries are dropped whole (at least one entry is always kept), so the retained tail can be much smaller than the cap. Must be a positive safe integer. Defaults to 8 MiB. | | `--mcp-client-budget ` | — | Positive integer cap on live MCP clients. When `mcp_workspace_pool` is advertised, the cap and transports are shared per workspace runtime; when the tag is absent, the legacy per-session manager enforces it. 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 rather than total live clients. Pre-flight `caps.features.mcp_guardrails` and `caps.features.mcp_workspace_pool`. | | `--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. | | `--external-tool-guard-mode ` | `off` | Managed ACP external pre-execution policy. `off` makes no provider calls and advertises no capability. `required` fails startup unless a compatible provider completes the v1 handshake, then fails every supported top-level tool invocation closed unless its single prepare request is allowed. | diff --git a/integration-tests/cli/qwen-serve-live-journal-recovery.test.ts b/integration-tests/cli/qwen-serve-live-journal-recovery.test.ts index fb00fc0f5ae..f15b03464e4 100644 --- a/integration-tests/cli/qwen-serve-live-journal-recovery.test.ts +++ b/integration-tests/cli/qwen-serve-live-journal-recovery.test.ts @@ -35,17 +35,65 @@ function updateKind(event: DaemonEvent): unknown { } describe('qwen serve live journal recovery', () => { + it('keeps a compatible live chunk stream within the replay entry cap', async () => { + const workspace = makeTempWorkspace('live-journal-aggregation'); + try { + activeDaemon = await spawnDaemon({ + workspaceCwd: workspace, + extraArgs: ['--max-journal-events', '3'], + env: { + QWEN_CLI_ENTRY: MOCK_AGENT_PATH, + MOCK_ACP_MODE: 'echo', + MOCK_ACP_EMIT_CHUNKS: '20', + MOCK_ACP_PROMPT_DELAY_MS: '1500', + }, + }); + const created = await activeDaemon.client.createOrAttachSession({ + sessionScope: 'thread', + }); + const prompt = activeDaemon.client.prompt(created.sessionId, { + prompt: [{ type: 'text', text: 'aggregate this live turn' }], + }); + + let duringTurn: DaemonSessionClient | undefined; + const deadline = Date.now() + 5_000; + while (Date.now() < deadline) { + const loaded = await DaemonSessionClient.load( + activeDaemon.client, + created.sessionId, + {}, + 'live-journal-observer', + ); + const replayText = JSON.stringify(loaded.replaySnapshot.liveJournal); + if (replayText.includes('chunk-19')) { + duringTurn = loaded; + break; + } + await new Promise((resolve) => setTimeout(resolve, 25)); + } + + expect(duringTurn).toBeDefined(); + expect(duringTurn!.replaySnapshot.liveJournal).not.toContainEqual( + expect.objectContaining({ type: 'history_truncated' }), + ); + const replayText = JSON.stringify(duringTurn!.replaySnapshot.liveJournal); + expect(replayText).toContain('chunk-0'); + expect(replayText).toContain('chunk-19'); + + await prompt; + } finally { + await activeDaemon?.dispose(); + activeDaemon = undefined; + fs.rmSync(workspace, { recursive: true, force: true }); + } + }, 30_000); + it('attributes a truncated live tail and exposes the complete turn after terminal', async () => { const workspace = makeTempWorkspace('live-journal-recovery'); try { activeDaemon = await spawnDaemon({ workspaceCwd: workspace, - extraArgs: [ - '--max-journal-events', - '3', - '--max-journal-bytes', - String(8 * 1024 * 1024), - ], + extraArgs: ['--max-journal-events', '3', '--max-journal-bytes', '300'], env: { QWEN_CLI_ENTRY: MOCK_AGENT_PATH, MOCK_ACP_MODE: 'echo', @@ -89,19 +137,19 @@ describe('qwen serve live journal recovery', () => { data: { scope: 'live_journal', maxEvents: 3, + truncatedEvents: expect.any(Number), fullTranscriptAvailable: true, }, }); - expect(duringTurn!.replaySnapshot.liveJournal).not.toContainEqual( - expect.objectContaining({ - type: 'session_update', - data: expect.objectContaining({ - update: expect.objectContaining({ - content: expect.objectContaining({ text: 'chunk-0' }), - }), - }), - }), - ); + expect( + (marker?.data as { truncatedEvents?: number } | undefined) + ?.truncatedEvents, + ).toBeGreaterThan(0); + // Merged entries concatenate up to 256 source chunks, so an exact + // per-entry match can never hold; assert on the serialized tail. + expect( + JSON.stringify(duringTurn!.replaySnapshot.liveJournal), + ).not.toContain('chunk-0'); await prompt; const repaired = await DaemonSessionClient.load( diff --git a/integration-tests/cli/qwen-serve-webui-live-journal-recovery.test.ts b/integration-tests/cli/qwen-serve-webui-live-journal-recovery.test.ts index 1e59d75d855..eaacdd2c3cd 100644 --- a/integration-tests/cli/qwen-serve-webui-live-journal-recovery.test.ts +++ b/integration-tests/cli/qwen-serve-webui-live-journal-recovery.test.ts @@ -121,7 +121,7 @@ describe('qwen serve WebUI live journal recovery', () => { }; activeDaemon = await spawnDaemon({ workspaceCwd: workspace, - extraArgs: ['--max-journal-events', '3'], + extraArgs: ['--max-journal-events', '3', '--max-journal-bytes', '300'], env: { QWEN_CLI_ENTRY: MOCK_AGENT_PATH, MOCK_ACP_MODE: 'echo', diff --git a/packages/acp-bridge/src/bridgeOptions.ts b/packages/acp-bridge/src/bridgeOptions.ts index 14b6f8f2390..af4235f6859 100644 --- a/packages/acp-bridge/src/bridgeOptions.ts +++ b/packages/acp-bridge/src/bridgeOptions.ts @@ -246,16 +246,19 @@ export interface BridgeOptions { */ compactedReplayMaxBytes?: number; /** - * Per-session cap on the number of raw events retained in the in-flight - * live journal (the current unfinished turn). When exceeded, the oldest - * journal entries are dropped. Defaults to 10 000. Must be a positive - * safe integer. + * Per-session cap on replay entries retained in the in-flight live journal + * (the current unfinished turn). Consecutive compatible text/thought chunks + * share bounded entries. When exceeded, the oldest journal entries are + * dropped. Defaults to 10 000. Must be a positive safe integer. */ maxJournalEvents?: number; /** - * Per-session byte cap on the in-flight live journal. When exceeded, the - * oldest journal entries are dropped (at least one entry is always kept). - * Defaults to 8 MiB. Must be a positive safe integer. + * Per-session source-event byte cap on the in-flight live journal (the + * current unfinished turn) — accounted from serialized source events even + * when compatible chunks share a replay entry. When exceeded, the oldest + * journal entries are dropped whole (at least one entry is always kept), + * so the retained tail can be much smaller than the cap. Defaults to + * 8 MiB. Must be a positive safe integer. */ maxJournalBytes?: number; /** diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 21c79d5fcbb..95cc18f0b93 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -359,7 +359,7 @@ export interface BridgeRestoredSession extends BridgeSession { replayError?: string; /** Compacted events for all completed turns (O(turns) size). */ compactedReplay?: BridgeEvent[]; - /** Raw events since last turn boundary (current incomplete turn). */ + /** Bounded replay events for the current incomplete turn. */ liveJournal?: BridgeEvent[]; /** True when persisted records exist before the returned replay page. */ historyHasMore?: boolean; diff --git a/packages/acp-bridge/src/compactionEngine.test.ts b/packages/acp-bridge/src/compactionEngine.test.ts index 91c93c9ad0e..dd8ab271f8c 100644 --- a/packages/acp-bridge/src/compactionEngine.test.ts +++ b/packages/acp-bridge/src/compactionEngine.test.ts @@ -175,6 +175,7 @@ function makeTextChunkWithParent( const event = makeTextChunk(id, text); (event.data as { update: Record }).update['_meta'] = { parentToolCallId, + subagentType: 'general-purpose', }; return event; } @@ -187,6 +188,7 @@ function makeThoughtChunkWithParent( const event = makeThoughtChunk(id, text); (event.data as { update: Record }).update['_meta'] = { parentToolCallId, + subagentType: 'general-purpose', }; return event; } @@ -201,6 +203,52 @@ function extractTexts(events: BridgeEvent[]): string[] { .filter((t) => t !== ''); } +type ChunkIdentity = { + parentToolCallId?: string; + subagentType?: string; + sourceRecordIds?: string[]; + promptId?: string; + originatorClientId?: string; + sessionId?: string; +}; + +function withIdentity( + event: BridgeEvent, + identity: ChunkIdentity, +): BridgeEvent { + const update = (event.data as { update: Record }).update; + if ( + identity.parentToolCallId !== undefined || + identity.subagentType !== undefined || + identity.sourceRecordIds !== undefined + ) { + update['_meta'] = { + ...(identity.parentToolCallId === undefined + ? {} + : { parentToolCallId: identity.parentToolCallId }), + ...(identity.subagentType === undefined + ? {} + : { subagentType: identity.subagentType }), + ...(identity.sourceRecordIds === undefined + ? {} + : { + qwenTranscript: { + sourceRecordIds: identity.sourceRecordIds, + }, + }), + }; + } + event.promptId = identity.promptId; + event.originatorClientId = identity.originatorClientId; + if (identity.sessionId !== undefined) { + event.data = { + sessionId: identity.sessionId, + ...(event.data as Record), + }; + } + return event; +} + describe('TurnBoundaryCompactionEngine', () => { describe('basic compaction', () => { it('merges consecutive text chunks into a single event on turn_complete', () => { @@ -278,6 +326,59 @@ describe('TurnBoundaryCompactionEngine', () => { expect(guardEvents.map((event) => event.id)).toEqual([2, 3, 4]); }); + it('keeps generic discrete message and thought chunks separate at turn boundaries', () => { + const makeDiscrete = ( + makeChunk: (id: number, text: string) => BridgeEvent, + id: number, + text: string, + taskId: string, + ): BridgeEvent => { + const event = makeChunk(id, text); + (event.data as { update: Record }).update['_meta'] = { + qwenDiscreteMessage: true, + backgroundTask: { taskId }, + }; + return event; + }; + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'Before')); + engine.ingest(makeDiscrete(makeTextChunk, 2, 'notify-a', 'task-a')); + engine.ingest(makeDiscrete(makeTextChunk, 3, 'notify-b', 'task-b')); + engine.ingest(makeDiscrete(makeThoughtChunk, 4, 'thought-a', 'task-a')); + engine.ingest(makeDiscrete(makeThoughtChunk, 5, 'thought-b', 'task-b')); + engine.ingest(makeThoughtChunk(6, 'ordinary')); + engine.ingest(makeTextChunk(7, 'After')); + engine.ingest(makeTurnComplete(8)); + + const events = engine.snapshot().compactedTurns; + expect(extractTexts(events)).toEqual([ + 'Before', + 'notify-a', + 'notify-b', + 'thought-a', + 'thought-b', + 'ordinary', + 'After', + ]); + const discrete = events.filter((event) => { + const data = event.data as { + update?: { _meta?: { qwenDiscreteMessage?: boolean } }; + }; + return data.update?._meta?.qwenDiscreteMessage === true; + }); + expect(discrete.map((event) => event.id)).toEqual([2, 3, 4, 5]); + expect( + discrete.map( + (event) => + ( + event.data as { + update: { _meta: { backgroundTask: { taskId: string } } }; + } + ).update._meta.backgroundTask.taskId, + ), + ).toEqual(['task-a', 'task-b', 'task-a', 'task-b']); + }); + it('keeps user messages as-is', () => { const engine = new TurnBoundaryCompactionEngine(); engine.ingest(makeUserMessage(1, 'How are you?')); @@ -590,15 +691,425 @@ describe('TurnBoundaryCompactionEngine', () => { }); describe('liveJournal (incomplete turn)', () => { - it('accumulates raw events in liveJournal before turn completes', () => { + it('merges consecutive text chunks for live replay', () => { const engine = new TurnBoundaryCompactionEngine(); - engine.ingest(makeTextChunk(1, 'H')); - engine.ingest(makeTextChunk(2, 'i')); + engine.ingest(makeThoughtChunk(1, 'Let me ')); + engine.ingest(makeThoughtChunk(2, 'think')); + engine.ingest(makeTextChunk(3, 'The ')); + engine.ingest(makeTextChunk(4, 'answer')); const snap = engine.snapshot(); expect(snap.compactedTurns).toHaveLength(0); expect(snap.liveJournal).toHaveLength(2); - expect(snap.lastEventId).toBe(2); + expect(extractTexts(snap.liveJournal)).toEqual([ + 'Let me think', + 'The answer', + ]); + expect(snap.liveJournal.map((event) => event.id)).toEqual([2, 4]); + expect(snap.lastEventId).toBe(4); + }); + + it('preserves tool boundaries in live replay', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'Before')); + engine.ingest(makeToolCall(2, 'tc1', 'running')); + engine.ingest(makeTextChunk(3, 'After')); + + const snap = engine.snapshot(); + expect(snap.liveJournal).toHaveLength(3); + expect(extractTexts(snap.liveJournal)).toEqual(['Before', 'After']); + expect(snap.liveJournal.map((event) => event.id)).toEqual([1, 2, 3]); + }); + + it.each([ + { + name: 'parentToolCallId', + first: { parentToolCallId: 'tool-a', subagentType: 'explore' }, + second: { parentToolCallId: 'tool-b', subagentType: 'explore' }, + }, + { + name: 'sourceRecordIds', + first: { sourceRecordIds: ['record-a'] }, + second: { sourceRecordIds: ['record-b'] }, + }, + { + name: 'promptId', + first: { promptId: 'prompt-a' }, + second: { promptId: 'prompt-b' }, + }, + { + name: 'originatorClientId', + first: { originatorClientId: 'client-a' }, + second: { originatorClientId: 'client-b' }, + }, + { + name: 'sessionId', + first: { sessionId: 'session-a' }, + second: { sessionId: 'session-b' }, + }, + ])('does not merge across $name boundaries', ({ first, second }) => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(withIdentity(makeTextChunk(1, 'first'), first)); + engine.ingest(withIdentity(makeTextChunk(2, 'second'), second)); + + const snap = engine.snapshot(); + expect(snap.liveJournal).toHaveLength(2); + expect(extractTexts(snap.liveJournal)).toEqual(['first', 'second']); + expect(snap.liveJournal.map((event) => event.id)).toEqual([1, 2]); + expect(snap.liveJournal[0]).toMatchObject( + withIdentity(makeTextChunk(1, 'first'), first), + ); + expect(snap.liveJournal[1]).toMatchObject( + withIdentity(makeTextChunk(2, 'second'), second), + ); + }); + + it.each([ + { + name: 'parentToolCallId', + identity: { parentToolCallId: 'tool-a', subagentType: 'explore' }, + }, + { + name: 'sourceRecordIds', + identity: { sourceRecordIds: ['record-a'] }, + }, + { name: 'promptId', identity: { promptId: 'prompt-a' } }, + { + name: 'originatorClientId', + identity: { originatorClientId: 'client-a' }, + }, + { name: 'sessionId', identity: { sessionId: 'session-a' } }, + ])('merges consecutive chunks sharing a defined $name', ({ identity }) => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(withIdentity(makeTextChunk(1, 'first'), identity)); + engine.ingest(withIdentity(makeTextChunk(2, 'second'), identity)); + + const snap = engine.snapshot(); + expect(snap.liveJournal).toHaveLength(1); + expect(extractTexts(snap.liveJournal)).toEqual(['firstsecond']); + expect(snap.liveJournal.map((event) => event.id)).toEqual([2]); + }); + + it.each([ + [ + 'ordinary then guard', + makeTextChunk(1, 'ordinary'), + makeDiscreteTextChunk(2, 'guard', 1), + ], + [ + 'guard then ordinary', + makeDiscreteTextChunk(1, 'guard', 1), + makeTextChunk(2, 'ordinary'), + ], + ])('keeps todo-stop-guard text discrete: %s', (_name, first, second) => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(first); + engine.ingest(second); + + const snap = engine.snapshot(); + expect(snap.liveJournal).toHaveLength(2); + expect(snap.liveJournal.map((event) => event.id)).toEqual([1, 2]); + expect( + snap.liveJournal.map( + (event) => + (event.data as { update: { content: { text: string } } }).update + .content.text, + ), + ).toEqual( + [first, second].map( + (event) => + (event.data as { update: { content: { text: string } } }).update + .content.text, + ), + ); + const guard = snap.liveJournal.find( + (event) => + (event.data as { update: { _meta?: { source?: string } } }).update + ._meta?.source === 'todo_stop_guard', + ); + expect( + (guard?.data as { update: { _meta: Record } }).update + ._meta, + ).toMatchObject({ + source: 'todo_stop_guard', + qwenDiscreteMessage: true, + attempt: 1, + maxAttempts: 2, + }); + }); + + it.each([ + ['message', makeTextChunk], + ['thought', makeThoughtChunk], + ] as const)( + 'keeps generic discrete %s chunks separate in live replay', + (_name, makeChunk) => { + const makeBackgroundMessage = ( + id: number, + text: string, + taskId: string, + ): BridgeEvent => { + const event = makeChunk(id, text); + (event.data as { update: Record }).update['_meta'] = + { + qwenDiscreteMessage: true, + backgroundTask: { taskId }, + }; + return event; + }; + const first = makeBackgroundMessage(1, 'first', 'task-a'); + const second = makeBackgroundMessage(2, 'second', 'task-b'); + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(first); + engine.ingest(second); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(2); + expect(extractTexts(live)).toEqual(['first', 'second']); + expect( + live.map( + (event) => + ( + event.data as { + update: { _meta: { backgroundTask: { taskId: string } } }; + } + ).update._meta.backgroundTask.taskId, + ), + ).toEqual(['task-a', 'task-b']); + }, + ); + + it('preserves semantic envelope metadata event boundaries', () => { + const withEnvelopeMeta = ( + event: BridgeEvent, + meta: Record, + ): BridgeEvent => ({ ...event, _meta: meta }); + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest( + withEnvelopeMeta(makeTextChunk(1, 'before'), { serverTimestamp: 1 }), + ); + engine.ingest( + withEnvelopeMeta(makeTextChunk(2, 'middle'), { + serverTimestamp: 2, + semantic: { kind: 'middle' }, + }), + ); + engine.ingest( + withEnvelopeMeta(makeTextChunk(3, 'after'), { serverTimestamp: 3 }), + ); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(3); + expect(live.map((event) => event.id)).toEqual([1, 2, 3]); + expect(live[1]?._meta).toEqual({ + serverTimestamp: 2, + semantic: { kind: 'middle' }, + }); + }); + + it('preserves semantic metadata event boundaries', () => { + const withMeta = ( + event: BridgeEvent, + meta: Record, + ): BridgeEvent => { + (event.data as { update: Record }).update['_meta'] = + meta; + return event; + }; + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'before')); + engine.ingest( + withMeta(makeTextChunk(2, ''), { usage: { totalTokens: 3 } }), + ); + engine.ingest( + withMeta(makeTextChunk(3, 'command'), { source: 'slash_command' }), + ); + engine.ingest(makeTextChunk(4, 'after')); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(4); + expect(live.map((event) => event.id)).toEqual([1, 2, 3, 4]); + expect( + live.map( + (event) => + (event.data as { update: { content: { text: string } } }).update + .content.text, + ), + ).toEqual(['before', '', 'command', 'after']); + expect( + (live[1]!.data as { update: { _meta: Record } }).update + ._meta, + ).toEqual({ usage: { totalTokens: 3 } }); + expect( + (live[2]!.data as { update: { _meta: Record } }).update + ._meta, + ).toEqual({ source: 'slash_command' }); + }); + + it.each([ + [ + 'an unmodeled update key', + (event: BridgeEvent) => { + (event.data as { update: Record }).update[ + 'annotations' + ] = []; + return event; + }, + ], + [ + 'an unmodeled data key', + (event: BridgeEvent) => { + (event.data as Record)['attachments'] = []; + return event; + }, + ], + ] as const)( + 'keeps text chunks carrying %s out of merged live entries', + (_name, decorate) => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'first')); + engine.ingest(decorate(makeTextChunk(2, 'second'))); + engine.ingest(makeTextChunk(3, 'third')); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(3); + expect(extractTexts(live)).toEqual(['first', 'second', 'third']); + expect(live.map((event) => event.id)).toEqual([1, 2, 3]); + }, + ); + + it('keeps ACP TextContent annotations and _meta out of merged live entries', () => { + // ACP TextContent permits `annotations` and `_meta` beside + // `type`/`text`; the merged-entry rebuild models only `{ type, + // text }`, so such chunks must replay exactly as SSE delivered them. + const withContentFields = (event: BridgeEvent): BridgeEvent => { + const content = ( + event.data as { update: { content: Record } } + ).update.content; + content['annotations'] = { audience: ['assistant'] }; + content['_meta'] = { vendor: 'keep' }; + return event; + }; + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'first')); + engine.ingest(withContentFields(makeTextChunk(2, 'hello'))); + engine.ingest(makeTextChunk(3, 'third')); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(3); + expect(extractTexts(live)).toEqual(['first', 'hello', 'third']); + expect(live.map((event) => event.id)).toEqual([1, 2, 3]); + expect( + (live[1]!.data as { update: { content: unknown } }).update.content, + ).toEqual({ + type: 'text', + text: 'hello', + annotations: { audience: ['assistant'] }, + _meta: { vendor: 'keep' }, + }); + }); + + it('merges live chunks whose empty-string parentToolCallId the extractor ignores', () => { + const withEmptyParent = (event: BridgeEvent): BridgeEvent => { + (event.data as { update: Record }).update['_meta'] = { + parentToolCallId: '', + }; + return event; + }; + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(withEmptyParent(makeTextChunk(1, 'first'))); + engine.ingest(withEmptyParent(makeTextChunk(2, 'second'))); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(1); + expect(extractTexts(live)).toEqual(['firstsecond']); + + engine.ingest(makeTurnComplete(3)); + expect(extractTexts(engine.snapshot().compactedTurns)).toEqual([ + 'firstsecond', + ]); + }); + + it('merges live subagent chunks carrying the producer-stamped meta pair', () => { + // SubAgentTracker stamps streamed subagent fragments with both keys. + const withSubagentMeta = (event: BridgeEvent): BridgeEvent => { + (event.data as { update: Record }).update['_meta'] = { + parentToolCallId: 'tool-a', + subagentType: 'explore', + }; + return event; + }; + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(withSubagentMeta(makeTextChunk(1, 'first'))); + engine.ingest(withSubagentMeta(makeTextChunk(2, 'second'))); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(1); + expect(extractTexts(live)).toEqual(['firstsecond']); + expect( + (live[0]!.data as { update: { _meta?: Record } }) + .update._meta, + ).toEqual({ parentToolCallId: 'tool-a', subagentType: 'explore' }); + + engine.ingest(makeTurnComplete(3)); + const compacted = engine.snapshot().compactedTurns; + expect(extractTexts(compacted)).toEqual(['firstsecond']); + expect( + (compacted[0]!.data as { update: { _meta?: Record } }) + .update._meta, + ).toEqual({ parentToolCallId: 'tool-a', subagentType: 'explore' }); + }); + + it('merges chunks carrying buildUpdateMeta timestamp and plan shapes', () => { + const withTranscriptMeta = ( + event: BridgeEvent, + meta: Record, + ): BridgeEvent => { + (event.data as { update: Record }).update['_meta'] = + meta; + return event; + }; + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest( + withTranscriptMeta(makeTextChunk(1, 'first'), { + timestamp: 1700000000000, + qwenTranscript: { + sourceRecordIds: ['record-a'], + planToolCallId: 'plan-1', + }, + }), + ); + engine.ingest( + withTranscriptMeta(makeTextChunk(2, 'second'), { + timestamp: 1700000000001, + serverTimestamp: 1700000000002, + qwenTranscript: { + sourceRecordIds: ['record-a'], + planToolCallId: 'plan-2', + }, + }), + ); + + const live = engine.snapshot().liveJournal; + expect(live).toHaveLength(1); + expect(extractTexts(live)).toEqual(['firstsecond']); + expect(live.map((event) => event.id)).toEqual([2]); + }); + + it('does not let snapshot frequency change journal eviction', () => { + const engine = new TurnBoundaryCompactionEngine({ + maxJournalEvents: 1, + }); + engine.ingest(makeTextChunk(1, 'first')); + expect(engine.snapshot().liveJournal).toHaveLength(1); + engine.ingest(makeTextChunk(2, ' second')); + + const snap = engine.snapshot(); + expect(snap.liveJournal).toHaveLength(1); + expect(extractTexts(snap.liveJournal)).toEqual(['first second']); + expect( + snap.liveJournal.find((event) => event.type === 'history_truncated'), + ).toBeUndefined(); }); it('clears liveJournal on turn completion', () => { @@ -618,16 +1129,34 @@ describe('TurnBoundaryCompactionEngine', () => { const markerOf = (snap: { liveJournal: BridgeEvent[] }) => snap.liveJournal.find((e) => e.type === 'history_truncated'); - it('drops the oldest journal entries past maxJournalEvents and prepends a marker', () => { + it('keeps a long compatible text stream below the event cap', () => { const engine = new TurnBoundaryCompactionEngine({ maxJournalEvents: 3, }); - for (let i = 1; i <= 5; i++) { + for (let i = 1; i <= 512; i++) { engine.ingest(makeTextChunk(i, `chunk-${i}`)); } const snap = engine.snapshot(); - // marker + the 3 newest raw events; the 2 oldest were dropped. + expect(markerOf(snap)).toBeUndefined(); + expect(snap.liveJournal).toHaveLength(2); + expect(extractTexts(snap.liveJournal).join('')).toBe( + Array.from({ length: 512 }, (_, index) => `chunk-${index + 1}`).join( + '', + ), + ); + expect(snap.liveJournal.map((event) => event.id)).toEqual([256, 512]); + }); + + it('drops the oldest non-mergeable entries past maxJournalEvents and prepends a marker', () => { + const engine = new TurnBoundaryCompactionEngine({ + maxJournalEvents: 3, + }); + for (let i = 1; i <= 5; i++) { + engine.ingest(makeUserMessage(i, `message-${i}`)); + } + + const snap = engine.snapshot(); expect(snap.liveJournal).toHaveLength(4); const marker = markerOf(snap); expect(marker?.data).toEqual({ @@ -643,12 +1172,12 @@ describe('TurnBoundaryCompactionEngine', () => { expect(snap.liveJournal.slice(1).map((e) => e.id)).toEqual([3, 4, 5]); }); - it('drops the oldest journal entries past maxJournalBytes but keeps at least one', () => { + it('drops the oldest non-mergeable entries past maxJournalBytes but keeps at least one', () => { const engine = new TurnBoundaryCompactionEngine({ maxJournalBytes: 300, }); - engine.ingest(makeTextChunk(1, 'x'.repeat(200))); - engine.ingest(makeTextChunk(2, 'y'.repeat(200))); + engine.ingest(makeUserMessage(1, 'x'.repeat(200))); + engine.ingest(makeUserMessage(2, 'y'.repeat(200))); const snap = engine.snapshot(); const marker = markerOf(snap); @@ -656,26 +1185,55 @@ describe('TurnBoundaryCompactionEngine', () => { expect( (marker?.data as { truncatedEvents: number }).truncatedEvents, ).toBe(1); - // The newest (still oversized alone) entry survives — first-item rule. expect(snap.liveJournal.filter((e) => e.id !== undefined)).toHaveLength( 1, ); expect(snap.liveJournal.at(-1)?.id).toBe(2); }); + it('starts a new segment before a merged entry exceeds maxJournalBytes', () => { + const engine = new TurnBoundaryCompactionEngine({ + maxJournalBytes: 600, + }); + engine.ingest(makeTextChunk(1, 'x'.repeat(400))); + engine.ingest(makeTextChunk(2, 'y'.repeat(400))); + + const snap = engine.snapshot(); + expect(markerOf(snap)?.data).toMatchObject({ + truncatedEvents: 1, + retainedEvents: 1, + }); + expect(extractTexts(snap.liveJournal)).toEqual(['y'.repeat(400)]); + expect(snap.liveJournal.at(-1)?.id).toBe(2); + }); + + it('reports raw event counts when an aggregated segment is dropped', () => { + const engine = new TurnBoundaryCompactionEngine({ + maxJournalEvents: 2, + }); + for (let i = 1; i <= 512; i++) { + engine.ingest(makeTextChunk(i, 'x')); + } + engine.ingest(makeUserMessage(513, 'later')); + + const marker = markerOf(engine.snapshot()); + expect(marker?.data).toMatchObject({ + truncatedEvents: 256, + retainedEvents: 257, + }); + }); + it('does not let journal truncation corrupt the compacted turn', () => { const engine = new TurnBoundaryCompactionEngine({ maxJournalEvents: 1, }); engine.ingest(makeTextChunk(1, 'Hello')); engine.ingest(makeTextChunk(2, ' world')); - engine.ingest(makeTurnComplete(3)); + engine.ingest(makeUserMessage(3, 'later')); + engine.ingest(makeTurnComplete(4)); const snap = engine.snapshot(); - // Compaction folds from the slots working set, not the journal: - // the merged text is complete even though the journal was capped. expect(extractTexts(snap.compactedTurns)).toContain('Hello world'); - // Turn boundary reset the journal AND the truncation counter. expect(snap.liveJournal).toHaveLength(0); expect(markerOf(snap)).toBeUndefined(); }); @@ -687,7 +1245,8 @@ describe('TurnBoundaryCompactionEngine', () => { const snap = engine.snapshot(); expect(markerOf(snap)).toBeUndefined(); - expect(snap.liveJournal).toHaveLength(2); + expect(snap.liveJournal).toHaveLength(1); + expect(extractTexts(snap.liveJournal)).toEqual(['Hi']); }); it('marker carries the last-seen recordId as a pagination anchor when the journal overflows', () => { @@ -712,10 +1271,10 @@ describe('TurnBoundaryCompactionEngine', () => { }, }; engine.ingest(turnBounded); - engine.ingest(makeTextChunk(2, 'a')); - engine.ingest(makeTextChunk(3, 'b')); - engine.ingest(makeTextChunk(4, 'c')); - engine.ingest(makeTextChunk(5, 'd')); + engine.ingest(makeUserMessage(2, 'a')); + engine.ingest(makeUserMessage(3, 'b')); + engine.ingest(makeUserMessage(4, 'c')); + engine.ingest(makeUserMessage(5, 'd')); const snap = engine.snapshot(); const marker = markerOf(snap); @@ -751,8 +1310,8 @@ describe('TurnBoundaryCompactionEngine', () => { }, }; engine.ingest(bounded); - engine.ingest(makeTextChunk(3, 'x')); - engine.ingest(makeTextChunk(4, 'y')); + engine.ingest(makeUserMessage(3, 'x')); + engine.ingest(makeUserMessage(4, 'y')); const snap = engine.snapshot(); const marker = markerOf(snap); @@ -1586,7 +2145,7 @@ describe('EventBus + CompactionEngine integration', () => { expect(bus.snapshotReplay()).toBeUndefined(); }); - it('liveJournal contains raw events for incomplete turn', () => { + it('liveJournal contains bounded replay events for incomplete turn', () => { const engine = new TurnBoundaryCompactionEngine(); const bus = new EventBus(100, undefined, engine); @@ -1611,7 +2170,9 @@ describe('EventBus + CompactionEngine integration', () => { const snapshot = bus.snapshotReplay()!; expect(snapshot.compactedTurns).toHaveLength(0); - expect(snapshot.liveJournal).toHaveLength(2); + expect(snapshot.liveJournal).toHaveLength(1); + expect(extractTexts(snapshot.liveJournal)).toEqual(['streaming...']); + expect(snapshot.liveJournal[0]?.id).toBe(2); expect(snapshot.lastEventId).toBe(2); }); diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts index b4eb1bc1979..857ca001c6e 100644 --- a/packages/acp-bridge/src/compactionEngine.ts +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -54,6 +54,7 @@ const LATEST_WINS_UPDATES = new Set([ 'current_mode_update', ]); const REPLAY_SEGMENT_COMPACT_THRESHOLD = 64; +const LIVE_JOURNAL_TEXT_CHUNKS_PER_EVENT = 256; type CompactedSlot = | { @@ -83,6 +84,18 @@ interface ReplaySegment { turnCount: number; } +interface LiveJournalTextSegment { + sessionUpdate: 'agent_message_chunk' | 'agent_thought_chunk'; + chunks: string[]; + sourceRecordIds?: readonly string[]; + parentToolCallId?: string; + promptId?: string; + originatorClientId?: string; + sessionId?: string; + firstEvent: BridgeEvent; + lastEvent: BridgeEvent; +} + function replayRecordId(event: BridgeEvent): string | undefined { if (event.type !== 'session_update') return undefined; const data = event.data; @@ -122,14 +135,15 @@ export interface TurnBoundaryCompactionEngineOptions { maxReplayBytes?: number; onReplayWindowEviction?: (eviction: ReplayWindowEviction) => void; /** - * Caps on the in-flight live journal (DAEMON-009). The journal holds the - * RAW events of the current unfinished turn and is only reset at turn - * boundaries, so a single long-running streaming turn grew it — and the - * cost of every `snapshot()` — without bound. When either cap is hit the - * oldest journal entries are dropped and `snapshot()` prepends a - * `history_truncated` marker (`reason: 'replay_window_exceeded'`, - * `scope: 'live_journal'`) to the live journal. Turn compaction is - * unaffected: it folds from the `slots` working set, not the journal. + * Caps on the in-flight live journal (DAEMON-009). Compatible consecutive + * text/thought chunks are grouped into bounded replay events; other events + * retain their original boundaries. When either cap is hit the oldest + * journal entries are dropped whole (merged segments included), so the + * retained tail can be much smaller than the byte cap, and `snapshot()` + * prepends a `history_truncated` marker + * (`reason: 'replay_window_exceeded'`, `scope: 'live_journal'`). Turn + * compaction is unaffected: it folds from the `slots` working set, not + * the journal. */ maxJournalEvents?: number; maxJournalBytes?: number; @@ -156,11 +170,15 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { private replaySegments: ReplaySegment[] = []; private replaySegmentStart = 0; private replayBytes = 0; - private liveJournal: BridgeEvent[] = []; - /** Serialized size of each `liveJournal` entry, index-parallel. */ + private liveJournal: Array = []; + /** Serialized source-event size of each `liveJournal` entry, index-parallel. */ private journalEntryBytes: number[] = []; + /** Raw event count represented by each `liveJournal` entry. */ + private journalEntryEvents: number[] = []; private journalTotalBytes = 0; + private journalTotalEvents = 0; private journalTruncatedEvents = 0; + private liveJournalTextSegment: LiveJournalTextSegment | undefined; private lastEventId = 0; private closed = false; private truncatedEvents = 0; @@ -220,27 +238,7 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { this.activeRecordId = seenRecordId; } - this.liveJournal.push(event); - // The bus passes the byte size it already computed at publish time; - // self-compute only for callers that don't (defensive — events in the - // journal passed the publish serializability gate, so `?? 0` is - // unreachable in practice). - const bytes = byteLength ?? serializedBridgeEventByteLength(event) ?? 0; - this.journalEntryBytes.push(bytes); - this.journalTotalBytes += bytes; - // Journal cap (DAEMON-009): drop the OLDEST entries once either limit - // is exceeded. The byte cap keeps at least one entry (first-item rule, - // matching the queue byte cap) so a single oversized event doesn't - // wedge the journal empty forever. - while ( - this.liveJournal.length > this.maxJournalEvents || - (this.journalTotalBytes > this.maxJournalBytes && - this.liveJournal.length > 1) - ) { - this.liveJournal.shift(); - this.journalTotalBytes -= this.journalEntryBytes.shift() ?? 0; - this.journalTruncatedEvents += 1; - } + this.appendLiveJournal(event, byteLength); if (TURN_BOUNDARY_TYPES.has(event.type)) { this.compactCurrentTurn(event); @@ -262,7 +260,15 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { this.makeHistoryTruncatedEvent(compactedTurns.length), ); } - const liveJournal = this.liveJournal.slice(); + const liveJournal = this.liveJournal.map((entry) => + isLiveJournalTextSegment(entry) + ? mergeLiveJournalTextEvent( + entry.firstEvent, + entry.lastEvent, + entry.chunks, + ) + : entry, + ); if (this.journalTruncatedEvents > 0) { // Same wire shape as the compacted-window marker: the SDK's // normalizer and type guard both REQUIRE @@ -277,7 +283,7 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { reason: 'replay_window_exceeded', scope: 'live_journal', truncatedEvents: this.journalTruncatedEvents, - retainedEvents: this.liveJournal.length, + retainedEvents: this.journalTotalEvents, maxBytes: this.maxJournalBytes, maxEvents: this.maxJournalEvents, // Pagination anchor — see makeHistoryTruncatedEvent. @@ -368,6 +374,74 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { this.clearTextSlotIndex(); } + private appendLiveJournal(event: BridgeEvent, byteLength?: number): void { + const bytes = byteLength ?? serializedBridgeEventByteLength(event) ?? 0; + const textChunk = liveJournalTextChunk(event); + const current = this.liveJournalTextSegment; + const currentIndex = this.liveJournal.length - 1; + if ( + textChunk && + current && + this.liveJournal[currentIndex] === current && + current.chunks.length < LIVE_JOURNAL_TEXT_CHUNKS_PER_EVENT && + this.journalEntryBytes[currentIndex]! + bytes <= this.maxJournalBytes && + current.sessionUpdate === textChunk.sessionUpdate && + current.parentToolCallId === textChunk.parentToolCallId && + stringArraysEqual(current.sourceRecordIds, textChunk.sourceRecordIds) && + current.promptId === event.promptId && + current.originatorClientId === event.originatorClientId && + current.sessionId === captureSessionId(event) && + hasOnlyTimestampEnvelopeMeta(current.lastEvent._meta) && + hasOnlyTimestampEnvelopeMeta(event._meta) + ) { + current.chunks.push(textChunk.text); + current.lastEvent = event; + this.journalEntryBytes[currentIndex]! += bytes; + this.journalEntryEvents[currentIndex]! += 1; + this.journalTotalBytes += bytes; + this.journalTotalEvents += 1; + } else { + let entry: BridgeEvent | LiveJournalTextSegment = event; + if (textChunk) { + const segment: LiveJournalTextSegment = { + sessionUpdate: textChunk.sessionUpdate, + chunks: [textChunk.text], + sourceRecordIds: textChunk.sourceRecordIds, + parentToolCallId: textChunk.parentToolCallId, + promptId: event.promptId, + originatorClientId: event.originatorClientId, + sessionId: captureSessionId(event), + firstEvent: event, + lastEvent: event, + }; + entry = segment; + this.liveJournalTextSegment = segment; + } else { + this.liveJournalTextSegment = undefined; + } + this.liveJournal.push(entry); + this.journalEntryBytes.push(bytes); + this.journalEntryEvents.push(1); + this.journalTotalBytes += bytes; + this.journalTotalEvents += 1; + } + + while ( + this.liveJournal.length > this.maxJournalEvents || + (this.journalTotalBytes > this.maxJournalBytes && + this.liveJournal.length > 1) + ) { + const dropped = this.liveJournal.shift(); + this.journalTotalBytes -= this.journalEntryBytes.shift() ?? 0; + const droppedEvents = this.journalEntryEvents.shift() ?? 0; + this.journalTotalEvents -= droppedEvents; + this.journalTruncatedEvents += droppedEvents; + if (dropped === this.liveJournalTextSegment) { + this.liveJournalTextSegment = undefined; + } + } + } + private classifySessionUpdate(event: BridgeEvent): void { const data = event.data as SessionUpdateData | undefined; const updateType = data?.update?.sessionUpdate; @@ -379,7 +453,7 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { switch (updateType) { case 'agent_message_chunk': { - if (hasTodoStopGuardDiscreteMeta(data?.update?._meta)) { + if (hasDiscreteMessageMeta(data?.update?._meta)) { this.slots.push({ kind: 'misc', event }); break; } @@ -387,6 +461,10 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { break; } case 'agent_thought_chunk': { + if (hasDiscreteMessageMeta(data?.update?._meta)) { + this.slots.push({ kind: 'misc', event }); + break; + } this.mergeTextSlot('thought', event, data); break; } @@ -574,8 +652,11 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { private resetJournal(): void { this.liveJournal = []; this.journalEntryBytes = []; + this.journalEntryEvents = []; this.journalTotalBytes = 0; + this.journalTotalEvents = 0; this.journalTruncatedEvents = 0; + this.liveJournalTextSegment = undefined; } private addReplaySegment(events: BridgeEvent[], turnCount: number): void { @@ -730,6 +811,93 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { } } +function isLiveJournalTextSegment( + entry: BridgeEvent | LiveJournalTextSegment, +): entry is LiveJournalTextSegment { + return 'firstEvent' in entry; +} + +function liveJournalTextChunk(event: BridgeEvent): + | { + sessionUpdate: 'agent_message_chunk' | 'agent_thought_chunk'; + text: string; + sourceRecordIds?: readonly string[]; + parentToolCallId?: string; + } + | undefined { + if (event.type !== 'session_update') return undefined; + const data = event.data as SessionUpdateData | undefined; + const sessionUpdate = data?.update?.sessionUpdate; + if ( + sessionUpdate !== 'agent_message_chunk' && + sessionUpdate !== 'agent_thought_chunk' + ) { + return undefined; + } + if (!hasOnlyModeledChunkKeys(data)) { + return undefined; + } + if ( + hasDiscreteMessageMeta(data?.update?._meta) || + hasUnmodeledTextMeta(data?.update?._meta) + ) { + return undefined; + } + const content = data?.update?.content; + if (content?.type !== 'text' || typeof content.text !== 'string') { + return undefined; + } + return { + sessionUpdate, + text: content.text, + sourceRecordIds: extractSourceRecordIdsFromMeta(data?.update?._meta), + parentToolCallId: extractParentToolCallIdFromMeta(data?.update?._meta), + }; +} + +// `mergeLiveJournalTextEvent` rebuilds a merged entry by spread-merging +// the segment's first and last source events, so only chunks whose data, +// update, and content carry exactly the modeled keys can join a segment — +// extra data/update keys would leak into the merged aggregate, and extra +// content keys (ACP TextContent `annotations` / `_meta`) would be dropped +// by the `{ type, text }` content rebuild, so such chunks stay raw entries. +function hasOnlyModeledChunkKeys(data: SessionUpdateData | undefined): boolean { + if (!data || !data.update) return false; + const content: unknown = data.update.content; + return ( + Object.keys(data).every((key) => key === 'sessionId' || key === 'update') && + Object.keys(data.update).every( + (key) => key === 'sessionUpdate' || key === 'content' || key === '_meta', + ) && + (content === undefined || + (typeof content === 'object' && + content !== null && + Object.keys(content).every((key) => key === 'type' || key === 'text'))) + ); +} + +function mergeLiveJournalTextEvent( + existing: BridgeEvent, + incoming: BridgeEvent, + chunks: readonly string[], +): BridgeEvent { + const existingData = existing.data as SessionUpdateData; + const incomingData = incoming.data as SessionUpdateData; + return { + ...existing, + ...incoming, + data: { + ...existingData, + ...incomingData, + update: { + ...existingData.update, + ...incomingData.update, + content: { type: 'text', text: chunks.join('') }, + }, + }, + }; +} + function makeMergedSessionUpdateEvent( sessionUpdate: string, text: string, @@ -837,15 +1005,70 @@ function stringArraysEqual( return left.every((value, index) => value === right[index]); } -function hasTodoStopGuardDiscreteMeta(meta: unknown): boolean { +function hasDiscreteMessageMeta(meta: unknown): boolean { return ( typeof meta === 'object' && meta !== null && - (meta as Record)['qwenDiscreteMessage'] === true && - (meta as Record)['source'] === 'todo_stop_guard' + (meta as Record)['qwenDiscreteMessage'] === true ); } +function hasOnlyTimestampEnvelopeMeta(meta: unknown): boolean { + if (meta === undefined) return true; + if (typeof meta !== 'object' || meta === null) return false; + return Object.keys(meta).every( + (key) => key === 'timestamp' || key === 'serverTimestamp', + ); +} + +function hasUnmodeledTextMeta(meta: unknown): boolean { + if (meta === undefined) return false; + if (typeof meta !== 'object' || meta === null) return true; + const record = meta as Record; + for (const [key, value] of Object.entries(record)) { + if (key === 'timestamp' || key === 'serverTimestamp') { + continue; + } + if (key === 'parentToolCallId') { + // Empty strings model "no parent": extractParentToolCallIdFromMeta + // ignores them, so both replay surfaces agree the chunk is top-level. + if (typeof value !== 'string') return true; + continue; + } + if (key === 'subagentType') { + // Display label SubAgentTracker pairs with parentToolCallId. The + // completed-turn path merges by parentToolCallId alone and lets the + // latest meta carry the label, so the live view must match instead + // of splitting on it. + if (typeof value !== 'string') return true; + continue; + } + if (key === 'qwenTranscript') { + if (typeof value !== 'object' || value === null) return true; + const transcript = value as Record; + for (const [field, fieldValue] of Object.entries(transcript)) { + if (field === 'sourceRecordIds') { + if ( + !Array.isArray(fieldValue) || + fieldValue.some((id) => typeof id !== 'string') + ) { + return true; + } + continue; + } + if (field === 'planToolCallId') { + if (typeof fieldValue !== 'string') return true; + continue; + } + return true; + } + continue; + } + return true; + } + return false; +} + function mergeToolCallEvent( existing: BridgeEvent, incoming: BridgeEvent, diff --git a/packages/cli/src/commands/serve.ts b/packages/cli/src/commands/serve.ts index dc79029301b..aab9ae6c513 100644 --- a/packages/cli/src/commands/serve.ts +++ b/packages/cli/src/commands/serve.ts @@ -399,17 +399,19 @@ export const serveCommand: CommandModule = { type: 'number', default: DEFAULT_MAX_JOURNAL_EVENTS, description: - 'Per-session cap on raw events retained in the in-flight live ' + - 'journal (current unfinished turn). When exceeded, the oldest ' + - 'entries are dropped. Must be a positive safe integer.', + 'Per-session cap on replay entries retained in the in-flight live ' + + 'journal (current unfinished turn). Compatible text/thought chunks ' + + 'share bounded entries. When exceeded, the oldest entries are ' + + 'dropped. Must be a positive safe integer.', }) .option('max-journal-bytes', { type: 'number', default: DEFAULT_MAX_JOURNAL_BYTES, description: - 'Per-session byte cap on the in-flight live journal. When ' + - 'exceeded, the oldest entries are dropped (at least one is ' + - 'always kept). Must be a positive safe integer.', + 'Per-session source-event byte cap on the in-flight live journal. ' + + 'When exceeded, the oldest entries are dropped whole (at least ' + + 'one is always kept), so the retained tail can be much smaller ' + + 'than the cap. Must be a positive safe integer.', }) .option('http-bridge', { type: 'boolean', diff --git a/packages/cli/src/serve/types.ts b/packages/cli/src/serve/types.ts index 83c98e1c9a5..16011597d80 100644 --- a/packages/cli/src/serve/types.ts +++ b/packages/cli/src/serve/types.ts @@ -127,15 +127,17 @@ export interface ServeOptions { */ compactedReplayMaxBytes?: number; /** - * Per-session cap on the number of raw events retained in the in-flight - * live journal. Threaded into `BridgeOptions.maxJournalEvents`. Defaults - * to 10 000. Must be a positive safe integer. + * Per-session cap on replay entries retained in the in-flight live journal. + * Compatible text/thought chunks share bounded entries. Threaded into + * `BridgeOptions.maxJournalEvents`. Defaults to 10 000. Must be a positive + * safe integer. */ maxJournalEvents?: number; /** - * Per-session byte cap on the in-flight live journal. Threaded into - * `BridgeOptions.maxJournalBytes`. Defaults to 8 MiB. Must be a positive - * safe integer. + * Per-session source-event byte cap on the in-flight live journal. + * Truncation drops whole entries, so the retained tail can be much smaller + * than the cap. Threaded into `BridgeOptions.maxJournalBytes`. Defaults to + * 8 MiB. Must be a positive safe integer. */ maxJournalBytes?: number; /** diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 5032dfde388..2a605846b33 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -932,7 +932,7 @@ export interface DaemonRestoredSession extends DaemonSession { artifactWarnings?: string[]; /** Compacted events for completed turns (load only). */ compactedReplay?: DaemonEvent[]; - /** Raw events since last turn boundary — current incomplete turn (load only). */ + /** Bounded replay events for the current incomplete turn (load only). */ liveJournal?: DaemonEvent[]; /** True when older persisted records precede this load replay page. */ historyHasMore?: boolean; diff --git a/packages/sdk-typescript/src/daemon/ui/normalizer.ts b/packages/sdk-typescript/src/daemon/ui/normalizer.ts index 064cf988f7f..97d43bac137 100644 --- a/packages/sdk-typescript/src/daemon/ui/normalizer.ts +++ b/packages/sdk-typescript/src/daemon/ui/normalizer.ts @@ -467,14 +467,16 @@ function normalizeHistoryTruncated( } const fullTranscriptAvailable = event.data['fullTranscriptAvailable']; const limits = [ - maxEvents === undefined ? undefined : `${maxEvents} events`, + maxEvents === undefined + ? undefined + : `${maxEvents} ${scope === 'live_journal' ? 'replay entries' : 'events'}`, `${maxBytes} bytes`, ] .filter((limit): limit is string => limit !== undefined) .join(' / '); const text = scope === 'live_journal' - ? `History truncated for live turn replay: kept the latest ${retainedEvents} events and dropped ${truncatedEvents} older replay events (limits: ${limits}). ${ + ? `History truncated for live turn replay: kept the latest ${retainedEvents} source events and dropped ${truncatedEvents} older source events (limits: ${limits}). ${ fullTranscriptAvailable ? 'Complete content remains available after the turn finishes.' : 'Complete content is not available for automatic recovery.' @@ -780,12 +782,14 @@ function normalizeSessionUpdate( const text = getTextContent(update['content']); if (!text) return []; const parentToolCallId = extractParentToolCallId(update); + const meta = extractUpdateMeta(update); return [ { ...base, type: 'thought.text.delta' as const, text, ...(parentToolCallId ? { parentToolCallId } : {}), + ...(meta ? { meta } : {}), }, ]; } diff --git a/packages/sdk-typescript/test/unit/daemonUi.test.ts b/packages/sdk-typescript/test/unit/daemonUi.test.ts index 93b10e7b977..ffa42b564fb 100644 --- a/packages/sdk-typescript/test/unit/daemonUi.test.ts +++ b/packages/sdk-typescript/test/unit/daemonUi.test.ts @@ -296,6 +296,53 @@ describe('daemon UI normalizer and transcript reducer', () => { }); }); + it('keeps discrete thought messages separate with their metadata', () => { + const makeThought = (id: number, text: string, taskId: string) => + normalizeDaemonEvent({ + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_thought_chunk', + content: { type: 'text', text }, + _meta: { + qwenDiscreteMessage: true, + backgroundTask: { taskId }, + }, + }, + }, + }); + + const state = reduceDaemonTranscriptEvents( + createDaemonTranscriptState({ now: 1 }), + [ + ...makeThought(11, 'first thought', 'task-a'), + ...makeThought(12, 'second thought', 'task-b'), + ], + { now: 2 }, + ); + + expect(state.blocks).toMatchObject([ + { + kind: 'thought', + text: 'first thought', + meta: { + qwenDiscreteMessage: true, + backgroundTask: { taskId: 'task-a' }, + }, + }, + { + kind: 'thought', + text: 'second thought', + meta: { + qwenDiscreteMessage: true, + backgroundTask: { taskId: 'task-b' }, + }, + }, + ]); + }); + it('keeps discrete assistant messages separate from normal text blocks', () => { const normalBefore = normalizeDaemonEvent({ id: 11, @@ -3711,9 +3758,12 @@ describe('daemon UI reducer state machine (PR-E)', () => { source: 'history_truncated', data, text: expect.stringContaining( - 'kept the latest 10000 events and dropped 16371 older replay events', + 'kept the latest 10000 source events and dropped 16371 older source events', ) as string, }); + expect((event as { text: string }).text).toContain( + 'limits: 10000 replay entries / 8388608 bytes', + ); expect((event as { text: string }).text).toContain( 'Complete content remains available after the turn finishes.', );