diff --git a/src/server/chat/snapshot-cadence.test.ts b/src/server/chat/snapshot-cadence.test.ts index 6e5c9e635..1fdc1c453 100644 --- a/src/server/chat/snapshot-cadence.test.ts +++ b/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') diff --git a/src/server/chat/snapshot-cadence.ts b/src/server/chat/snapshot-cadence.ts index cb25df400..15d4d82c9 100644 --- a/src/server/chat/snapshot-cadence.ts +++ b/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() @@ -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(), } diff --git a/src/server/events/fold-state.ts b/src/server/events/fold-state.ts index df780d442..28e07d86f 100644 --- a/src/server/events/fold-state.ts +++ b/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(items: T[], key: (item: T) => string): T[] { + const seen = new Set() + return items.filter((item) => { + const k = key(item) + if (seen.has(k)) return false + seen.add(k) + return true + }) +} diff --git a/src/server/events/store.test.ts b/src/server/events/store.test.ts index 94eb5fd64..aca2bd471 100644 --- a/src/server/events/store.test.ts +++ b/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', () => { diff --git a/src/server/events/store.ts b/src/server/events/store.ts index a1869b5d3..09a446477 100644 --- a/src/server/events/store.ts +++ b/src/server/events/store.ts @@ -626,13 +626,29 @@ export class EventStore { -- bytes, so a non-ASCII payload (accents, CJK, emoji) would -- understate the tail by up to 3x and overshoot the byte budget. COALESCE(SUM(LENGTH(CAST(payload AS BLOB))), 0) AS bytes, - COALESCE(SUM(CASE WHEN event_type = 'message.start' THEN 1 ELSE 0 END), 0) - - COALESCE(SUM(CASE WHEN event_type = 'message.done' THEN 1 ELSE 0 END), 0) AS openMessages, - COALESCE(SUM(CASE WHEN event_type = 'tool.call' THEN 1 ELSE 0 END), 0) - - COALESCE(SUM(CASE WHEN event_type = 'tool.result' THEN 1 ELSE 0 END), 0) AS pendingToolCalls - FROM events WHERE session_id = ? AND seq > ?`, + -- Ids started and not done in the tail. A set, not a count: + -- a successful compaction starts its summary attempt's + -- message a second time, then closes it once, and a done + -- whose start is before the tail must not hide another open + -- message. The NOT IN subqueries skip NULL ids: x NOT IN + -- (..., NULL) is never true and would read every count as 0. + -- Subqueries so they only read their own event types + -- (session/type/seq index), not every tail row. + (SELECT COUNT(DISTINCT json_extract(payload, '$.messageId')) FROM events + WHERE session_id = @sessionId AND event_type = 'message.start' AND seq > @fromSeq + AND json_extract(payload, '$.messageId') NOT IN ( + SELECT json_extract(payload, '$.messageId') FROM events + WHERE session_id = @sessionId AND event_type = 'message.done' AND seq > @fromSeq + AND json_extract(payload, '$.messageId') IS NOT NULL)) AS openMessages, + (SELECT COUNT(DISTINCT json_extract(payload, '$.toolCall.id')) FROM events + WHERE session_id = @sessionId AND event_type = 'tool.call' AND seq > @fromSeq + AND json_extract(payload, '$.toolCall.id') NOT IN ( + SELECT json_extract(payload, '$.toolCallId') FROM events + WHERE session_id = @sessionId AND event_type = 'tool.result' AND seq > @fromSeq + AND json_extract(payload, '$.toolCallId') IS NOT NULL)) AS pendingToolCalls + FROM events WHERE session_id = @sessionId AND seq > @fromSeq`, ) - .get(sessionId, fromSeq) as { + .get({ sessionId, fromSeq }) as { events: number bytes: number openMessages: number