fix(events): snapshot cadence follow-ups (history doubling, open counts, deferred checks)

src/server/chat/snapshot-cadence.test.ts

@@ -7,7 +7,7 @@
  * the tail — the only part a state load replays — bounded during the turn.
  */
 
-import { describe, it, expect, beforeEach, afterEach } from 'vitest'
+import { describe, it, expect, beforeEach, afterEach, vi } from 'vitest'
 import Database from 'better-sqlite3'
 import { initEventStore, getEventStore } from '../events/store.js'
 import { emitSessionInitialized, emitUserMessage } from '../events/session.js'
@@ -30,6 +30,7 @@ beforeEach(() => {
 })
 
 afterEach(() => {
+  vi.useRealTimers()
   db.close()
 })
 
@@ -146,6 +147,76 @@ describe('createSnapshotCadence', () => {
     expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(0)
   })
 
+  it('snapshots at the next quiet point when the periodic check found the turn busy', () => {
+    vi.useFakeTimers({ toFake: ['Date'] })
+    // The check runs at most every checkIntervalMs, on whatever event comes
+    // in. A sub-agent is nearly always streaming or running a tool, so the
+    // check kept landing on a busy moment and the snapshot was put off to the
+    // end of the turn (seen live: a 7-12 minute sub-agent run, tail over
+    // budget, no mid-turn snapshot). Once deferred, the next message or tool
+    // round end checks again.
+    emitSessionInitialized('s1', 'p', '/tmp', 'win-1')
+    const store = getEventStore()
+    // Already over budget when the cadence starts.
+    for (let i = 0; i < 10; i++)
+      store.append('s1', {
+        type: 'tool.output',
+        data: { messageId: 'm0', toolCallId: 't0', stream: 'stdout', content: 'x'.repeat(200) },
+      } as never)
+
+    const cadence = createSnapshotCadence({
+      sessionManager: fakeSessionManager(),
+      sessionId: 's1',
+      append: rawAppend('s1'),
+      maxBytes: 500,
+      checkIntervalMs: 60 * 60_000,
+    })
+
+    // The periodic check lands mid-message: deferred.
+    cadence.append({ type: 'message.start', data: { messageId: 'a1', role: 'assistant' } })
+    for (let i = 0; i < 10; i++) cadence.append(delta('x'.repeat(200)))
+    expect(store.getLatestSnapshotSeq('s1')).toBe(0)
+
+    // Far from the next periodic check, but a message ends: quiet point.
+    vi.setSystemTime(Date.now() + 5_000)
+    cadence.append({ type: 'message.done', data: { messageId: 'a1' } })
+
+    expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(0)
+  })
+
+  it('checks a deferred snapshot again at most every 5 s', () => {
+    vi.useFakeTimers({ toFake: ['Date'] })
+    // Each check scans the tail; parallel tools or sub-agents can keep the
+    // turn busy over many quiet points in a row.
+    emitSessionInitialized('s1', 'p', '/tmp', 'win-1')
+    const store = getEventStore()
+    for (let i = 0; i < 10; i++)
+      store.append('s1', {
+        type: 'tool.output',
+        data: { messageId: 'm0', toolCallId: 't0', stream: 'stdout', content: 'x'.repeat(200) },
+      } as never)
+
+    const cadence = createSnapshotCadence({
+      sessionManager: fakeSessionManager(),
+      sessionId: 's1',
+      append: rawAppend('s1'),
+      maxBytes: 500,
+      checkIntervalMs: 60 * 60_000,
+    })
+    cadence.append({ type: 'message.start', data: { messageId: 'a1', role: 'assistant' } })
+    const tail = vi.spyOn(store, 'getEventLogTail')
+
+    vi.setSystemTime(Date.now() + 4_000)
+    cadence.append({ type: 'message.done', data: { messageId: 'a1' } })
+    expect(tail).not.toHaveBeenCalled()
+    expect(store.getLatestSnapshotSeq('s1')).toBe(0)
+
+    vi.setSystemTime(Date.now() + 1_000)
+    cadence.append({ type: 'message.start', data: { messageId: 'a2', role: 'assistant' } })
+    cadence.append({ type: 'message.done', data: { messageId: 'a2' } })
+    expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(0)
+  })
+
   it('is not blocked by a tool call left unresolved by a crash', () => {
     emitSessionInitialized('s1', 'p', '/tmp', 'win-1')
     // Interrupted before its result: nothing will ever close it.
@@ -212,6 +283,67 @@ describe('createSnapshotCadence', () => {
     expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(0)
   })
 
+  it('keeps one history entry per compaction and retry across snapshots', () => {
+    // A snapshot was built from the previous one plus the events since, and
+    // that stream also replays the previous snapshot's compactions and retries
+    // as synthetic events: the fold counted them twice, so every snapshot
+    // doubled them. Seen live: 7 compactions stored as 2.3 million entries,
+    // a 433 MB snapshot rewritten at every turn end, and the disk full.
+    emitSessionInitialized('s1', 'p', '/tmp', 'win-1')
+    const store = getEventStore()
+    store.append('s1', {
+      type: 'context.compacted',
+      data: { closedWindowId: 'win-1', newWindowId: 'win-2', beforeTokens: 70_000, afterTokens: 0, summary: 'sum' },
+    })
+    store.append('s1', {
+      type: 'pattern.retry',
+      data: { attempt: 1, maxAttempts: 3, messageId: 'm1', pattern: 'p', field: 'content', matchedContent: 'x' },
+    } as never)
+    const cadence = createSnapshotCadence({
+      sessionManager: fakeSessionManager(),
+      sessionId: 's1',
+      append: rawAppend('s1'),
+    })
+
+    for (let i = 0; i < 4; i++) {
+      cadence.append(delta('more'))
+      cadence.flush()
+    }
+
+    const snapshot = store.getLatestSnapshot('s1')!.data
+    expect(snapshot.contextWindows).toHaveLength(1)
+    expect(snapshot.formatRetries).toHaveLength(1)
+  })
+
+  it('shrinks a snapshot whose history was already inflated', () => {
+    emitSessionInitialized('s1', 'p', '/tmp', 'win-1')
+    const store = getEventStore()
+    const cadence = createSnapshotCadence({
+      sessionManager: fakeSessionManager(),
+      sessionId: 's1',
+      append: rawAppend('s1'),
+    })
+    cadence.flush()
+    const inflated = store.getLatestSnapshot('s1')!.data
+    const compaction = {
+      closedWindowId: 'win-1',
+      newWindowId: 'win-2',
+      beforeTokens: 1,
+      afterTokens: 0,
+      summary: 's',
+      timestamp: 1,
+    }
+    store.append('s1', {
+      type: 'turn.snapshot',
+      data: { ...inflated, contextWindows: [compaction, compaction, compaction, compaction] },
+    })
+
+    cadence.append(delta('more'))
+    cadence.flush()
+
+    expect(store.getLatestSnapshot('s1')!.data.contextWindows).toHaveLength(1)
+  })
+
   it('flush always snapshots and prunes, even with an empty tail', () => {
     emitSessionInitialized('s1', 'p', '/tmp', 'win-1')
     emitUserMessage('s1', 'go')

src/server/chat/snapshot-cadence.ts

@@ -28,6 +28,12 @@ const DEFAULT_MAX_MS = 10 * 60 * 1000
 const MIN_BYTES_FOR_TIME_TRIGGER = 4 * 1024 * 1024
 /** Minimum delay between two tail measurements. */
 const DEFAULT_CHECK_INTERVAL_MS = 30_000
+/**
+ * Minimum delay between two checks while a snapshot is deferred. Each check
+ * scans the tail (~0.1 s at 46 MB), and parallel tools or sub-agents can keep
+ * the turn busy across many quiet points in a row.
+ */
+const DEFAULT_DEFERRED_CHECK_INTERVAL_MS = 5_000
 
 export interface SnapshotCadence {
   /** Append an event; a snapshot is taken when the tail budget is exceeded. */
@@ -44,6 +50,7 @@ export interface SnapshotCadenceOptions {
   maxBytes?: number
   maxMs?: number
   checkIntervalMs?: number
+  deferredCheckIntervalMs?: number
 }
 
 /**
@@ -62,6 +69,14 @@ interface CadenceState {
    */
   openMessagesAtRest: number
   pendingToolCallsAtRest: number
+  /**
+   * A check found the tail over budget but the turn busy. The periodic check
+   * nearly always lands on a busy moment in a sub-agent run (streaming or a
+   * tool round), so the next quiet point (a message or tool round ending)
+   * checks again, at most every DEFAULT_DEFERRED_CHECK_INTERVAL_MS, instead
+   * of waiting for the next period.
+   */
+  deferred: boolean
 }
 
 const cadenceStates = new Map<string, CadenceState>()
@@ -83,6 +98,7 @@ export function createSnapshotCadence(options: SnapshotCadenceOptions): Snapshot
   const maxMs =
     options.maxMs ?? (readPositiveIntEnv('OPENFOX_SNAPSHOT_MAX_MINUTES') ?? DEFAULT_MAX_MS / 60_000) * 60_000
   const checkIntervalMs = options.checkIntervalMs ?? DEFAULT_CHECK_INTERVAL_MS
+  const deferredCheckIntervalMs = options.deferredCheckIntervalMs ?? DEFAULT_DEFERRED_CHECK_INTERVAL_MS
 
   let state = cadenceStates.get(sessionId)
   if (!state) {
@@ -92,6 +108,7 @@ export function createSnapshotCadence(options: SnapshotCadenceOptions): Snapshot
       lastCheckAt: 0,
       openMessagesAtRest: 0,
       pendingToolCallsAtRest: 0,
+      deferred: false,
     }
     cadenceStates.set(sessionId, state)
     if (cadenceStates.size > MAX_TRACKED_SESSIONS) {
@@ -169,7 +186,10 @@ export function createSnapshotCadence(options: SnapshotCadenceOptions): Snapshot
       append(event)
 
       const now = Date.now()
-      if (now - state.lastCheckAt < checkIntervalMs) return
+      const quietPoint = event.type === 'message.done' || event.type === 'tool.result'
+      const interval =
+        state.deferred && quietPoint ? Math.min(deferredCheckIntervalMs, checkIntervalMs) : checkIntervalMs
+      if (now - state.lastCheckAt < interval) return
       state.lastCheckAt = now
 
       const { events, bytes, openMessages, pendingToolCalls } = eventStore.getEventLogTail(
@@ -178,21 +198,26 @@ export function createSnapshotCadence(options: SnapshotCadenceOptions): Snapshot
       )
       if (events === 0) return
 
+      const overBytes = bytes >= maxBytes
+      const overTime = now - state.lastSnapshotAt >= maxMs && bytes >= MIN_BYTES_FOR_TIME_TRIGGER
+      if (!overBytes && !overTime) {
+        state.deferred = false
+        return
+      }
+
       // Snapshot only at a quiescent boundary. A snapshot taken while a message
       // streams (or a tool round is in flight) freezes a partial message, and
       // the context fold then drops every later event carrying that messageId —
       // the rest of the answer, or the tool results of the round.
       if (openMessages > state.openMessagesAtRest || pendingToolCalls > state.pendingToolCallsAtRest) {
         logger.debug('Snapshot cadence deferred, turn in flight', { sessionId, openMessages, pendingToolCalls })
+        state.deferred = true
         return
       }
 
-      const overBytes = bytes >= maxBytes
-      const overTime = now - state.lastSnapshotAt >= maxMs && bytes >= MIN_BYTES_FOR_TIME_TRIGGER
-      if (overBytes || overTime) {
-        logger.info('Snapshot cadence triggered', { sessionId, bytes, events, overBytes, overTime })
-        takeSnapshot()
-      }
+      state.deferred = false
+      logger.info('Snapshot cadence triggered', { sessionId, bytes, events, overBytes, overTime })
+      takeSnapshot()
     },
     flush: () => takeSnapshot(),
   }

src/server/events/fold-state.ts

@@ -310,12 +310,24 @@ export function foldSessionState(
         // the raw store) still count when the snapshot is the only
         // surviving artifact. The post-snapshot events following this
         // snapshot continue to accumulate on top.
+        // The stream may also replay the snapshot's own history as synthetic
+        // events (combineEventsWithSnapshot), just before it: keep one entry
+        // per compaction or retry. Counted twice, a snapshot built from such a
+        // stream doubled them, and so did every snapshot after it (seen: 7
+        // compactions stored as 2.3 million entries, a 433 MB snapshot);
+        // deduplicating here also shrinks a snapshot already inflated that way.
         const snapData = event.data as SessionSnapshot
         if (Array.isArray(snapData.contextWindows) && snapData.contextWindows.length > 0) {
-          contextWindows = [...snapData.contextWindows, ...contextWindows]
+          contextWindows = uniqueBy(
+            [...snapData.contextWindows, ...contextWindows],
+            (c) => `${c.closedWindowId}>${c.newWindowId}@${c.timestamp}`,
+          )
         }
         if (Array.isArray(snapData.formatRetries) && snapData.formatRetries.length > 0) {
-          formatRetries = [...snapData.formatRetries, ...formatRetries]
+          formatRetries = uniqueBy(
+            [...snapData.formatRetries, ...formatRetries],
+            (r) => `${r.attempt}/${r.maxAttempts}@${r.timestamp}`,
+          )
         }
         break
       }
@@ -609,3 +621,14 @@ export function buildSnapshotFromSessionState(input: {
     ...(foldedState.waitingWorkflow !== undefined && { waitingWorkflow: foldedState.waitingWorkflow }),
   }
 }
+
+/** Keeps the first item of each key, in order. */
+function uniqueBy<T>(items: T[], key: (item: T) => string): T[] {
+  const seen = new Set<string>()
+  return items.filter((item) => {
+    const k = key(item)
+    if (seen.has(k)) return false
+    seen.add(k)
+    return true
+  })
+}

src/server/events/store.test.ts

@@ -1444,6 +1444,67 @@ describe('EventStore - Event Cleanup', () => {
       expect(tail.bytes).toBeGreaterThan(content.length)
       expect(tail.bytes).toBeGreaterThanOrEqual(Buffer.byteLength(content, 'utf8'))
     })
+
+    it('counts a message started twice as one open message', () => {
+      // A successful compaction starts its summary attempt's message again,
+      // with the summary as content, then closes it once.
+      store.append('session-1', { type: 'message.start', data: { messageId: 'a1', role: 'assistant', content: '' } })
+      store.append('session-1', {
+        type: 'message.start',
+        data: { messageId: 'a1', role: 'assistant', content: 'summary', isCompactionSummary: true },
+      })
+      store.append('session-1', { type: 'message.done', data: { messageId: 'a1' } })
+
+      expect(store.getEventLogTail('session-1', 0).openMessages).toBe(0)
+    })
+
+    it('does not let a done whose start is before the tail hide an open message', () => {
+      store.append('session-1', { type: 'message.start', data: { messageId: 'old', role: 'assistant', content: '' } })
+      const fromSeq = store.getLatestSeq('session-1')!
+      store.append('session-1', { type: 'message.done', data: { messageId: 'old' } })
+      store.append('session-1', { type: 'message.start', data: { messageId: 'new', role: 'assistant', content: '' } })
+
+      expect(store.getEventLogTail('session-1', fromSeq).openMessages).toBe(1)
+    })
+
+    it('keeps counting open messages and pending calls when a done or result has no id', () => {
+      // SQL `x NOT IN (..., NULL)` is never true: one id-less row in the
+      // subquery would read every count as 0 and let the cadence snapshot
+      // mid-message.
+      store.append('session-1', { type: 'message.start', data: { messageId: 'a1', role: 'assistant', content: '' } })
+      store.append('session-1', {
+        type: 'tool.call',
+        data: { messageId: 'a1', toolCall: { id: 'c1', name: 'read_file', arguments: {} } },
+      })
+      store.append('session-1', { type: 'message.done', data: {} } as never)
+      store.append('session-1', { type: 'tool.result', data: { messageId: 'a1' } } as never)
+
+      const tail = store.getEventLogTail('session-1', 0)
+      expect(tail.openMessages).toBe(1)
+      expect(tail.pendingToolCalls).toBe(1)
+    })
+
+    it('does not let a tool result whose call is before the tail hide a pending call', () => {
+      store.append('session-1', {
+        type: 'tool.call',
+        data: { messageId: 'a1', toolCall: { id: 'old', name: 'read_file', arguments: {} } },
+      })
+      const fromSeq = store.getLatestSeq('session-1')!
+      store.append('session-1', {
+        type: 'tool.result',
+        data: {
+          messageId: 'a1',
+          toolCallId: 'old',
+          result: { success: true, output: '', durationMs: 1, truncated: false },
+        },
+      })
+      store.append('session-1', {
+        type: 'tool.call',
+        data: { messageId: 'a1', toolCall: { id: 'new', name: 'read_file', arguments: {} } },
+      })
+
+      expect(store.getEventLogTail('session-1', fromSeq).pendingToolCalls).toBe(1)
+    })
   })
 
   describe('getContextWindowEvents', () => {