From 7d0b03f228491416daca1f0c504603d1dfe9a7fe Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Tue, 2 Jun 2026 16:40:42 +0800 Subject: [PATCH 1/6] fix(daemon): compacted session replay for long-session recovery MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace unbounded raw-event replay with turn-boundary compaction. On each turn_complete, streaming chunks merge into single events, tool call sequences fold to final state, transient signals drop. loadSession returns O(turns) compacted events instead of O(streaming_tokens) raw events. Key decisions: - Synchronous snapshot() eliminates watermark vs async-read race - Slot-based compaction preserves event ordering across types - liveJournal carries raw events for current incomplete turn - resume only returns lastEventId (no replay payload) - All new fields optional for backward compatibility ๐Ÿค– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/acp-bridge/package.json | 4 + packages/acp-bridge/src/bridge.ts | 30 +- packages/acp-bridge/src/bridgeTypes.ts | 6 + .../acp-bridge/src/compactionEngine.test.ts | 594 ++++++++++++++++++ packages/acp-bridge/src/compactionEngine.ts | 314 +++++++++ packages/acp-bridge/src/eventBus.ts | 19 + .../src/daemon/DaemonSessionClient.ts | 52 +- packages/sdk-typescript/src/daemon/index.ts | 1 + packages/sdk-typescript/src/daemon/types.ts | 6 + 9 files changed, 1006 insertions(+), 20 deletions(-) create mode 100644 packages/acp-bridge/src/compactionEngine.test.ts create mode 100644 packages/acp-bridge/src/compactionEngine.ts diff --git a/packages/acp-bridge/package.json b/packages/acp-bridge/package.json index 0e65f1c6cf3..0b08592ecd6 100644 --- a/packages/acp-bridge/package.json +++ b/packages/acp-bridge/package.json @@ -75,6 +75,10 @@ "types": "./dist/internal/testUtils.d.ts", "import": "./dist/internal/testUtils.js" }, + "./compactionEngine": { + "types": "./dist/compactionEngine.d.ts", + "import": "./dist/compactionEngine.js" + }, "./package.json": "./package.json" }, "scripts": { diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 60ce3a97408..580e4942e73 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -29,6 +29,7 @@ import { import type { ShellCommandResult } from './bridgeTypes.js'; import type { AcpChannel } from './channel.js'; import { EventBus, DEFAULT_RING_SIZE, type BridgeEvent } from './eventBus.js'; +import { TurnBoundaryCompactionEngine } from './compactionEngine.js'; import { BridgeChannelClosedError, BridgeTimeoutError, @@ -1669,11 +1670,14 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { } }; + const createSessionEventBus = (): EventBus => + new EventBus(eventRingSize, undefined, new TurnBoundaryCompactionEngine()); + const createSessionEntry = ( ci: ChannelInfo, sessionId: string, workspaceCwd: string, - events = new EventBus(eventRingSize), + events = createSessionEventBus(), ): SessionEntry => { const entry: SessionEntry = { sessionId, @@ -1732,6 +1736,25 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { ); }; + const replayFieldsFor = ( + entry: { events: EventBus }, + action: 'load' | 'resume', + ): Pick< + BridgeRestoredSession, + 'compactedReplay' | 'liveJournal' | 'lastEventId' + > => { + const snapshot = entry.events.snapshotReplay(); + if (!snapshot) return { lastEventId: entry.events.lastEventId }; + if (action === 'load') { + return { + compactedReplay: snapshot.compactedTurns, + liveJournal: snapshot.liveJournal, + lastEventId: snapshot.lastEventId, + }; + } + return { lastEventId: snapshot.lastEventId }; + }; + async function restoreSession( action: 'load' | 'resume', req: BridgeRestoreSessionRequest, @@ -1754,6 +1777,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { // Late attachers get the same ACP state the original restore // caller saw; spawn-only sessions don't carry a state payload. state: existing.restoreState ?? {}, + ...replayFieldsFor(existing, action), }; } @@ -1820,7 +1844,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { throw new SessionLimitExceededError(maxSessions); } - const restoreEvents = new EventBus(eventRingSize); + const restoreEvents = createSessionEventBus(); let registeredEntry: SessionEntry | undefined; let ci: ChannelInfo | undefined; // Live counter shared with coalesced waiters (see InFlightRestore @@ -1937,6 +1961,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { clientId, createdAt: racedEntry.createdAt, state: racedEntry.restoreState ?? {}, + ...replayFieldsFor(racedEntry, action), }; } @@ -1970,6 +1995,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { clientId, createdAt: entry.createdAt, state, + ...replayFieldsFor(entry, action), }; })().finally(() => { ci?.pendingRestoreIds.delete(req.sessionId); diff --git a/packages/acp-bridge/src/bridgeTypes.ts b/packages/acp-bridge/src/bridgeTypes.ts index 0db96d6dc02..99f274b25b1 100644 --- a/packages/acp-bridge/src/bridgeTypes.ts +++ b/packages/acp-bridge/src/bridgeTypes.ts @@ -79,6 +79,12 @@ export type BridgeSessionState = LoadSessionResponse | ResumeSessionResponse; export interface BridgeRestoredSession extends BridgeSession { /** ACP state returned by `session/load` / `session/resume`. */ state: BridgeSessionState; + /** Compacted events for all completed turns (O(turns) size). */ + compactedReplay?: BridgeEvent[]; + /** Raw events since last turn boundary (current incomplete turn). */ + liveJournal?: BridgeEvent[]; + /** High-water mark event ID โ€” client uses this as initial SSE cursor. */ + lastEventId?: number; } /** Sparse summary used by `GET /workspace/:id/sessions`. */ diff --git a/packages/acp-bridge/src/compactionEngine.test.ts b/packages/acp-bridge/src/compactionEngine.test.ts new file mode 100644 index 00000000000..bd309920b33 --- /dev/null +++ b/packages/acp-bridge/src/compactionEngine.test.ts @@ -0,0 +1,594 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, it, expect } from 'vitest'; +import { TurnBoundaryCompactionEngine } from './compactionEngine.js'; +import type { BridgeEvent } from './eventBus.js'; + +function makeTextChunk(id: number, text: string): BridgeEvent { + return { + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text }, + }, + }, + }; +} + +function makeThoughtChunk(id: number, text: string): BridgeEvent { + return { + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_thought_chunk', + content: { type: 'text', text }, + }, + }, + }; +} + +function makeUserMessage(id: number, text: string): BridgeEvent { + return { + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'user_message_chunk', + content: { type: 'text', text }, + }, + }, + }; +} + +function makeToolCall( + id: number, + toolCallId: string, + status: string, + extra: Record = {}, +): BridgeEvent { + return { + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'tool_call', + toolCallId, + status, + ...extra, + }, + }, + }; +} + +function makeToolCallUpdate( + id: number, + toolCallId: string, + status: string, + extra: Record = {}, +): BridgeEvent { + return { + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'tool_call_update', + toolCallId, + status, + ...extra, + }, + }, + }; +} + +function makeTurnComplete(id: number): BridgeEvent { + return { + id, + v: 1, + type: 'turn_complete', + data: { stopReason: 'end_turn' }, + }; +} + +function makeTurnError(id: number): BridgeEvent { + return { + id, + v: 1, + type: 'turn_error', + data: { error: 'cancelled' }, + }; +} + +function makePermissionRequest(id: number, requestId: string): BridgeEvent { + return { + id, + v: 1, + type: 'permission_request', + data: { requestId, request: { tool: 'Bash', command: 'ls' } }, + }; +} + +function makePermissionResolved(id: number, requestId: string): BridgeEvent { + return { + id, + v: 1, + type: 'permission_resolved', + data: { requestId, outcome: 'approved' }, + }; +} + +function makeModelSwitched(id: number, modelId: string): BridgeEvent { + return { + id, + v: 1, + type: 'model_switched', + data: { modelId }, + }; +} + +function makeAvailableCommandsUpdate(id: number): BridgeEvent { + return { + id, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'available_commands_update', + commands: ['/help'], + }, + }, + }; +} + +function extractTexts(events: BridgeEvent[]): string[] { + return events + .filter((e) => e.type === 'session_update') + .map((e) => { + const data = e.data as { update?: { content?: { text?: string } } }; + return data?.update?.content?.text ?? ''; + }) + .filter((t) => t !== ''); +} + +describe('TurnBoundaryCompactionEngine', () => { + describe('basic compaction', () => { + it('merges consecutive text chunks into a single event on turn_complete', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'Hello')); + engine.ingest(makeTextChunk(2, ' ')); + engine.ingest(makeTextChunk(3, 'world')); + engine.ingest(makeTurnComplete(4)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(2); // merged text + turn_complete + expect(snap.liveJournal).toHaveLength(0); + expect(snap.lastEventId).toBe(4); + + const textEvent = snap.compactedTurns[0]!; + expect(textEvent.id).toBe(3); // last chunk's id + expect(textEvent.type).toBe('session_update'); + const data = textEvent.data as { + update: { sessionUpdate: string; content: { text: string } }; + }; + expect(data.update.sessionUpdate).toBe('agent_message_chunk'); + expect(data.update.content.text).toBe('Hello world'); + }); + + it('merges consecutive thought chunks', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeThoughtChunk(1, 'Let me ')); + engine.ingest(makeThoughtChunk(2, 'think...')); + engine.ingest(makeTextChunk(3, 'Answer')); + engine.ingest(makeTurnComplete(4)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(3); // thought + text + turn_complete + + const thoughtEvent = snap.compactedTurns[0]!; + const data = thoughtEvent.data as { + update: { sessionUpdate: string; content: { text: string } }; + }; + expect(data.update.sessionUpdate).toBe('agent_thought_chunk'); + expect(data.update.content.text).toBe('Let me think...'); + }); + + it('keeps user messages as-is', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeUserMessage(1, 'How are you?')); + engine.ingest(makeTextChunk(2, 'I am fine')); + engine.ingest(makeTurnComplete(3)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(3); + const data = snap.compactedTurns[0]!.data as { + update: { sessionUpdate: string; content: { text: string } }; + }; + expect(data.update.sessionUpdate).toBe('user_message_chunk'); + expect(data.update.content.text).toBe('How are you?'); + expect(snap.compactedTurns[0]!.id).toBe(1); + }); + }); + + describe('tool call folding', () => { + it('folds tool_call + tool_call_updates into single final-state event', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'Let me check')); + engine.ingest(makeToolCall(2, 'tc1', 'running', { title: 'Read file' })); + engine.ingest( + makeToolCallUpdate(3, 'tc1', 'running', { + content: 'reading...', + }), + ); + engine.ingest( + makeToolCallUpdate(4, 'tc1', 'done', { + rawOutput: 'file contents', + }), + ); + engine.ingest(makeTextChunk(5, 'Done')); + engine.ingest(makeTurnComplete(6)); + + const snap = engine.snapshot(); + // text("Let me check") + tool(tc1 final) + text("Done") + turn_complete + expect(snap.compactedTurns).toHaveLength(4); + + const toolEvent = snap.compactedTurns[1]!; + const data = toolEvent.data as { + update: { + toolCallId: string; + status: string; + title: string; + rawOutput: string; + }; + }; + expect(data.update.toolCallId).toBe('tc1'); + expect(data.update.status).toBe('done'); + expect(data.update.title).toBe('Read file'); + expect(data.update.rawOutput).toBe('file contents'); + expect(toolEvent.id).toBe(4); // last update's id + }); + + it('preserves tool call order when multiple tools run', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeToolCall(1, 'tc1', 'running', { title: 'Tool A' })); + engine.ingest(makeToolCall(2, 'tc2', 'running', { title: 'Tool B' })); + engine.ingest(makeToolCallUpdate(3, 'tc1', 'done')); + engine.ingest(makeToolCallUpdate(4, 'tc2', 'done')); + engine.ingest(makeTurnComplete(5)); + + const snap = engine.snapshot(); + const toolEvents = snap.compactedTurns.filter( + (e) => + e.type === 'session_update' && + (e.data as { update?: { sessionUpdate?: string } })?.update + ?.sessionUpdate === 'tool_call', + ); + expect(toolEvents).toHaveLength(2); + expect( + (toolEvents[0]!.data as { update: { title: string } }).update.title, + ).toBe('Tool A'); + expect( + (toolEvents[1]!.data as { update: { title: string } }).update.title, + ).toBe('Tool B'); + }); + }); + + describe('text segmentation across tool calls', () => { + it('preserves separate text segments before and after tool calls', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'Before')); + engine.ingest(makeTextChunk(2, ' tool')); + engine.ingest(makeToolCall(3, 'tc1', 'running')); + engine.ingest(makeToolCallUpdate(4, 'tc1', 'done')); + engine.ingest(makeTextChunk(5, 'After')); + engine.ingest(makeTextChunk(6, ' tool')); + engine.ingest(makeTurnComplete(7)); + + const texts = extractTexts(engine.snapshot().compactedTurns); + expect(texts).toEqual(['Before tool', 'After tool']); + }); + }); + + describe('transient event filtering', () => { + it('drops transient events (slow_client_warning, replay_complete, etc.)', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'Hello')); + engine.ingest({ + v: 1, + type: 'slow_client_warning', + data: { queueSize: 200 }, + }); + engine.ingest({ + id: 2, + v: 1, + type: 'replay_complete', + data: { replayedCount: 5 }, + }); + engine.ingest(makeTurnComplete(3)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(2); // text + turn_complete + expect(snap.liveJournal).toHaveLength(0); + }); + }); + + describe('latest-wins events', () => { + it('keeps only the most recent available_commands_update per turn', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeAvailableCommandsUpdate(1)); + engine.ingest(makeAvailableCommandsUpdate(2)); + engine.ingest(makeAvailableCommandsUpdate(3)); + engine.ingest(makeTurnComplete(4)); + + const snap = engine.snapshot(); + const cmdUpdates = snap.compactedTurns.filter( + (e) => + (e.data as { update?: { sessionUpdate?: string } })?.update + ?.sessionUpdate === 'available_commands_update', + ); + expect(cmdUpdates).toHaveLength(1); + expect(cmdUpdates[0]!.id).toBe(3); + }); + }); + + describe('permission events', () => { + it('preserves permission_request and permission_resolved', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'I need permission')); + engine.ingest(makePermissionRequest(2, 'perm-1')); + engine.ingest(makePermissionResolved(3, 'perm-1')); + engine.ingest(makeTextChunk(4, 'Done')); + engine.ingest(makeTurnComplete(5)); + + const snap = engine.snapshot(); + const permEvents = snap.compactedTurns.filter( + (e) => + e.type === 'permission_request' || e.type === 'permission_resolved', + ); + expect(permEvents).toHaveLength(2); + }); + }); + + describe('model_switched events', () => { + it('preserves model_switched events', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeModelSwitched(1, 'opus-4')); + engine.ingest(makeTextChunk(2, 'Response')); + engine.ingest(makeTurnComplete(3)); + + const snap = engine.snapshot(); + const modelEvents = snap.compactedTurns.filter( + (e) => e.type === 'model_switched', + ); + expect(modelEvents).toHaveLength(1); + expect((modelEvents[0]!.data as { modelId: string }).modelId).toBe( + 'opus-4', + ); + }); + }); + + describe('liveJournal (incomplete turn)', () => { + it('accumulates raw events in liveJournal before turn completes', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'H')); + engine.ingest(makeTextChunk(2, 'i')); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(0); + expect(snap.liveJournal).toHaveLength(2); + expect(snap.lastEventId).toBe(2); + }); + + it('clears liveJournal on turn completion', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'Hello')); + engine.ingest(makeTurnComplete(2)); + engine.ingest(makeTextChunk(3, 'New turn')); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(2); + expect(snap.liveJournal).toHaveLength(1); + expect(snap.liveJournal[0]!.id).toBe(3); + }); + }); + + describe('multi-turn sessions', () => { + it('compacts multiple turns independently', () => { + const engine = new TurnBoundaryCompactionEngine(); + // Turn 1 + engine.ingest(makeUserMessage(1, 'Hello')); + engine.ingest(makeTextChunk(2, 'Hi')); + engine.ingest(makeTextChunk(3, ' there')); + engine.ingest(makeTurnComplete(4)); + // Turn 2 + engine.ingest(makeUserMessage(5, 'Bye')); + engine.ingest(makeTextChunk(6, 'Good')); + engine.ingest(makeTextChunk(7, 'bye')); + engine.ingest(makeTurnComplete(8)); + + const snap = engine.snapshot(); + expect(snap.lastEventId).toBe(8); + // Turn 1: user + merged_text + turn_complete + // Turn 2: user + merged_text + turn_complete + expect(snap.compactedTurns).toHaveLength(6); + const texts = extractTexts(snap.compactedTurns); + expect(texts).toContain('Hello'); + expect(texts).toContain('Hi there'); + expect(texts).toContain('Bye'); + expect(texts).toContain('Goodbye'); + }); + }); + + describe('turn_error compaction', () => { + it('compacts on turn_error the same as turn_complete', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'partial')); + engine.ingest(makeTextChunk(2, ' response')); + engine.ingest(makeTurnError(3)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(2); // merged text + turn_error + expect(snap.liveJournal).toHaveLength(0); + const texts = extractTexts(snap.compactedTurns); + expect(texts).toEqual(['partial response']); + }); + }); + + describe('snapshot consistency', () => { + it('returns defensive copies', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'test')); + engine.ingest(makeTurnComplete(2)); + + const a = engine.snapshot(); + const b = engine.snapshot(); + expect(a.compactedTurns).not.toBe(b.compactedTurns); + expect(a.compactedTurns).toEqual(b.compactedTurns); + expect(a.liveJournal).not.toBe(b.liveJournal); + }); + + it('lastEventId is always consistent with content', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'a')); + expect(engine.snapshot().lastEventId).toBe(1); + + engine.ingest(makeTextChunk(2, 'b')); + expect(engine.snapshot().lastEventId).toBe(2); + + engine.ingest(makeTurnComplete(3)); + expect(engine.snapshot().lastEventId).toBe(3); + }); + }); + + describe('seed', () => { + it('seeds the engine from a persisted snapshot', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.seed({ + compactedTurns: [makeTextChunk(10, 'from disk'), makeTurnComplete(11)], + lastEventId: 11, + }); + + // New events build on top of the seeded state + engine.ingest(makeTextChunk(12, 'live')); + engine.ingest(makeTurnComplete(13)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(4); // 2 seeded + 2 new + expect(snap.lastEventId).toBe(13); + }); + }); + + describe('close', () => { + it('ignores events after close', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTextChunk(1, 'before')); + engine.close(); + engine.ingest(makeTextChunk(2, 'after')); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(0); + expect(snap.liveJournal).toHaveLength(0); + }); + }); + + describe('_meta preservation', () => { + it('preserves _meta from the last text chunk', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest({ + id: 1, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'Hello' }, + _meta: { usage: { input: 10 } }, + }, + }, + }); + engine.ingest({ + id: 2, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: ' world' }, + _meta: { usage: { input: 10, output: 50 }, durationMs: 1200 }, + }, + }, + }); + engine.ingest(makeTurnComplete(3)); + + const snap = engine.snapshot(); + const textEvent = snap.compactedTurns[0]!; + const data = textEvent.data as { update: { _meta: unknown } }; + expect(data.update._meta).toEqual({ + usage: { input: 10, output: 50 }, + durationMs: 1200, + }); + }); + }); + + describe('edge cases', () => { + it('handles empty turn (turn_complete with no preceding events)', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeTurnComplete(1)); + + const snap = engine.snapshot(); + expect(snap.compactedTurns).toHaveLength(1); // just turn_complete + expect(snap.compactedTurns[0]!.type).toBe('turn_complete'); + }); + + it('handles events without id (synthetic frames)', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest({ + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'no id' }, + }, + }, + }); + engine.ingest(makeTurnComplete(1)); + + const snap = engine.snapshot(); + expect(snap.lastEventId).toBe(1); + const texts = extractTexts(snap.compactedTurns); + expect(texts).toEqual(['no id']); + }); + + it('handles thought then text interleaved with tool calls', () => { + const engine = new TurnBoundaryCompactionEngine(); + engine.ingest(makeThoughtChunk(1, 'thinking')); + engine.ingest(makeThoughtChunk(2, '...')); + engine.ingest(makeTextChunk(3, 'answer')); + engine.ingest(makeToolCall(4, 'tc1', 'running')); + engine.ingest(makeToolCallUpdate(5, 'tc1', 'done')); + engine.ingest(makeTextChunk(6, 'after tool')); + engine.ingest(makeTurnComplete(7)); + + const snap = engine.snapshot(); + // thought + text("answer") + tool + text("after tool") + turn_complete + expect(snap.compactedTurns).toHaveLength(5); + + const thoughtData = snap.compactedTurns[0]!.data as { + update: { sessionUpdate: string; content: { text: string } }; + }; + expect(thoughtData.update.sessionUpdate).toBe('agent_thought_chunk'); + expect(thoughtData.update.content.text).toBe('thinking...'); + }); + }); +}); diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts new file mode 100644 index 00000000000..1b39afabb0c --- /dev/null +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -0,0 +1,314 @@ +/** + * @license + * Copyright 2025 Qwen Team + * SPDX-License-Identifier: Apache-2.0 + */ + +import type { + BridgeEvent, + CompactionEngine, + SessionReplaySnapshot, +} from './eventBus.js'; + +export type { CompactionEngine, SessionReplaySnapshot }; + +interface SessionUpdateData { + update?: { + sessionUpdate?: string; + content?: { type?: string; text?: string }; + toolCallId?: string; + status?: string; + _meta?: unknown; + [key: string]: unknown; + }; + [key: string]: unknown; +} + +const TURN_BOUNDARY_TYPES = new Set(['turn_complete', 'turn_error']); +const TRANSIENT_TYPES = new Set([ + 'slow_client_warning', + 'client_evicted', + 'replay_complete', + 'stream_error', +]); +const LATEST_WINS_UPDATES = new Set([ + 'available_commands_update', + 'current_mode_update', +]); + +type CompactedSlot = + | { kind: 'text'; chunks: string[]; lastEventId: number; lastMeta: unknown } + | { + kind: 'thought'; + chunks: string[]; + lastEventId: number; + lastMeta: unknown; + } + | { kind: 'tool'; toolCallId: string; event: BridgeEvent } + | { kind: 'misc'; event: BridgeEvent } + | { kind: 'latestWins'; key: string; event: BridgeEvent }; + +/** + * Compaction engine that merges events at turn boundaries. + * + * On each `turn_complete` / `turn_error`, all accumulated events for that + * turn are folded: consecutive text/thought chunks merge into single events, + * tool call sequences fold to final state, transient signals are dropped. + * The relative ordering of different event types is preserved. + * + * The result is a replay log whose size is O(conversation_turns), not + * O(streaming_tokens). Typical compression: 25-30x for chatty sessions. + */ +export class TurnBoundaryCompactionEngine implements CompactionEngine { + private compactedTurns: BridgeEvent[] = []; + private liveJournal: BridgeEvent[] = []; + private lastEventId = 0; + private closed = false; + + private slots: CompactedSlot[] = []; + private toolSlotIndex: Map = new Map(); + + ingest(event: BridgeEvent): void { + if (this.closed) return; + if (event.id !== undefined) { + this.lastEventId = event.id; + } + + if (TRANSIENT_TYPES.has(event.type)) return; + + this.liveJournal.push(event); + + if (TURN_BOUNDARY_TYPES.has(event.type)) { + this.compactCurrentTurn(event); + return; + } + + if (event.type === 'session_update') { + this.classifySessionUpdate(event); + return; + } + + this.slots.push({ kind: 'misc', event }); + } + + snapshot(): SessionReplaySnapshot { + return { + compactedTurns: this.compactedTurns.slice(), + liveJournal: this.liveJournal.slice(), + lastEventId: this.lastEventId, + }; + } + + seed(snapshot: { compactedTurns: BridgeEvent[]; lastEventId: number }): void { + if (this.closed) return; + this.compactedTurns = snapshot.compactedTurns.slice(); + this.lastEventId = snapshot.lastEventId; + } + + close(): void { + if (this.closed) return; + this.closed = true; + this.compactedTurns = []; + this.liveJournal = []; + this.slots = []; + this.toolSlotIndex.clear(); + } + + private classifySessionUpdate(event: BridgeEvent): void { + const data = event.data as SessionUpdateData | undefined; + const updateType = data?.update?.sessionUpdate; + + if (!updateType) { + this.slots.push({ kind: 'misc', event }); + return; + } + + switch (updateType) { + case 'agent_message_chunk': { + const text = data?.update?.content?.text ?? ''; + const lastSlot = this.slots[this.slots.length - 1]; + if (lastSlot && lastSlot.kind === 'text') { + lastSlot.chunks.push(text); + if (event.id !== undefined) lastSlot.lastEventId = event.id; + lastSlot.lastMeta = data?.update?._meta ?? lastSlot.lastMeta; + } else { + this.slots.push({ + kind: 'text', + chunks: [text], + lastEventId: event.id ?? 0, + lastMeta: data?.update?._meta, + }); + } + break; + } + case 'agent_thought_chunk': { + const text = data?.update?.content?.text ?? ''; + const lastSlot = this.slots[this.slots.length - 1]; + if (lastSlot && lastSlot.kind === 'thought') { + lastSlot.chunks.push(text); + if (event.id !== undefined) lastSlot.lastEventId = event.id; + lastSlot.lastMeta = data?.update?._meta ?? lastSlot.lastMeta; + } else { + this.slots.push({ + kind: 'thought', + chunks: [text], + lastEventId: event.id ?? 0, + lastMeta: data?.update?._meta, + }); + } + break; + } + case 'tool_call': + case 'tool_call_update': { + const toolCallId = data?.update?.toolCallId; + if (!toolCallId) { + this.slots.push({ kind: 'misc', event }); + break; + } + const existingIdx = this.toolSlotIndex.get(toolCallId); + if (existingIdx !== undefined) { + const slot = this.slots[existingIdx] as Extract< + CompactedSlot, + { kind: 'tool' } + >; + slot.event = mergeToolCallEvent(slot.event, event); + } else { + const normalizedEvent = normalizeToolCallType(event); + this.toolSlotIndex.set(toolCallId, this.slots.length); + this.slots.push({ + kind: 'tool', + toolCallId, + event: normalizedEvent, + }); + } + break; + } + default: { + if (LATEST_WINS_UPDATES.has(updateType)) { + const existingIdx = this.slots.findIndex( + (s) => s.kind === 'latestWins' && s.key === updateType, + ); + if (existingIdx !== -1) { + ( + this.slots[existingIdx] as Extract< + CompactedSlot, + { kind: 'latestWins' } + > + ).event = event; + } else { + this.slots.push({ kind: 'latestWins', key: updateType, event }); + } + } else { + this.slots.push({ kind: 'misc', event }); + } + break; + } + } + } + + private compactCurrentTurn(boundaryEvent: BridgeEvent): void { + const compacted: BridgeEvent[] = []; + + for (const slot of this.slots) { + switch (slot.kind) { + case 'text': + compacted.push( + makeMergedSessionUpdateEvent( + 'agent_message_chunk', + slot.chunks.join(''), + slot.lastEventId, + slot.lastMeta, + ), + ); + break; + case 'thought': + compacted.push( + makeMergedSessionUpdateEvent( + 'agent_thought_chunk', + slot.chunks.join(''), + slot.lastEventId, + slot.lastMeta, + ), + ); + break; + case 'tool': + case 'misc': + case 'latestWins': + compacted.push(slot.event); + break; + default: + break; + } + } + + compacted.push(boundaryEvent); + this.compactedTurns.push(...compacted); + this.liveJournal = []; + this.slots = []; + this.toolSlotIndex.clear(); + } +} + +function makeMergedSessionUpdateEvent( + sessionUpdate: string, + text: string, + eventId: number, + meta: unknown, +): BridgeEvent { + return { + id: eventId || undefined, + v: 1, + type: 'session_update', + data: { + update: { + sessionUpdate, + content: { type: 'text', text }, + ...(meta != null ? { _meta: meta } : {}), + }, + }, + }; +} + +function normalizeToolCallType(event: BridgeEvent): BridgeEvent { + const data = event.data as SessionUpdateData | undefined; + if (data?.update?.sessionUpdate === 'tool_call_update') { + return { + ...event, + data: { + ...data, + update: { ...data.update, sessionUpdate: 'tool_call' }, + }, + }; + } + return event; +} + +function mergeToolCallEvent( + existing: BridgeEvent, + incoming: BridgeEvent, +): BridgeEvent { + const existingData = existing.data as SessionUpdateData | undefined; + const incomingData = incoming.data as SessionUpdateData | undefined; + const existingUpdate = existingData?.update ?? {}; + const incomingUpdate = incomingData?.update ?? {}; + + const merged: Record = { ...existingUpdate }; + for (const [key, value] of Object.entries(incomingUpdate)) { + if (value !== undefined && value !== null) { + merged[key] = value; + } + } + // Always use 'tool_call' as the compacted type + merged['sessionUpdate'] = 'tool_call'; + + return { + id: incoming.id ?? existing.id, + v: 1, + type: 'session_update', + data: { + ...existingData, + ...incomingData, + update: merged, + }, + }; +} diff --git a/packages/acp-bridge/src/eventBus.ts b/packages/acp-bridge/src/eventBus.ts index 6a17c0a57ea..0b692722efd 100644 --- a/packages/acp-bridge/src/eventBus.ts +++ b/packages/acp-bridge/src/eventBus.ts @@ -19,6 +19,18 @@ * Aborting the supplied AbortSignal closes the iterator promptly. */ +export interface SessionReplaySnapshot { + compactedTurns: BridgeEvent[]; + liveJournal: BridgeEvent[]; + lastEventId: number; +} + +export interface CompactionEngine { + ingest(event: BridgeEvent): void; + snapshot(): SessionReplaySnapshot; + close(): void; +} + export const EVENT_SCHEMA_VERSION = 1 as const; /** A single frame published on the bus. */ @@ -166,8 +178,13 @@ export class EventBus { constructor( private readonly ringSize: number = DEFAULT_RING_SIZE, private readonly maxSubscribers: number = DEFAULT_MAX_SUBSCRIBERS, + private readonly compactionEngine?: CompactionEngine, ) {} + snapshotReplay(): SessionReplaySnapshot | undefined { + return this.compactionEngine?.snapshot(); + } + /** Most recent id ever assigned by `publish`. 0 if no events published. */ get lastEventId(): number { return this.nextId - 1; @@ -211,6 +228,7 @@ export class EventBus { ...input, }; this.ring.push(event); + this.compactionEngine?.ingest(event); // Eviction-by-shift is O(n) once the ring is full. At the current // default `ringSize=8000` (#3803 ยง02) the per-publish shift work // measures in low milliseconds on chatty sessions โ€” still well @@ -550,6 +568,7 @@ export class EventBus { this.closed = true; for (const sub of this.subs) sub.queue.close(); this.subs.clear(); + this.compactionEngine?.close(); } } diff --git a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts index 2ff7eb40d32..7c40807ab8c 100644 --- a/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts +++ b/packages/sdk-typescript/src/daemon/DaemonSessionClient.ts @@ -30,6 +30,12 @@ import type { SessionMetadataResult, } from './types.js'; +/** Compacted replay snapshot returned by the daemon on session load. */ +export interface DaemonReplaySnapshot { + compactedReplay: DaemonEvent[]; + liveJournal: DaemonEvent[]; +} + export interface DaemonSessionClientOptions { client: DaemonClient; session: DaemonSession; @@ -42,6 +48,8 @@ export interface DaemonSessionClientOptions { * `Last-Event-ID` resume cursors. */ lastEventId?: number; + /** Compacted replay snapshot from daemon load response. */ + replaySnapshot?: DaemonReplaySnapshot; } export interface DaemonSessionSubscribeOptions extends SubscribeOptions { @@ -68,6 +76,7 @@ export class DaemonSessionClient { readonly client: DaemonClient; readonly session: DaemonSession; readonly state: DaemonSessionState; + readonly replaySnapshot: DaemonReplaySnapshot; private lastSeenEventId: number | undefined; private subscriptionActive = false; private readonly _pendingPrompts = new Map< @@ -82,6 +91,10 @@ export class DaemonSessionClient { this.client = opts.client; this.session = { ...opts.session }; this.state = { ...(opts.state ?? {}) }; + this.replaySnapshot = opts.replaySnapshot ?? { + compactedReplay: [], + liveJournal: [], + }; this.lastSeenEventId = validateLastEventId(opts.lastEventId); } @@ -136,27 +149,30 @@ export class DaemonSessionClient { req: RestoreSessionRequest = {}, clientId?: string, ): Promise { - const { state, ...session } = await client.loadSession( - sessionId, - req, - clientId, - ); + const { + state, + compactedReplay, + liveJournal, + lastEventId: serverLastEventId, + ...session + } = await client.loadSession(sessionId, req, clientId); return new DaemonSessionClient({ client, session, state, - lastEventId: 0, + lastEventId: serverLastEventId ?? 0, + replaySnapshot: { + compactedReplay: compactedReplay ?? [], + liveJournal: liveJournal ?? [], + }, }); } /** * Resumes an existing daemon session without requesting history replay. - * Seeds the first event subscription from the start of the daemon - * replay ring (`lastEventId: 0`) symmetric with `load()` โ€” the agent's - * `unstable_resumeSession` schedules an `available_commands_update` - * via `setTimeout(0)`, which can publish to the daemon bus between - * the HTTP response and the consumer's first `events()` call. Seeding - * ensures that frame is observed instead of dropped. + * When the daemon returns a watermark (`lastEventId`), uses it as the + * initial SSE cursor. Falls back to 0 for older daemons so + * post-resume events (e.g. `available_commands_update`) are captured. */ static async resume( client: DaemonClient, @@ -164,16 +180,16 @@ export class DaemonSessionClient { req: RestoreSessionRequest = {}, clientId?: string, ): Promise { - const { state, ...session } = await client.resumeSession( - sessionId, - req, - clientId, - ); + const { + state, + lastEventId: serverLastEventId, + ...session + } = await client.resumeSession(sessionId, req, clientId); return new DaemonSessionClient({ client, session, state, - lastEventId: 0, + lastEventId: serverLastEventId ?? 0, }); } diff --git a/packages/sdk-typescript/src/daemon/index.ts b/packages/sdk-typescript/src/daemon/index.ts index bef503db4c2..9daa3a5f91e 100644 --- a/packages/sdk-typescript/src/daemon/index.ts +++ b/packages/sdk-typescript/src/daemon/index.ts @@ -24,6 +24,7 @@ export { } from './DaemonAuthFlow.js'; export { DaemonSessionClient, + type DaemonReplaySnapshot, type DaemonSessionClientOptions, type DaemonSessionSubscribeOptions, } from './DaemonSessionClient.js'; diff --git a/packages/sdk-typescript/src/daemon/types.ts b/packages/sdk-typescript/src/daemon/types.ts index 1f8cbfec30b..b4ca2fab5ae 100644 --- a/packages/sdk-typescript/src/daemon/types.ts +++ b/packages/sdk-typescript/src/daemon/types.ts @@ -152,6 +152,12 @@ export interface DaemonSessionState { /** Returned from `POST /session/:id/load` and `POST /session/:id/resume`. */ export interface DaemonRestoredSession extends DaemonSession { state: DaemonSessionState; + /** Compacted events for completed turns (load only). */ + compactedReplay?: DaemonEvent[]; + /** Raw events since last turn boundary โ€” current incomplete turn (load only). */ + liveJournal?: DaemonEvent[]; + /** Event bus watermark โ€” used as initial SSE cursor. */ + lastEventId?: number; } /** Sparse session record returned by `GET /workspace/:id/sessions`. */ From 6ac1245f83d40fbf993ff15317e3a6ff02b54fce Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Tue, 2 Jun 2026 23:27:23 +0800 Subject: [PATCH 2/6] refactor: extract mergeTextSlot helper + add integration tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review suggestions: - Extract shared mergeTextSlot() for agent_message_chunk/thought_chunk - Add 4 EventBus+CompactionEngine integration tests covering snapshotReplay(), liveJournal, and close lifecycle ๐Ÿค– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- .../acp-bridge/src/compactionEngine.test.ts | 55 +++++++++++++++++++ packages/acp-bridge/src/compactionEngine.ts | 51 ++++++++--------- 2 files changed, 78 insertions(+), 28 deletions(-) diff --git a/packages/acp-bridge/src/compactionEngine.test.ts b/packages/acp-bridge/src/compactionEngine.test.ts index bd309920b33..70b865f0f04 100644 --- a/packages/acp-bridge/src/compactionEngine.test.ts +++ b/packages/acp-bridge/src/compactionEngine.test.ts @@ -6,6 +6,7 @@ import { describe, it, expect } from 'vitest'; import { TurnBoundaryCompactionEngine } from './compactionEngine.js'; +import { EventBus } from './eventBus.js'; import type { BridgeEvent } from './eventBus.js'; function makeTextChunk(id: number, text: string): BridgeEvent { @@ -592,3 +593,57 @@ describe('TurnBoundaryCompactionEngine', () => { }); }); }); + +describe('EventBus + CompactionEngine integration', () => { + it('snapshotReplay returns compacted state after publish + turn_complete', () => { + const engine = new TurnBoundaryCompactionEngine(); + const bus = new EventBus(100, undefined, engine); + + bus.publish({ type: 'session_update', data: { update: { sessionUpdate: 'user_message_chunk', content: { type: 'text', text: 'hello' } } } }); + bus.publish({ type: 'session_update', data: { update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'Hi' } } } }); + bus.publish({ type: 'session_update', data: { update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: ' there' } } } }); + bus.publish({ type: 'turn_complete', data: { stopReason: 'end_turn' } }); + + const snapshot = bus.snapshotReplay(); + expect(snapshot).toBeDefined(); + expect(snapshot!.lastEventId).toBe(4); + expect(snapshot!.compactedTurns).toHaveLength(3); + expect(snapshot!.liveJournal).toHaveLength(0); + + const mergedText = snapshot!.compactedTurns[1]!.data as { + update: { content: { text: string } }; + }; + expect(mergedText.update.content.text).toBe('Hi there'); + }); + + it('snapshotReplay returns undefined when no engine is configured', () => { + const bus = new EventBus(100); + bus.publish({ type: 'session_update', data: {} }); + expect(bus.snapshotReplay()).toBeUndefined(); + }); + + it('liveJournal contains raw events for incomplete turn', () => { + const engine = new TurnBoundaryCompactionEngine(); + const bus = new EventBus(100, undefined, engine); + + bus.publish({ type: 'session_update', data: { update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'streaming' } } } }); + bus.publish({ type: 'session_update', data: { update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: '...' } } } }); + + const snapshot = bus.snapshotReplay()!; + expect(snapshot.compactedTurns).toHaveLength(0); + expect(snapshot.liveJournal).toHaveLength(2); + expect(snapshot.lastEventId).toBe(2); + }); + + it('compaction engine is closed when bus closes', () => { + const engine = new TurnBoundaryCompactionEngine(); + const bus = new EventBus(100, undefined, engine); + + bus.publish({ type: 'session_update', data: { update: { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'test' } } } }); + bus.close(); + + const snapshot = engine.snapshot(); + expect(snapshot.compactedTurns).toHaveLength(0); + expect(snapshot.liveJournal).toHaveLength(0); + }); +}); diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts index 1b39afabb0c..2ab9543cbe2 100644 --- a/packages/acp-bridge/src/compactionEngine.ts +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -125,37 +125,11 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { switch (updateType) { case 'agent_message_chunk': { - const text = data?.update?.content?.text ?? ''; - const lastSlot = this.slots[this.slots.length - 1]; - if (lastSlot && lastSlot.kind === 'text') { - lastSlot.chunks.push(text); - if (event.id !== undefined) lastSlot.lastEventId = event.id; - lastSlot.lastMeta = data?.update?._meta ?? lastSlot.lastMeta; - } else { - this.slots.push({ - kind: 'text', - chunks: [text], - lastEventId: event.id ?? 0, - lastMeta: data?.update?._meta, - }); - } + this.mergeTextSlot('text', event, data); break; } case 'agent_thought_chunk': { - const text = data?.update?.content?.text ?? ''; - const lastSlot = this.slots[this.slots.length - 1]; - if (lastSlot && lastSlot.kind === 'thought') { - lastSlot.chunks.push(text); - if (event.id !== undefined) lastSlot.lastEventId = event.id; - lastSlot.lastMeta = data?.update?._meta ?? lastSlot.lastMeta; - } else { - this.slots.push({ - kind: 'thought', - chunks: [text], - lastEventId: event.id ?? 0, - lastMeta: data?.update?._meta, - }); - } + this.mergeTextSlot('thought', event, data); break; } case 'tool_call': @@ -206,6 +180,27 @@ export class TurnBoundaryCompactionEngine implements CompactionEngine { } } + private mergeTextSlot( + kind: 'text' | 'thought', + event: BridgeEvent, + data: SessionUpdateData | undefined, + ): void { + const text = data?.update?.content?.text ?? ''; + const lastSlot = this.slots[this.slots.length - 1]; + if (lastSlot && lastSlot.kind === kind) { + lastSlot.chunks.push(text); + if (event.id !== undefined) lastSlot.lastEventId = event.id; + lastSlot.lastMeta = data?.update?._meta ?? lastSlot.lastMeta; + } else { + this.slots.push({ + kind, + chunks: [text], + lastEventId: event.id ?? 0, + lastMeta: data?.update?._meta, + }); + } + } + private compactCurrentTurn(boundaryEvent: BridgeEvent): void { const compacted: BridgeEvent[] = []; From 38e4e2de4dbebb4d35931e25bc822a0944460357 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 3 Jun 2026 02:46:00 +0800 Subject: [PATCH 3/6] fix: update bridge test assertions for new replay fields MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add compactedReplay/liveJournal/lastEventId to toEqual assertions in load/resume/attach bridge tests. ๐Ÿค– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/acp-bridge/src/bridge.test.ts | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/packages/acp-bridge/src/bridge.test.ts b/packages/acp-bridge/src/bridge.test.ts index 0af227547e0..1847a4d6719 100644 --- a/packages/acp-bridge/src/bridge.test.ts +++ b/packages/acp-bridge/src/bridge.test.ts @@ -818,6 +818,9 @@ describe('createHttpAcpBridge', () => { clientId: expect.stringMatching(/^client_/), createdAt: expect.any(String), state: { configOptions: [] }, + compactedReplay: [], + liveJournal: [], + lastEventId: 0, }); expect(handles[0]?.agent.loadSessionCalls).toEqual([ { sessionId: 'persisted-1', cwd: WS_A, mcpServers: [] }, @@ -919,6 +922,7 @@ describe('createHttpAcpBridge', () => { clientId: expect.stringMatching(/^client_/), createdAt: expect.any(String), state: { modes: null }, + lastEventId: 0, }); expect(handles[0]?.agent.loadSessionCalls).toHaveLength(0); expect(handles[0]?.agent.resumeSessionCalls).toEqual([ @@ -962,6 +966,7 @@ describe('createHttpAcpBridge', () => { clientId: expect.stringMatching(/^client_/), createdAt: expect.any(String), state: { _meta: { tag: 'restored-foo' } }, + lastEventId: expect.any(Number), }); expect(attached.clientId).not.toBe(loaded.clientId); expect(handles[0]?.agent.loadSessionCalls).toHaveLength(1); From 0fcdc86934ef0f38c02a8babfd164a98e4826b16 Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 3 Jun 2026 03:44:07 +0800 Subject: [PATCH 4/6] fix: use EVENT_SCHEMA_VERSION constant instead of hardcoded v:1 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ๐Ÿค– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/acp-bridge/src/compactionEngine.ts | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/packages/acp-bridge/src/compactionEngine.ts b/packages/acp-bridge/src/compactionEngine.ts index 2ab9543cbe2..89c25f6cf35 100644 --- a/packages/acp-bridge/src/compactionEngine.ts +++ b/packages/acp-bridge/src/compactionEngine.ts @@ -4,10 +4,11 @@ * SPDX-License-Identifier: Apache-2.0 */ -import type { - BridgeEvent, - CompactionEngine, - SessionReplaySnapshot, +import { + EVENT_SCHEMA_VERSION, + type BridgeEvent, + type CompactionEngine, + type SessionReplaySnapshot, } from './eventBus.js'; export type { CompactionEngine, SessionReplaySnapshot }; @@ -252,7 +253,7 @@ function makeMergedSessionUpdateEvent( ): BridgeEvent { return { id: eventId || undefined, - v: 1, + v: EVENT_SCHEMA_VERSION, type: 'session_update', data: { update: { @@ -298,7 +299,7 @@ function mergeToolCallEvent( return { id: incoming.id ?? existing.id, - v: 1, + v: EVENT_SCHEMA_VERSION, type: 'session_update', data: { ...existingData, From 327a5ce3e8313f8b7a60703d2f7357a92bacb9ef Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 3 Jun 2026 10:44:52 +0800 Subject: [PATCH 5/6] fix: verify SDK replay fields in tests + guard ingest on publish MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Update load/resume test mocks to return lastEventId/compactedReplay/liveJournal - Verify replaySnapshot population and SSE cursor from server watermark - Wrap compactionEngine.ingest() in try/catch to maintain BX9_p contract ๐Ÿค– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/acp-bridge/src/eventBus.ts | 7 ++++++- .../test/unit/DaemonSessionClient.test.ts | 17 +++++++++++------ 2 files changed, 17 insertions(+), 7 deletions(-) diff --git a/packages/acp-bridge/src/eventBus.ts b/packages/acp-bridge/src/eventBus.ts index 0b692722efd..cc7bb18e73f 100644 --- a/packages/acp-bridge/src/eventBus.ts +++ b/packages/acp-bridge/src/eventBus.ts @@ -228,7 +228,12 @@ export class EventBus { ...input, }; this.ring.push(event); - this.compactionEngine?.ingest(event); + try { + this.compactionEngine?.ingest(event); + } catch { + // CompactionEngine is best-effort; a throw must not break the + // publish() never-throws contract (BX9_p). + } // Eviction-by-shift is O(n) once the ring is full. At the current // default `ringSize=8000` (#3803 ยง02) the per-publish shift work // measures in low milliseconds on chatty sessions โ€” still well diff --git a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts index 947ff9d6c5f..a7fbc38fe67 100644 --- a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts @@ -198,6 +198,9 @@ describe('DaemonSessionClient', () => { attached: false, clientId: 'client-1', state: { configOptions: [] }, + lastEventId: 42, + compactedReplay: [{ id: 1, v: 1, type: 'session_update', data: {} }], + liveJournal: [{ id: 42, v: 1, type: 'session_update', data: {} }], }); } if (req.url.endsWith('/session/s-1/events')) { @@ -214,15 +217,17 @@ describe('DaemonSessionClient', () => { expect(session.sessionId).toBe('s-1'); expect(session.clientId).toBe('client-1'); expect(session.state).toEqual({ configOptions: [] }); + expect(session.replaySnapshot.compactedReplay).toHaveLength(1); + expect(session.replaySnapshot.liveJournal).toHaveLength(1); expect(JSON.parse(calls[0]!.body!)).toEqual({ cwd: '/work/a' }); for await (const _event of session.events()) { /* empty */ } - expect(calls[1]?.headers['last-event-id']).toBe('0'); + expect(calls[1]?.headers['last-event-id']).toBe('42'); }); - it('resumes an existing daemon session and seeds replay from the start', async () => { + it('resumes an existing daemon session using server watermark', async () => { const { fetch, calls } = recordingFetch((req) => { if (req.url.endsWith('/session/s-1/resume')) { return jsonResponse(200, { @@ -231,6 +236,7 @@ describe('DaemonSessionClient', () => { attached: true, clientId: 'client-1', state: { modes: null }, + lastEventId: 99, }); } if (req.url.endsWith('/session/s-1/events')) { @@ -245,13 +251,12 @@ describe('DaemonSessionClient', () => { expect(session.attached).toBe(true); expect(session.clientId).toBe('client-1'); expect(session.state).toEqual({ modes: null }); + expect(session.replaySnapshot.compactedReplay).toHaveLength(0); + expect(session.replaySnapshot.liveJournal).toHaveLength(0); for await (const _event of session.events()) { /* empty */ } - // Symmetric to load(): `unstable_resumeSession` schedules an - // `available_commands_update` via setTimeout(0) on the agent side, - // so the SDK seeds the subscription from the start of the ring. - expect(calls[1]?.headers['last-event-id']).toBe('0'); + expect(calls[1]?.headers['last-event-id']).toBe('99'); }); it('replays from id 0 on freshly-created sessions so startup-window guardrail events are observable (codex review fix #1)', async () => { From d4172921fb899ed1f3b40e613408bdd547691eec Mon Sep 17 00:00:00 2001 From: doudouOUC Date: Wed, 3 Jun 2026 13:43:46 +0800 Subject: [PATCH 6/6] fix: update stale comment and test title for new watermark semantics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ๐Ÿค– Generated with [Qwen Code](https://github.com/QwenLM/qwen-code) --- packages/acp-bridge/src/bridge.ts | 13 +++++-------- .../test/unit/DaemonSessionClient.test.ts | 2 +- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/packages/acp-bridge/src/bridge.ts b/packages/acp-bridge/src/bridge.ts index 580e4942e73..f047084493d 100644 --- a/packages/acp-bridge/src/bridge.ts +++ b/packages/acp-bridge/src/bridge.ts @@ -1784,14 +1784,11 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge { const inFlight = inFlightRestores.get(req.sessionId); if (inFlight) { // Cross-action races BOTH ways must reject. A `resume` arriving - // while a `load` is in flight cannot quietly coalesce: the load - // is replaying full history through SSE on a shared EventBus, - // and `DaemonSessionClient.resume()` seeds `lastEventId: 0`, - // which means the resume client would receive every replayed - // frame โ€” directly violating resume's "no UI replay" contract. - // The mirror direction (`load` onto `resume`) is rejected for - // the same reason: a load caller expects history but resume - // didn't replay any. Same-action coalescing is unaffected. + // while a `load` is in flight cannot quietly coalesce: load + // returns compacted replay + watermark while resume returns only + // a watermark โ€” mixing the two on a shared EventBus would give + // the resume client unexpected replay data or the load client a + // missing snapshot. Same-action coalescing is unaffected. if (action !== inFlight.action) { throw new RestoreInProgressError( req.sessionId, diff --git a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts index a7fbc38fe67..667847bc82f 100644 --- a/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts +++ b/packages/sdk-typescript/test/unit/DaemonSessionClient.test.ts @@ -189,7 +189,7 @@ describe('DaemonSessionClient', () => { expect(calls[1]?.headers['last-event-id']).toBe('0'); }); - it('loads an existing daemon session and seeds replay from the start', async () => { + it('loads an existing daemon session using server watermark and replay snapshot', async () => { const { fetch, calls } = recordingFetch((req) => { if (req.url.endsWith('/session/s-1/load')) { return jsonResponse(200, {