fix(daemon): cross-client sync follow-up cleanup (epoch-reset resync, approval-mode serialization, catch-up indicator)

packages/acp-bridge/src/eventBus.test.ts

@@ -90,7 +90,12 @@ describe('EventBus', () => {
     expect(events.map((e) => e.data)).toEqual([
       'a',
       'b',
-      expect.objectContaining({ lastEventId: 2, replayedCount: 2 }),
+      // D4: canonical `lastReplayedEventId` + deprecated `lastEventId` alias.
+      expect.objectContaining({
+        lastReplayedEventId: 2,
+        lastEventId: 2,
+        replayedCount: 2,
+      }),
       'c',
     ]);
     abort.abort();
@@ -619,9 +624,15 @@ describe('EventBus', () => {
       abort.abort();
     });
 
-    it('does NOT emit state_resync_required when ring is empty', async () => {
-      // No publishes yet → earliestInRing is undefined → resync check
-      // skipped. Subscriber waits for live events.
+    it('emits epoch_reset resync when lastEventId is past the bus high-water (D1)', async () => {
+      // doudouOUC #4484 post-merge review (D1): a fresh bus (nextId=1,
+      // empty ring) that receives a consumer presenting `lastEventId: 5`
+      // means the consumer's cursor is from a PREVIOUS bus epoch (daemon
+      // restart rebuilt the EventBus). Pre-fix this slid past the
+      // `ring_evicted` check (empty ring) and emitted a bare
+      // `replay_complete{replayedCount:0}` — a false "you're caught up"
+      // while the consumer's reducer still held dead-epoch state. Now it
+      // must emit `state_resync_required{reason:'epoch_reset'}` first.
       const bus = new EventBus(10);
       const abort = new AbortController();
       const iter = bus.subscribe({
@@ -633,18 +644,75 @@ describe('EventBus', () => {
       const out: BridgeEvent[] = [];
       for await (const e of iter) {
         out.push(e);
-        // The empty-ring case still emits `replay_complete` (zero
-        // frames replayed) so consumers always see the catch-up signal
-        // — then the one live event. 2 total.
+        // resync + replay_complete (0 frames) + 1 live = 3 total.
+        if (out.length === 3) break;
+      }
+      expect(out[0]?.type).toBe('state_resync_required');
+      expect(out[0]?.id).toBeUndefined();
+      const data = out[0]?.data as {
+        reason: string;
+        lastDeliveredId: number;
+        earliestAvailableId: number;
+      };
+      expect(data.reason).toBe('epoch_reset');
+      expect(data.lastDeliveredId).toBe(5);
+      expect(data.earliestAvailableId).toBe(1);
+      expect(out[1]?.type).toBe('replay_complete');
+      expect(out[1]?.data).toMatchObject({ replayedCount: 0 });
+      expect(out[2]?.type).toBe('foo');
+      expect(out[2]?.id).toBe(1);
+      abort.abort();
+    });
+
+    it('epoch_reset replays the WHOLE fresh ring (stale cursor must not filter new low ids)', async () => {
+      // After a restart the new epoch starts ids at 1 again. A consumer
+      // reconnecting with `lastEventId: 50` (dead epoch) must still receive
+      // the fresh ring's low-id events — filtering replay by 50 would drop
+      // ids 1..3 entirely, leaving the consumer permanently behind.
+      const bus = new EventBus(10);
+      for (let i = 1; i <= 3; i++) bus.publish({ type: 'foo', data: i });
+      const abort = new AbortController();
+      const iter = bus.subscribe({
+        lastEventId: 50,
+        signal: abort.signal,
+      });
+      const out: BridgeEvent[] = [];
+      for await (const e of iter) {
+        out.push(e);
+        // resync + 3 replay frames + replay_complete = 5.
+        if (out.length === 5) break;
+      }
+      expect(out[0]?.type).toBe('state_resync_required');
+      expect((out[0]?.data as { reason: string }).reason).toBe('epoch_reset');
+      // All three fresh events replay despite ids < stale cursor.
+      expect(out.slice(1, 4).map((e) => e.id)).toEqual([1, 2, 3]);
+      expect(out[4]?.type).toBe('replay_complete');
+      expect(out[4]?.data).toMatchObject({ replayedCount: 3 });
+      abort.abort();
+    });
+
+    it('does NOT emit epoch_reset at the caught-up boundary (lastEventId === high-water)', async () => {
+      // Consumer fully caught up: lastEventId equals the bus high-water
+      // (nextId - 1). nextId is one past it, so `lastEventId >= nextId` is
+      // false — no epoch reset. Off-by-one guard for D1.
+      const bus = new EventBus(10);
+      for (let i = 1; i <= 3; i++) bus.publish({ type: 'foo', data: i });
+      // high-water is 3; nextId is 4. lastEventId: 3 is the caught-up case.
+      const abort = new AbortController();
+      const iter = bus.subscribe({
+        lastEventId: 3,
+        signal: abort.signal,
+      });
+      setTimeout(() => bus.publish({ type: 'foo', data: 99 }), 0);
+      const out: BridgeEvent[] = [];
+      for await (const e of iter) {
+        out.push(e);
+        // replay_complete (0 frames) + 1 live = 2.
         if (out.length === 2) break;
       }
-      // No resync frame — but replay_complete (id-less sentinel) +
-      // the live event.
       expect(out.some((e) => e.type === 'state_resync_required')).toBe(false);
       expect(out[0]?.type).toBe('replay_complete');
-      expect(out[0]?.data).toMatchObject({ replayedCount: 0 });
-      expect(out[1]?.type).toBe('foo');
-      expect(out[1]?.id).toBe(1);
+      expect(out[1]?.id).toBe(4);
       abort.abort();
     });
 

packages/acp-bridge/src/eventBus.ts

@@ -386,21 +386,54 @@ export class EventBus {
       // loadSession clears the flag, but the frames stay on the
       // wire so SDK has the option to compute a "what you missed"
       // diff later. This is network-friendly (no extra reconnect).
-      const earliestInRing = this.ring[0]?.id;
-      if (
-        earliestInRing !== undefined &&
-        earliestInRing > opts.lastEventId + 1
-      ) {
+      // Epoch-reset detection (doudouOUC #4484 post-merge review, D1).
+      // `this.nextId` is the next id this bus will assign, so the bus has
+      // only ever emitted ids `< nextId` THIS epoch. A consumer presenting
+      // `lastEventId >= nextId` therefore saw an id this epoch never
+      // produced — the only way that happens is a previous bus epoch
+      // (daemon restart / EventBus rebuild resets `nextId` to 1 and clears
+      // the ring). The `ring_evicted` check below is structurally blind to
+      // this: after a restart the ring is empty (`earliestInRing ===
+      // undefined`), so it is skipped and the consumer would otherwise get
+      // a bare `replay_complete{replayedCount:0}` — a false "you're caught
+      // up" while its accumulated reducer state is stale data from the dead
+      // epoch. Emit `state_resync_required` (reason `epoch_reset`) first.
+      const epochReset = opts.lastEventId >= this.nextId;
+      if (epochReset) {
         queue.forcePush({
           v: EVENT_SCHEMA_VERSION,
           type: 'state_resync_required',
           data: {
-            reason: 'ring_evicted',
+            reason: 'epoch_reset',
             lastDeliveredId: opts.lastEventId,
-            earliestAvailableId: earliestInRing,
+            // Ring is typically empty right after a restart; fall back to
+            // `nextId` (the first id this epoch will assign) so the field
+            // stays meaningful ("fresh sequence starts here").
+            earliestAvailableId: this.ring[0]?.id ?? this.nextId,
           },
         });
+      } else {
+        const earliestInRing = this.ring[0]?.id;
+        if (
+          earliestInRing !== undefined &&
+          earliestInRing > opts.lastEventId + 1
+        ) {
+          queue.forcePush({
+            v: EVENT_SCHEMA_VERSION,
+            type: 'state_resync_required',
+            data: {
+              reason: 'ring_evicted',
+              lastDeliveredId: opts.lastEventId,
+              earliestAvailableId: earliestInRing,
+            },
+          });
+        }
       }
+      // After an epoch reset the consumer's cursor belongs to a dead epoch,
+      // so every current-epoch event is "new" to it. Filtering replay by the
+      // stale `lastEventId` (e.g. 50) would drop the fresh low-id events
+      // (1,2,3…) entirely. Replay the whole current ring in that case.
+      const replayFrom = epochReset ? 0 : opts.lastEventId;
       // Force-push replay frames so they bypass the per-subscriber size
       // cap. The cap protects against a slow live consumer; replay is
       // already historical and silently dropping it would undermine the
@@ -415,7 +448,7 @@ export class EventBus {
         // undefined here — but the type system can't see that since
         // BridgeEvent.id is optional for synthetic terminal frames.
         // Guard explicitly to keep narrow typing without runtime cost.
-        if (e.id !== undefined && e.id > opts.lastEventId) {
+        if (e.id !== undefined && e.id > replayFrom) {
           queue.forcePush(e);
           replayedCount += 1;
           lastReplayedId = e.id;
@@ -443,8 +476,17 @@ export class EventBus {
         v: EVENT_SCHEMA_VERSION,
         type: 'replay_complete',
         data: {
+          // D4 (doudouOUC #4484 post-merge review): `lastReplayedEventId`
+          // is the canonical wire name — the old `lastEventId` collided
+          // semantically with the SSE protocol's `Last-Event-ID` (envelope
+          // `id`) in raw daemon traces. Emit both: `lastReplayedEventId`
+          // for current SDKs and `lastEventId` as a deprecated alias so
+          // pre-rename consumers keep working (additive, non-breaking).
           ...(lastReplayedId !== undefined
-            ? { lastEventId: lastReplayedId }
+            ? {
+                lastReplayedEventId: lastReplayedId,
+                lastEventId: lastReplayedId,
+              }
             : {}),
           replayedCount,
         },

packages/sdk-typescript/src/daemon/events.ts

@@ -277,9 +277,13 @@ export interface DaemonStreamErrorData {
  */
 export interface DaemonStateResyncRequiredData {
   /**
-   * Machine-readable resync reason. Currently always `'ring_evicted'`
-   * (the only case the daemon emits this frame for); reserved for
-   * future causes (e.g. `'schema_version_bump'`).
+   * Machine-readable resync reason. One of:
+   * - `'ring_evicted'`: consumer's `Last-Event-ID` fell behind the ring's
+   *   earliest surviving id (same-epoch gap).
+   * - `'epoch_reset'`: consumer's `Last-Event-ID` is past the bus
+   *   high-water — its cursor is from a previous bus epoch (daemon
+   *   restart rebuilt the EventBus). The whole fresh ring is replayed.
+   * Reserved for future causes (e.g. `'schema_version_bump'`).
    */
   reason: string;
   /** Consumer's `Last-Event-ID` at reconnect time. */

packages/sdk-typescript/src/daemon/ui/normalizer.ts

@@ -155,12 +155,23 @@ export function normalizeDaemonEvent(
     case 'state_resync_required':
       return normalizeStateResyncRequired(event, base);
 
-    case 'prompt_cancelled':
-      return [{ ...base, type: 'prompt.cancelled' }];
+    case 'prompt_cancelled': {
+      // Forward the optional `reason` (e.g. `'forward_failed'` from the
+      // bridge's C3 compensating broadcast) so consumers can distinguish a
+      // user cancel from a forward failure.
+      const reason = stringField(event.data, 'reason');
+      return [
+        { ...base, type: 'prompt.cancelled', ...(reason ? { reason } : {}) },
+      ];
+    }
 
     case 'replay_complete': {
       const replayedCount = numberField(event.data, 'replayedCount') ?? 0;
-      const lastReplayedEventId = numberField(event.data, 'lastEventId');
+      // D4: prefer the canonical `lastReplayedEventId`; fall back to the
+      // deprecated `lastEventId` alias for daemons predating the rename.
+      const lastReplayedEventId =
+        numberField(event.data, 'lastReplayedEventId') ??
+        numberField(event.data, 'lastEventId');
       return [
         {
           ...base,
@@ -1188,3 +1199,9 @@ function numberField(value: unknown, key: string): number | undefined {
   const v = value[key];
   return typeof v === 'number' && Number.isFinite(v) ? v : undefined;
 }
+
+function stringField(value: unknown, key: string): string | undefined {
+  if (!isRecord(value)) return undefined;
+  const v = value[key];
+  return typeof v === 'string' && v.length > 0 ? v : undefined;
+}

packages/sdk-typescript/src/daemon/ui/types.ts

@@ -257,6 +257,14 @@ export interface DaemonUiStateResyncRequiredEvent extends DaemonUiEventBase {
  */
 export interface DaemonUiPromptCancelledEvent extends DaemonUiEventBase {
   type: 'prompt.cancelled';
+  /**
+   * Why the turn was cancelled. Absent for a user-initiated cancel;
+   * `'forward_failed'` when the daemon synthesized the cancel because the
+   * prompt forward rejected after the user echo was already published (the
+   * bridge's C3 compensating broadcast). Lets the UI distinguish "peer
+   * cancelled" from "the request failed to reach the agent".
+   */
+  reason?: 'forward_failed' | (string & {});
 }
 
 /**

packages/sdk-typescript/test/unit/daemonUi.test.ts

@@ -5043,6 +5043,21 @@ describe('cross-client event recognition (prompt_cancelled / replay_complete)',
         originatorClientId: 'client-X',
       }),
     ]);
+    // No reason for a plain user cancel.
+    expect(events[0]).not.toHaveProperty('reason');
+  });
+
+  it('forwards the prompt_cancelled reason (C3 forward_failed)', () => {
+    const events = normalizeDaemonEvent({
+      id: 1,
+      v: 1,
+      type: 'prompt_cancelled',
+      data: { sessionId: 's1', reason: 'forward_failed' },
+    } as never);
+    expect(events[0]).toMatchObject({
+      type: 'prompt.cancelled',
+      reason: 'forward_failed',
+    });
   });
 
   it('normalizes replay_complete to session.replay_complete with count', () => {
@@ -5059,6 +5074,20 @@ describe('cross-client event recognition (prompt_cancelled / replay_complete)',
     });
   });
 
+  it('prefers canonical lastReplayedEventId over the deprecated lastEventId alias (D4)', () => {
+    const events = normalizeDaemonEvent({
+      id: 1,
+      v: 1,
+      type: 'replay_complete',
+      // Both present, different values — canonical must win.
+      data: { replayedCount: 2, lastReplayedEventId: 9, lastEventId: 7 },
+    } as never);
+    expect(events[0]).toMatchObject({
+      type: 'session.replay_complete',
+      lastReplayedEventId: 9,
+    });
+  });
+
   it('replay_complete with zero replay (empty ring) normalizes cleanly', () => {
     const events = normalizeDaemonEvent({
       id: 1,