fix(daemon): compacted session replay for long-session recovery

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": {

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);

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,20 +1777,18 @@ 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),
       };
     }
 
     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,
@@ -1820,7 +1841,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 +1958,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
           clientId,
           createdAt: racedEntry.createdAt,
           state: racedEntry.restoreState ?? {},
+          ...replayFieldsFor(racedEntry, action),
         };
       }
 
@@ -1970,6 +1992,7 @@ export function createHttpAcpBridge(opts: BridgeOptions): HttpAcpBridge {
         clientId,
         createdAt: entry.createdAt,
         state,
+        ...replayFieldsFor(entry, action),
       };
     })().finally(() => {
       ci?.pendingRestoreIds.delete(req.sessionId);

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`. */