From 2d43fc449c50c61c03c7ecf3317c6b0b495f371f Mon Sep 17 00:00:00 2001 From: mossaab Date: Thu, 1 Oct 2026 11:49:11 +0000 Subject: [PATCH 1/6] fix(events): keep the event log bounded during long turns MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A turn is a single runTopLevelAgentLoop call that can run for hours, so the end-of-turn snapshot — and the cleanup that runs right after it — never ran. The log grew without bound while every state load re-read it: the incident session reached 320k events / 909 MB of payload behind a 95 KB snapshot, and a single state load cost ~12 s of synchronous better-sqlite3 work several times a minute, pinning the main thread at 98% of a core for 9h30 and starving the HTTP/WS loop until the UI could no longer stop the session. Bounded writes: - snapshot on a byte (48 MB) / time (10 min) cadence during the turn, at the end of every workflow step and inside sub-agent loops, pruning what the snapshot absorbed. Snapshots only happen at a quiescent point (no message streaming, no tool call awaiting its result) — the context fold drops later events for a messageId already in the snapshot. - throttle streamed tool.preparing events to one per 250 ms per tool call; they re-send the accumulated argument JSON, which produced 146k events for 691 real calls. Bounded reads: - the context-window fold reads only the four event types that can change the window id (6.9 s -> 1.4 ms on the incident session) - the conversation build replays the latest snapshot instead of the raw log - the user-message count is a SQL COUNT instead of a full materialization - the snapshot builder folds the previous snapshot plus the tail Store: - add the covering index (session_id, event_type, seq) and drop the two redundant ones; the snapshot lookup walked 320k index entries per call (105 ms -> <1 ms) - stop evicting the parsed snapshot cache on non-snapshot appends Observability: - warn when a state load exceeds 1 s (throttled) or when the database exceeds 300 MB at startup; the incident produced no log line at all. --- docs/SESSION-DEBUGGING.md | 3 +- src/server/chat/agent-loop-pause.test.ts | 1 + src/server/chat/agent-loop.test.ts | 8 + src/server/chat/conversation-history.ts | 15 +- src/server/chat/orchestrator.test.ts | 10 + src/server/chat/orchestrator.ts | 59 ++--- src/server/chat/snapshot-cadence.test.ts | 231 ++++++++++++++++++ src/server/chat/snapshot-cadence.ts | 199 +++++++++++++++ src/server/chat/stream-pure.test.ts | 84 +++++-- src/server/chat/stream-pure.ts | 32 ++- src/server/chat/system-reminder.test.ts | 10 + src/server/db/events-schema.ts | 27 ++ src/server/db/index.ts | 11 +- src/server/db/migrations.test.ts | 24 ++ src/server/events/session.test.ts | 18 +- src/server/events/session.ts | 26 +- src/server/events/store.test.ts | 223 +++++++++++++++++ src/server/events/store.ts | 191 +++++++++++++-- src/server/queue/processor.test.ts | 1 + src/server/sub-agents/manager.ts | 12 +- src/server/utils/session-utils.ts | 20 +- src/server/workflows/executor-mode.test.ts | 4 + .../workflows/executor-resume-nudge.test.ts | 4 + src/server/workflows/executor-retry.test.ts | 4 + .../workflows/executor-subgroup.test.ts | 4 + src/server/workflows/executor.ts | 12 +- 26 files changed, 1097 insertions(+), 136 deletions(-) create mode 100644 src/server/chat/snapshot-cadence.test.ts create mode 100644 src/server/chat/snapshot-cadence.ts create mode 100644 src/server/db/events-schema.ts diff --git a/docs/SESSION-DEBUGGING.md b/docs/SESSION-DEBUGGING.md index 5579e5903..877b6685f 100644 --- a/docs/SESSION-DEBUGGING.md +++ b/docs/SESSION-DEBUGGING.md @@ -74,8 +74,7 @@ This is the single source of truth. All session state derives from events. **Indexes**: -- `idx_events_session_seq` - For fetching events by session + sequence -- `idx_events_session_type` - For filtering by event type +- `idx_events_session_type_seq` - Covering index `(session_id, event_type, seq)`; serves the snapshot lookup, the context-window fold and the cleanup DELETE. `(session_id, seq)` is also covered by the UNIQUE constraint's auto-index. ## Event Types diff --git a/src/server/chat/agent-loop-pause.test.ts b/src/server/chat/agent-loop-pause.test.ts index 41ae953ab..ed07ec5a6 100644 --- a/src/server/chat/agent-loop-pause.test.ts +++ b/src/server/chat/agent-loop-pause.test.ts @@ -111,6 +111,7 @@ describe('runTopLevelAgentLoop pause gate', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore diff --git a/src/server/chat/agent-loop.test.ts b/src/server/chat/agent-loop.test.ts index 60acef830..54b7eec20 100644 --- a/src/server/chat/agent-loop.test.ts +++ b/src/server/chat/agent-loop.test.ts @@ -78,6 +78,7 @@ describe('executeTools', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), } as unknown as EventStore // Mock the event store singleton @@ -318,6 +319,7 @@ describe('runTopLevelAgentLoop assembleRequest', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore @@ -417,6 +419,7 @@ describe('runTopLevelAgentLoop compaction', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore @@ -630,6 +633,7 @@ describe('maxTokens clamping', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore @@ -1999,6 +2003,7 @@ describe('runTopLevelAgentLoop live stats', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore @@ -2137,6 +2142,7 @@ describe('runTopLevelAgentLoop queue draining', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore @@ -2289,6 +2295,7 @@ describe('runTopLevelAgentLoop plugin message transforms', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore @@ -2408,6 +2415,7 @@ describe('runTopLevelAgentLoop night mode', () => { mockEventStore = { append: vi.fn(), getEvents: vi.fn().mockReturnValue([]), + getContextWindowEvents: vi.fn().mockReturnValue([]), getLatestSeq: vi.fn().mockReturnValue(0), cleanupOldEvents: vi.fn().mockReturnValue(0), } as unknown as EventStore diff --git a/src/server/chat/conversation-history.ts b/src/server/chat/conversation-history.ts index ff9916880..295f4ba4b 100644 --- a/src/server/chat/conversation-history.ts +++ b/src/server/chat/conversation-history.ts @@ -24,7 +24,7 @@ import { import type { RequestContextMessage } from './request-context.js' import { minimalMessagesToRequestContextMessages } from './request-context.js' import { buildContextMessagesFromEventHistory, foldContextState } from '../events/folding.js' -import { getEventStore } from '../events/index.js' +import { getEventStore, combineEventsWithSnapshot } from '../events/index.js' import { processContextImages, loadResolvedVisionModel } from '../context/image-processor.js' import { modelSupportsVision } from '../llm/profiles.js' import type { Attachment } from '../../shared/types.js' @@ -237,7 +237,18 @@ export async function processEventsForConversation( onEvent: (event: TurnEvent) => void, ): Promise { const eventStore = getEventStore() - const rawEvents = eventStore.getEvents(sessionId) + // Bounded read: the latest snapshot already carries the folded history, so + // replaying it plus the events emitted after it reconstructs the session. + // Reading the raw log here re-read and re-parsed every event of a long + // session on each LLM round (~5 s of main-thread work on the incident + // session). Passing the snapshot's real seq lets attachment enrichment + // persist back into its row instead of no-oping against a synthetic seq. + const snapshotEvent = eventStore.getLatestSnapshot(sessionId) + const rawEvents = snapshotEvent + ? combineEventsWithSnapshot(sessionId, snapshotEvent.data, eventStore.getEvents(sessionId, snapshotEvent.seq + 1), { + snapshotSeq: snapshotEvent.seq, + }) + : eventStore.getEvents(sessionId) const modelVision = modelSupportsVision(llmClient.getModel()) const visionModel = await loadResolvedVisionModel() const { events: processedEvents } = await processContextImages(rawEvents, { diff --git a/src/server/chat/orchestrator.test.ts b/src/server/chat/orchestrator.test.ts index 861871aad..c78b09819 100644 --- a/src/server/chat/orchestrator.test.ts +++ b/src/server/chat/orchestrator.test.ts @@ -27,6 +27,7 @@ const { vi.mock('../events/index.js', () => ({ getEventStore: getEventStoreMock, getContextMessages: getContextMessagesMock, + combineEventsWithSnapshot: (_sessionId: string, _snapshot: unknown, events: unknown[]) => events, getCurrentContextWindowId: getCurrentContextWindowIdMock, getCurrentWindowMessageOptions: vi.fn((sessionId: string) => { const id = getCurrentContextWindowIdMock(sessionId) @@ -199,6 +200,15 @@ function createEventStore() { }), cleanupOldEvents: vi.fn((_sessionId: string) => 0), getLatestSnapshot: vi.fn((_sessionId: string) => undefined), + getLatestSnapshotSeq: vi.fn((sessionId: string) => { + const events = eventsBySession.get(sessionId) ?? [] + return [...events].reverse().find((event) => event.type === 'turn.snapshot')?.seq ?? 0 + }), + getEventsSinceSnapshot: vi.fn((sessionId: string) => ({ + snapshot: undefined, + events: eventsBySession.get(sessionId) ?? [], + })), + getEventLogTail: vi.fn(() => ({ events: 0, bytes: 0 })), } } diff --git a/src/server/chat/orchestrator.ts b/src/server/chat/orchestrator.ts index 2dfa1fadf..f4bfde332 100644 --- a/src/server/chat/orchestrator.ts +++ b/src/server/chat/orchestrator.ts @@ -10,16 +10,16 @@ * This is the ONE place where events get appended to the store. */ -import type { MessageStats, StatsIdentity, ToolCall, ToolResult } from '../../shared/types.js' +import type { StatsIdentity, ToolCall, ToolResult } from '../../shared/types.js' import type { ServerMessage } from '../../shared/protocol.js' import type { LLMClientWithModel } from '../llm/client.js' -import type { SessionSnapshot } from '../events/types.js' +import type { TurnEvent } from '../events/types.js' import type { AgentDefinition } from '../agents/types.js' import { getEventStore, getCurrentContextWindowId, getCurrentWindowMessageOptions } from '../events/index.js' -import { buildSnapshotFromSessionState } from '../events/folding.js' import type { SessionManager } from '../session/index.js' import { getToolRegistryForAgent, PathAccessDeniedError } from '../tools/index.js' import { buildAgentReminder, buildAgentSmallReminder, buildTopLevelSystemPrompt } from './prompts.js' +import { createSnapshotCadence } from './snapshot-cadence.js' import { serverT } from '../i18n.js' import { TurnMetrics, @@ -178,7 +178,6 @@ function resolveStatsIdentity(options: OrchestratorOptions): StatsIdentity { export async function runChatTurn(options: OrchestratorOptions): Promise { const { sessionManager, sessionId } = options const eventStore = getEventStore() - const statsIdentity = resolveStatsIdentity(options) const session = sessionManager.requireSession(sessionId) const mode = session.mode @@ -189,7 +188,7 @@ export async function runChatTurn(options: OrchestratorOptions): Promise { sessionManager.setRunning(sessionId, true) // Create append closure — the only write path to EventStore from the loop - const append = (event: import('../events/types.js').TurnEvent) => { + const append = (event: TurnEvent) => { try { eventStore.append(sessionId, event) } catch { @@ -197,6 +196,13 @@ export async function runChatTurn(options: OrchestratorOptions): Promise { } } + // Snapshot on a cadence *during* the turn. A turn can run for hours, and the + // end-of-turn snapshot below only runs once it finally resolves — the + // incident left 909 MB of payload behind a stale 95 KB snapshot, so every + // state load re-read the log (~12 s of synchronous work, several times a + // minute) and starved the event loop. + const cadence = createSnapshotCadence({ sessionManager, sessionId, append }) + // Track metrics across the turn const turnMetrics = new TurnMetrics() @@ -204,16 +210,10 @@ export async function runChatTurn(options: OrchestratorOptions): Promise { // Generic: use session mode as the agent ID. Workflow-specific callbacks // (kickoff injection, step_done tracking) are handled by the workflow executor // which calls runAgentTurn directly — not through runChatTurn. - await runAgentTurn(options, turnMetrics, mode, append) - - // Create end-of-turn snapshot - const snapshot = buildSnapshot(sessionManager, sessionId, turnMetrics.buildStats(statsIdentity, mode)) - const snapshotEvent = eventStore.append(sessionId, { type: 'turn.snapshot', data: snapshot }) + await runAgentTurn(options, turnMetrics, mode, cadence.append) - const deletedCount = eventStore.cleanupOldEvents(sessionId) - if (deletedCount > 0) { - logger.debug('Cleaned up old events after snapshot', { sessionId, deletedCount, snapshotSeq: snapshotEvent.seq }) - } + // End-of-turn snapshot + prune. + cadence.flush() } catch (error) { if (error instanceof PathAccessDeniedError) { const errorMsgId = crypto.randomUUID() @@ -263,12 +263,9 @@ export async function runChatTurn(options: OrchestratorOptions): Promise { } if (error instanceof Error && error.message === 'Aborted') { - try { - const snapshot = buildSnapshot(sessionManager, sessionId, turnMetrics.buildStats(statsIdentity, mode)) - eventStore.append(sessionId, { type: 'turn.snapshot', data: snapshot }) - } catch { - // Session may have been deleted during abort — skip cleanup - } + // A snapshot without the prune left the log unbounded on aborts; the + // cadence does both, and swallows the "session deleted" case itself. + cadence.flush() return } @@ -519,25 +516,3 @@ export async function runAgentTurn( turnMetrics, ) } - -// ============================================================================ -// Shared Helpers -// ============================================================================ - -/** - * Build a snapshot of current session state. - */ -function buildSnapshot(sessionManager: SessionManager, sessionId: string, _lastStats?: MessageStats): SessionSnapshot { - const eventStore = getEventStore() - const session = sessionManager.requireSession(sessionId) - const events = eventStore.getEvents(sessionId) - const latestSeq = eventStore.getLatestSeq(sessionId) ?? 0 - const cachedPrompt = sessionManager.getCachedPrompt(sessionId) - - return buildSnapshotFromSessionState({ - session, - events, - latestSeq, - ...(cachedPrompt ? { cachedSystemPrompt: cachedPrompt.systemPrompt, dynamicContextHash: cachedPrompt.hash } : {}), - }) -} diff --git a/src/server/chat/snapshot-cadence.test.ts b/src/server/chat/snapshot-cadence.test.ts new file mode 100644 index 000000000..6e5c9e635 --- /dev/null +++ b/src/server/chat/snapshot-cadence.test.ts @@ -0,0 +1,231 @@ +/** + * Snapshot cadence tests. + * + * The incident: a turn that runs for hours never reaches its end-of-turn + * snapshot, so the event log grows without bound (909 MB of payload behind a + * 95 KB snapshot) while every state load re-reads all of it. The cadence keeps + * the tail — the only part a state load replays — bounded during the turn. + */ + +import { describe, it, expect, beforeEach, afterEach } from 'vitest' +import Database from 'better-sqlite3' +import { initEventStore, getEventStore } from '../events/store.js' +import { emitSessionInitialized, emitUserMessage } from '../events/session.js' +import type { TurnEvent } from '../events/types.js' +import type { SessionManager } from '../session/manager.js' +import { createSnapshotCadence } from './snapshot-cadence.js' + +let db: Database.Database + +beforeEach(() => { + db = new Database(':memory:') + db.exec(` + CREATE TABLE IF NOT EXISTS sessions ( + id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + workdir TEXT NOT NULL + ) + `) + initEventStore(db) +}) + +afterEach(() => { + db.close() +}) + +function fakeSessionManager(): SessionManager { + return { + requireSession: () => ({ + mode: 'planner', + phase: 'plan', + isRunning: true, + criteria: [], + executionState: null, + }), + getCachedPrompt: () => undefined, + } as unknown as SessionManager +} + +function rawAppend(sessionId: string) { + return (event: TurnEvent) => { + getEventStore().append(sessionId, event) + } +} + +function delta(content: string): TurnEvent { + return { type: 'message.delta', data: { messageId: 'm1', content } } +} + +describe('createSnapshotCadence', () => { + it('snapshots and prunes mid-turn once the tail exceeds the byte budget', () => { + emitSessionInitialized('s1', 'p', '/tmp', 'win-1') + emitUserMessage('s1', 'go') + const store = getEventStore() + + const cadence = createSnapshotCadence({ + sessionManager: fakeSessionManager(), + sessionId: 's1', + append: rawAppend('s1'), + maxBytes: 2_000, + checkIntervalMs: 0, + }) + + for (let i = 0; i < 40; i++) { + cadence.append(delta('x'.repeat(200))) + } + + const snapshotSeq = store.getLatestSnapshotSeq('s1') + expect(snapshotSeq).toBeGreaterThan(0) + // Everything the snapshot absorbed is gone: only the initialization, the + // snapshot itself and the short tail remain. + expect(store.getEventLogTail('s1', snapshotSeq).events).toBeLessThan(20) + }) + + it('defers the snapshot while a message is streaming', () => { + emitSessionInitialized('s1', 'p', '/tmp', 'win-1') + const store = getEventStore() + + const cadence = createSnapshotCadence({ + sessionManager: fakeSessionManager(), + sessionId: 's1', + append: rawAppend('s1'), + maxBytes: 500, + checkIntervalMs: 0, + }) + + cadence.append({ type: 'message.start', data: { messageId: 'a1', role: 'assistant' } }) + for (let i = 0; i < 10; i++) { + cadence.append(delta('x'.repeat(200))) + } + + // Over budget, but snapshotting here would freeze a partial message and the + // context fold would drop every later delta carrying that messageId. + expect(store.getLatestSnapshotSeq('s1')).toBe(0) + + cadence.append({ type: 'message.done', data: { messageId: 'a1' } }) + + expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(0) + }) + + it('defers the snapshot while a tool round is in flight', () => { + emitSessionInitialized('s1', 'p', '/tmp', 'win-1') + const store = getEventStore() + + const cadence = createSnapshotCadence({ + sessionManager: fakeSessionManager(), + sessionId: 's1', + append: rawAppend('s1'), + maxBytes: 500, + checkIntervalMs: 0, + }) + + cadence.append({ + type: 'tool.call', + data: { + messageId: 'a1', + toolCall: { id: 'tc-1', name: 'read_file', arguments: { path: '/tmp/a.ts' } }, + }, + }) + for (let i = 0; i < 10; i++) { + cadence.append(delta('x'.repeat(200))) + } + + // The result of this call would be dropped from the context if the snapshot + // absorbed the round before it lands. + expect(store.getLatestSnapshotSeq('s1')).toBe(0) + + cadence.append({ + type: 'tool.result', + data: { + messageId: 'a1', + toolCallId: 'tc-1', + result: { success: true, output: 'ok', durationMs: 5, truncated: false }, + }, + }) + + 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. + getEventStore().append('s1', { + type: 'tool.call', + data: { + messageId: 'crashed', + toolCall: { id: 'tc-leftover', name: 'run_command', arguments: { command: 'sleep 999' } }, + }, + }) + const store = getEventStore() + + const cadence = createSnapshotCadence({ + sessionManager: fakeSessionManager(), + sessionId: 's1', + append: rawAppend('s1'), + maxBytes: 500, + checkIntervalMs: 0, + }) + + for (let i = 0; i < 10; i++) { + cadence.append(delta('x'.repeat(200))) + } + + expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(0) + }) + + it('leaves the log alone while the tail stays under the budget', () => { + emitSessionInitialized('s1', 'p', '/tmp', 'win-1') + const store = getEventStore() + + const cadence = createSnapshotCadence({ + sessionManager: fakeSessionManager(), + sessionId: 's1', + append: rawAppend('s1'), + maxBytes: 1_000_000, + checkIntervalMs: 0, + }) + + for (let i = 0; i < 10; i++) { + cadence.append(delta('x')) + } + + expect(store.getLatestSnapshotSeq('s1')).toBe(0) + }) + + it('snapshots on a time trigger even under the byte budget', () => { + emitSessionInitialized('s1', 'p', '/tmp', 'win-1') + const store = getEventStore() + + const cadence = createSnapshotCadence({ + sessionManager: fakeSessionManager(), + sessionId: 's1', + append: rawAppend('s1'), + maxBytes: Number.MAX_SAFE_INTEGER, + maxMs: 0, + checkIntervalMs: 0, + }) + + for (let i = 0; i < 40; i++) { + cadence.append(delta('x'.repeat(150_000))) + } + + expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(0) + }) + + it('flush always snapshots and prunes, even with an empty tail', () => { + emitSessionInitialized('s1', 'p', '/tmp', 'win-1') + emitUserMessage('s1', 'go') + const store = getEventStore() + + const cadence = createSnapshotCadence({ + sessionManager: fakeSessionManager(), + sessionId: 's1', + append: rawAppend('s1'), + maxBytes: Number.MAX_SAFE_INTEGER, + }) + cadence.flush() + + expect(store.getLatestSnapshotSeq('s1')).toBeGreaterThan(1) + expect(store.getEvents('s1').map((event) => event.type)).toEqual(['session.initialized', 'turn.snapshot']) + }) +}) diff --git a/src/server/chat/snapshot-cadence.ts b/src/server/chat/snapshot-cadence.ts new file mode 100644 index 000000000..cb25df400 --- /dev/null +++ b/src/server/chat/snapshot-cadence.ts @@ -0,0 +1,199 @@ +/** + * Mid-turn snapshots + event-log retention. + * + * A turn is a single `runTopLevelAgentLoop` call that can run for hours. The + * end-of-turn snapshot — and the cleanup that runs right after it — therefore + * never happens while the turn is alive: the log grows without bound and every + * state load re-reads (and re-parses) everything appended since the last + * snapshot. On the incident session that reached 909 MB of payload behind a + * 95 KB snapshot, costing ~12 s of synchronous work per load, several times a + * minute, which starved the event loop. + * + * This cadence snapshots *during* the turn and prunes what the snapshot + * absorbed, so the tail — the only part a state load has to replay — stays + * bounded no matter how long the turn runs. + */ + +import type { SessionManager } from '../session/manager.js' +import type { TurnEvent } from '../events/types.js' +import { combineEventsWithSnapshot, getEventStore } from '../events/index.js' +import { buildSnapshotFromSessionState } from '../events/folding.js' +import { logger } from '../utils/logger.js' + +/** Payload bytes appended since the last snapshot that force a new one. */ +const DEFAULT_MAX_BYTES = 48 * 1024 * 1024 +/** Wall-clock ceiling between snapshots, so a slow trickle is bounded too. */ +const DEFAULT_MAX_MS = 10 * 60 * 1000 +/** A time-triggered snapshot still needs some content, or idle turns snapshot for nothing. */ +const MIN_BYTES_FOR_TIME_TRIGGER = 4 * 1024 * 1024 +/** Minimum delay between two tail measurements. */ +const DEFAULT_CHECK_INTERVAL_MS = 30_000 + +export interface SnapshotCadence { + /** Append an event; a snapshot is taken when the tail budget is exceeded. */ + append: (event: TurnEvent) => void + /** Snapshot + prune now (end of turn, or before the turn's state is final). */ + flush: () => void +} + +export interface SnapshotCadenceOptions { + sessionManager: SessionManager + sessionId: string + append: (event: TurnEvent) => void + /** Test seam — production uses the defaults. */ + maxBytes?: number + maxMs?: number + checkIntervalMs?: number +} + +/** + * Per-session cadence state, shared by every writer of that session's log (the + * turn, its workflow steps and its sub-agents): they all append to the same + * events table, so they must not each pay for their own snapshot. + */ +interface CadenceState { + lastSnapshotSeq: number + lastSnapshotAt: number + lastCheckAt: number + /** + * In-flight counters already present when this cadence started. A tool call + * interrupted by a crash never gets its result, and that leftover must not + * block every later snapshot of the session. + */ + openMessagesAtRest: number + pendingToolCallsAtRest: number +} + +const cadenceStates = new Map() +/** Entries are two numbers; the cap only exists so a long-lived server does not keep one per session forever. */ +const MAX_TRACKED_SESSIONS = 256 + +function readPositiveIntEnv(name: string): number | undefined { + const raw = process.env[name] + if (!raw) return undefined + const parsed = Number.parseInt(raw, 10) + return Number.isFinite(parsed) && parsed > 0 ? parsed : undefined +} + +export function createSnapshotCadence(options: SnapshotCadenceOptions): SnapshotCadence { + const { sessionManager, sessionId, append } = options + const eventStore = getEventStore() + + const maxBytes = options.maxBytes ?? readPositiveIntEnv('OPENFOX_SNAPSHOT_MAX_BYTES') ?? DEFAULT_MAX_BYTES + const maxMs = + options.maxMs ?? (readPositiveIntEnv('OPENFOX_SNAPSHOT_MAX_MINUTES') ?? DEFAULT_MAX_MS / 60_000) * 60_000 + const checkIntervalMs = options.checkIntervalMs ?? DEFAULT_CHECK_INTERVAL_MS + + let state = cadenceStates.get(sessionId) + if (!state) { + state = { + lastSnapshotSeq: 0, + lastSnapshotAt: 0, + lastCheckAt: 0, + openMessagesAtRest: 0, + pendingToolCallsAtRest: 0, + } + cadenceStates.set(sessionId, state) + if (cadenceStates.size > MAX_TRACKED_SESSIONS) { + const oldest = cadenceStates.keys().next().value + if (oldest !== undefined && oldest !== sessionId) cadenceStates.delete(oldest) + } + } + // Whoever enters last owns the freshest view of the log — a sub-agent's + // cadence must not measure the tail from before its parent's snapshot. + state.lastSnapshotSeq = eventStore.getLatestSnapshotSeq(sessionId) + state.lastSnapshotAt = Date.now() + state.lastCheckAt = Date.now() - checkIntervalMs + // Anything already in flight at this point predates this cadence (a parent + // turn's message, a tool the caller is waiting on) — only new opens block. + const atRest = eventStore.getEventLogTail(sessionId, state.lastSnapshotSeq) + state.openMessagesAtRest = atRest.openMessages + state.pendingToolCallsAtRest = atRest.pendingToolCalls + + let insideSnapshot = false + + /** + * Build the snapshot from the previous one plus the tail, never from the raw + * log: `buildSnapshotFromSessionState` folds whatever it is given, so feeding + * it the whole log would reintroduce the cost this module exists to remove. + */ + const takeSnapshot = (): void => { + if (insideSnapshot) return + insideSnapshot = true + try { + const session = sessionManager.requireSession(sessionId) + const { snapshot, events } = eventStore.getEventsSinceSnapshot(sessionId) + const boundedEvents = combineEventsWithSnapshot(sessionId, snapshot, events) + const cachedPrompt = sessionManager.getCachedPrompt(sessionId) + + const next = buildSnapshotFromSessionState({ + session, + events: boundedEvents, + latestSeq: eventStore.getLatestSeq(sessionId) ?? 0, + ...(cachedPrompt + ? { cachedSystemPrompt: cachedPrompt.systemPrompt, dynamicContextHash: cachedPrompt.hash } + : {}), + }) + + const appended = eventStore.append(sessionId, { type: 'turn.snapshot', data: next }) + const tailBefore = eventStore.getEventLogTail(sessionId, state.lastSnapshotSeq).bytes + state.lastSnapshotSeq = appended.seq + state.lastSnapshotAt = Date.now() + const afterSnapshot = eventStore.getEventLogTail(sessionId, state.lastSnapshotSeq) + state.openMessagesAtRest = afterSnapshot.openMessages + state.pendingToolCallsAtRest = afterSnapshot.pendingToolCalls + + const deleted = eventStore.cleanupOldEvents(sessionId) + if (deleted > 0) { + logger.info('Pruned event log behind a snapshot', { + sessionId, + snapshotSeq: appended.seq, + tailBytes: tailBefore, + deleted, + }) + } + } catch (error) { + // Session may have been deleted (e.g. during abort) — the caller's own + // bookkeeping must not fail because of a snapshot. + logger.debug('Snapshot failed', { + sessionId, + error: error instanceof Error ? error.message : String(error), + }) + } finally { + insideSnapshot = false + } + } + + return { + append: (event) => { + append(event) + + const now = Date.now() + if (now - state.lastCheckAt < checkIntervalMs) return + state.lastCheckAt = now + + const { events, bytes, openMessages, pendingToolCalls } = eventStore.getEventLogTail( + sessionId, + state.lastSnapshotSeq, + ) + if (events === 0) 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 }) + 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() + } + }, + flush: () => takeSnapshot(), + } +} diff --git a/src/server/chat/stream-pure.test.ts b/src/server/chat/stream-pure.test.ts index 9dd01834d..f0d741adf 100644 --- a/src/server/chat/stream-pure.test.ts +++ b/src/server/chat/stream-pure.test.ts @@ -105,6 +105,43 @@ describe('stream-pure', () => { }) }) + it('throttles streamed tool.preparing events while keeping the latest arguments', async () => { + // Streaming arguments re-emit the *accumulated* JSON on every chunk, which + // is quadratic in the argument size. Unthrottled, the incident session + // logged 146k tool.preparing events (up to 30 KB each) for 691 real calls. + const fragments = Array.from({ length: 200 }, (_, i) => `{"path":"src/a.ts","chunk":${i}}`) + const client = createMockClient([ + { type: 'tool_call_delta', index: 0, name: 'write_file' }, + ...fragments.map((argumentsJson) => ({ type: 'tool_call_delta' as const, index: 0, arguments: argumentsJson })), + { + type: 'done', + response: { + ...mockResponse, + toolCalls: [{ id: 'call-1', name: 'write_file', arguments: { path: 'src/a.ts' } }], + }, + }, + ]) + + const gen = streamLLMPure({ + messageId: 'msg-throttle', + systemPrompt: 'system', + llmClient: client, + messages: [{ role: 'user', content: 'hello' }], + tools: [{ type: 'function', function: { name: 'write_file', description: 'Write', parameters: {} } }], + }) + + const preparing: Array<{ type: string; data: { arguments?: string } }> = [] + await consumeStreamGenerator(gen, (event) => { + if (event.type === 'tool.preparing') preparing.push(event as never) + }) + + // The name-only event, then the first chunk carrying arguments: every later + // chunk inside the interval is coalesced. + expect(preparing).toHaveLength(2) + expect(preparing[0]!.data.arguments).toBeUndefined() + expect(preparing[1]!.data.arguments).toBe(fragments[0]) + }) + it('excludes omitted params from result.modelParams so stats reflect the wire request', async () => { const client = createMockClient([ { type: 'text_delta', content: 'hi' }, @@ -179,15 +216,18 @@ describe('stream-pure', () => { }) const events: Array<{ type: string; data: unknown }> = [] - await consumeStreamGenerator(gen, (event) => { + const result = await consumeStreamGenerator(gen, (event) => { events.push(event) }) const preparingEvents = events.filter((e) => e.type === 'tool.preparing') - expect(preparingEvents).toHaveLength(3) + // Name-only event, then the first chunk carrying arguments. Remaining chunks + // fall inside the coalescing interval (PREPARING_ARG_EMIT_INTERVAL_MS). + expect(preparingEvents).toHaveLength(2) expect(preparingEvents[0]!).toMatchObject({ data: { name: 'run_command' } }) expect(preparingEvents[1]!).toMatchObject({ data: { name: 'run_command', arguments: '{"command":"echo' } }) - expect(preparingEvents[2]!).toMatchObject({ data: { name: 'run_command', arguments: '{"command":"echo hello"}' } }) + // Coalescing never loses arguments: the complete set arrives with tool.call. + expect(result.toolCalls[0]!.arguments).toEqual({ command: 'echo hello' }) }) it('streams partial arguments for session_metadata', async () => { @@ -226,12 +266,12 @@ describe('stream-pure', () => { }) const events: Array<{ type: string; data: unknown }> = [] - await consumeStreamGenerator(gen, (event) => { + const result = await consumeStreamGenerator(gen, (event) => { events.push(event) }) const preparingEvents = events.filter((e) => e.type === 'tool.preparing') - expect(preparingEvents).toHaveLength(3) + expect(preparingEvents).toHaveLength(2) expect(preparingEvents[0]!).toMatchObject({ data: { name: 'session_metadata' } }) expect(preparingEvents[1]!).toMatchObject({ data: { @@ -239,11 +279,11 @@ describe('stream-pure', () => { arguments: '{"action":"add","key":"criteria","id":"criterion-1","description":"Implement', }, }) - expect(preparingEvents[2]!).toMatchObject({ - data: { - name: 'session_metadata', - arguments: '{"action":"add","key":"criteria","id":"criterion-1","description":"Implement the thing"}', - }, + expect(result.toolCalls[0]!.arguments).toEqual({ + action: 'add', + key: 'criteria', + id: 'criterion-1', + description: 'Implement the thing', }) }) @@ -280,22 +320,17 @@ describe('stream-pure', () => { }) const events: Array<{ type: string; data: unknown }> = [] - await consumeStreamGenerator(gen, (event) => { + const result = await consumeStreamGenerator(gen, (event) => { events.push(event) }) const preparingEvents = events.filter((e) => e.type === 'tool.preparing') - expect(preparingEvents).toHaveLength(4) + expect(preparingEvents).toHaveLength(2) expect(preparingEvents[0]!).toMatchObject({ data: { name: 'write_file' } }) expect(preparingEvents[1]!).toMatchObject({ data: { name: 'write_file', arguments: '{"path":"src/app.ts","content":"const x = 1' }, }) - expect(preparingEvents[2]!).toMatchObject({ - data: { name: 'write_file', arguments: '{"path":"src/app.ts","content":"const x = 1;\\n' }, - }) - expect(preparingEvents[3]!).toMatchObject({ - data: { name: 'write_file', arguments: '{"path":"src/app.ts","content":"const x = 1;\\nexport default x"}' }, - }) + expect(result.toolCalls[0]!.arguments).toEqual({ path: 'src/app.ts', content: 'const x = 1;\nexport default x' }) }) it('streams partial arguments for edit_file', async () => { @@ -334,12 +369,12 @@ describe('stream-pure', () => { }) const events: Array<{ type: string; data: unknown }> = [] - await consumeStreamGenerator(gen, (event) => { + const result = await consumeStreamGenerator(gen, (event) => { events.push(event) }) const preparingEvents = events.filter((e) => e.type === 'tool.preparing') - expect(preparingEvents).toHaveLength(3) + expect(preparingEvents).toHaveLength(2) expect(preparingEvents[0]!).toMatchObject({ data: { name: 'edit_file' } }) expect(preparingEvents[1]!).toMatchObject({ data: { @@ -347,11 +382,10 @@ describe('stream-pure', () => { arguments: '{"path":"src/app.ts","old_string":"const x = 1","new_string":"const x', }, }) - expect(preparingEvents[2]!).toMatchObject({ - data: { - name: 'edit_file', - arguments: '{"path":"src/app.ts","old_string":"const x = 1","new_string":"const x = 2"}', - }, + expect(result.toolCalls[0]!.arguments).toEqual({ + path: 'src/app.ts', + old_string: 'const x = 1', + new_string: 'const x = 2', }) }) diff --git a/src/server/chat/stream-pure.ts b/src/server/chat/stream-pure.ts index 34e2253bb..491b52039 100644 --- a/src/server/chat/stream-pure.ts +++ b/src/server/chat/stream-pure.ts @@ -241,6 +241,16 @@ function siblingReadsSamePath( // Pure Streaming Generator // ============================================================================ +/** + * Minimum delay between two argument-carrying `tool.preparing` events for the + * same tool call. Each such event re-sends the whole accumulated argument JSON, + * so emitting one per streamed chunk is quadratic in the argument size: the + * incident session logged 146k events (up to 30 KB each) for 691 real calls. + * The first argument chunk always goes out, so the live view starts instantly; + * the final arguments always arrive with `tool.call`. + */ +const PREPARING_ARG_EMIT_INTERVAL_MS = 250 + /** * Pure generator that streams a SINGLE LLM request and yields TurnEvents live. * @@ -331,7 +341,10 @@ export async function* streamLLMPure(options: PureStreamOptions): AsyncGenerator const returnValueArgs = new Map() // Track accumulated tool arguments by index (for streaming partial args) const toolArgs = new Map() - + // Last time an argument-carrying preparing event went out per index. Streamed + // arguments re-emit the accumulated JSON on every chunk (quadratic in the + // argument size), so chunks inside the interval are coalesced. + const lastPreparingArgEmitAt = new Map() let result: Awaited>['value'] = null let aborted = false let streamError: string | undefined @@ -425,11 +438,18 @@ export async function* streamLLMPure(options: PureStreamOptions): AsyncGenerator name === 'write_file' || name === 'edit_file' ) { - const accumulatedArgs = toolArgs.get(value.index) - if (accumulatedArgs) { - yield { - type: 'tool.preparing', - data: { messageId, index: value.index, name, arguments: accumulatedArgs }, + // Coalesce: one event per interval per tool call. The first chunk + // carrying arguments always goes out (lastEmitAt undefined). + const now = Date.now() + const lastEmitAt = lastPreparingArgEmitAt.get(value.index) + if (lastEmitAt === undefined || now - lastEmitAt >= PREPARING_ARG_EMIT_INTERVAL_MS) { + lastPreparingArgEmitAt.set(value.index, now) + const accumulatedArgs = toolArgs.get(value.index) + if (accumulatedArgs) { + yield { + type: 'tool.preparing', + data: { messageId, index: value.index, name, arguments: accumulatedArgs }, + } } } } diff --git a/src/server/chat/system-reminder.test.ts b/src/server/chat/system-reminder.test.ts index d96e90336..f5114d924 100644 --- a/src/server/chat/system-reminder.test.ts +++ b/src/server/chat/system-reminder.test.ts @@ -28,8 +28,18 @@ function createEventStore(initialEvents: any[] = []) { events.push(event) }), getEvents: vi.fn(() => events), + // Mirrors EventStore.getContextWindowEvents: only these four types feed + // the current window id. + getContextWindowEvents: vi.fn(() => + events.filter((event) => + ['session.initialized', 'turn.snapshot', 'context.state', 'context.compacted'].includes(event.type), + ), + ), getAllEvents: vi.fn(() => events), getLatestSnapshot: vi.fn().mockReturnValue(undefined), + getLatestSnapshotSeq: vi.fn(() => [...events].reverse().find((event) => event.type === 'turn.snapshot')?.seq ?? 0), + getEventsSinceSnapshot: vi.fn(() => ({ snapshot: undefined, events })), + getEventLogTail: vi.fn(() => ({ events: 0, bytes: 0 })), cleanupOldEvents: vi.fn(), getLatestSeq: vi.fn().mockReturnValue(0), deleteSession: vi.fn(), diff --git a/src/server/db/events-schema.ts b/src/server/db/events-schema.ts new file mode 100644 index 000000000..7d69081a9 --- /dev/null +++ b/src/server/db/events-schema.ts @@ -0,0 +1,27 @@ +/** + * Schema shared by the two places that bootstrap the `events` table: + * `runMigrations()` (server startup) and `EventStore.initSchema()` (any store, + * including tests and tools that open the database directly). + * + * The index set lives here so the two DDL copies cannot drift. + */ + +/** + * Indexes for `events`. + * + * The covering index serves the per-session, type-filtered reads: the snapshot + * lookup (`event_type = 'turn.snapshot' ORDER BY seq DESC LIMIT 1`), the + * context-window fold and the cleanup DELETE. Without `seq` in the index, + * SQLite walks the whole session backwards and fetches each row — measured at + * 105 ms per snapshot lookup on a 322k-event session, on every state load. + * + * The two dropping statements remove the superseded indexes on existing + * databases: `(session_id, seq)` is already covered by the UNIQUE constraint's + * auto-index, and `(session_id, event_type)` is a prefix of the covering index. + * Both only cost a b-tree write per appended event — the hottest write path. + */ +export const EVENTS_INDEX_DDL = ` + CREATE INDEX IF NOT EXISTS idx_events_session_type_seq ON events(session_id, event_type, seq); + DROP INDEX IF EXISTS idx_events_session_seq; + DROP INDEX IF EXISTS idx_events_session_type; +` diff --git a/src/server/db/index.ts b/src/server/db/index.ts index 55e2c42f3..61bdd5983 100644 --- a/src/server/db/index.ts +++ b/src/server/db/index.ts @@ -1,6 +1,7 @@ import Database from 'better-sqlite3' import type { Config } from '../config.js' import { logger } from '../utils/logger.js' +import { EVENTS_INDEX_DDL } from './events-schema.js' let db: Database.Database | null = null @@ -222,7 +223,7 @@ function runMigrations(db: Database.Database): void { // Create events table for EventStore (single source of truth) // Note: EventStore creates this table with its own schema in initSchema() - // We just ensure the index exists for the event_type column + // We just ensure the event indexes exist on databases it did not create db.exec(` CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -236,13 +237,7 @@ function runMigrations(db: Database.Database): void { ) `) - db.exec(` - CREATE INDEX IF NOT EXISTS idx_events_session_seq ON events(session_id, seq) - `) - - db.exec(` - CREATE INDEX IF NOT EXISTS idx_events_session_type ON events(session_id, event_type) - `) + db.exec(EVENTS_INDEX_DDL) // Migration: Add per-session provider/model columns if (!columnNames.includes('provider_id')) { diff --git a/src/server/db/migrations.test.ts b/src/server/db/migrations.test.ts index c0e7cd0d7..e11cbd05a 100644 --- a/src/server/db/migrations.test.ts +++ b/src/server/db/migrations.test.ts @@ -195,6 +195,30 @@ describe('db migrations', () => { db.close() }) + it('replaces the legacy event indexes with the covering index on upgrade', () => { + // The old schema ships idx_events_session_seq and idx_events_session_type. + // Neither can serve the snapshot lookup (`event_type = ? ORDER BY seq DESC + // LIMIT 1`), so the covering index replaces both. + createOldSchemaDatabase(dbPath) + + const config = loadConfig() + config.database.path = dbPath + initDatabase(config) + + const db = new Database(dbPath) + const indexNames = ( + db.prepare(`SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'events'`).all() as { + name: string + }[] + ).map((row) => row.name) + + expect(indexNames).toContain('idx_events_session_type_seq') + expect(indexNames).not.toContain('idx_events_session_seq') + expect(indexNames).not.toContain('idx_events_session_type') + + db.close() + }) + it('adds workspace column when neither worktree nor workspace exists', () => { // Create a fresh database (no worktree, no workspace — the initial // CREATE TABLE already has workdir, not worktree) diff --git a/src/server/events/session.test.ts b/src/server/events/session.test.ts index b0981fbcb..4843c2115 100644 --- a/src/server/events/session.test.ts +++ b/src/server/events/session.test.ts @@ -4,7 +4,7 @@ * Tests the event-sourced session API: emitting events and reading derived state. */ -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 './store.js' import { buildSnapshot } from './folding.js' @@ -105,6 +105,22 @@ describe('getCurrentContextWindowId', () => { expect(getCurrentContextWindowId('s1')).toBe('new-window') }) + + it('reads only the window-relevant events, never the whole log', () => { + initSession('s1', 'win-1') + emitUserMessage('s1', 'Hello', { contextWindowId: 'win-1' }) + emitContextCompacted('s1', 'win-1', 'win-2', 1000, 0, 'Summary') + + // This runs inside tool execution, several times per turn. Reading the full + // event log here cost ~6 s of synchronous work per call on the incident + // session (322k events) — the main cause of the pinned main thread. + const getEvents = vi.spyOn(getEventStore(), 'getEvents') + + expect(getCurrentContextWindowId('s1')).toBe('win-2') + expect(getEvents).not.toHaveBeenCalled() + + getEvents.mockRestore() + }) }) describe('getCurrentWindowMessages', () => { diff --git a/src/server/events/session.ts b/src/server/events/session.ts index d3daaa9bc..b4b15ac22 100644 --- a/src/server/events/session.ts +++ b/src/server/events/session.ts @@ -30,7 +30,7 @@ import type { MessageSegment, Attachment, } from '../../shared/types.js' -import type { SessionSnapshot, SnapshotMessage, ReadFileEntry } from './types.js' +import type { SessionSnapshot, SnapshotMessage, ReadFileEntry, StoredEvent } from './types.js' import { getEventStore } from './store.js' import { getRuntimeConfig } from '../runtime-config.js' import { @@ -45,13 +45,16 @@ import { export function combineEventsWithSnapshot( sessionId: string, - snapshot: import('./types.js').SessionSnapshot | undefined, - events: import('./types.js').StoredEvent[], -): import('./types.js').StoredEvent[] { + snapshot: SessionSnapshot | undefined, + events: StoredEvent[], + options?: { snapshotSeq?: number }, +): StoredEvent[] { if (!snapshot) return events - const snapshotEvent: import('./types.js').StoredEvent = { - seq: 0, + const snapshotEvent: StoredEvent = { + // Callers that persist back into the snapshot row (image enrichment) must + // pass its real seq; 0 marks a purely synthetic view with no row to update. + seq: options?.snapshotSeq ?? 0, timestamp: snapshot.snapshotAt, sessionId, type: 'turn.snapshot', @@ -77,7 +80,7 @@ export function combineEventsWithSnapshot( // `success` flag drives error classification. We dedupe by `toolCall.id` // against the post-snapshot event stream so a tool call that happens to // be replayed post-snapshot is not double-counted. - const reconstructed: import('./types.js').StoredEvent[] = [] + const reconstructed: StoredEvent[] = [] let syntheticSeq = -1 for (const compaction of snapshot.contextWindows ?? []) { @@ -175,7 +178,7 @@ export function combineEventsWithSnapshot( * the count is zero / unknown. */ export function getLegacyCompactionBaseline( - snapshot: import('./types.js').SessionSnapshot | undefined, + snapshot: SessionSnapshot | undefined, ): { legacyCompactionCount: number; compactionsDetailsAvailable: boolean } | null { if (!snapshot) return null const cw = snapshot.contextWindows @@ -318,10 +321,11 @@ export function getContextMessages(sessionId: string): ContextMessage[] { * Get current context window ID */ export function getCurrentContextWindowId(sessionId: string): string | undefined { + // Only four event types feed the window id. Reading the whole log to recover + // it cost ~6 s of synchronous work per call on a 322k-event session, and this + // runs inside tool execution (sub-agents, MCP config, every LLM round). const eventStore = getEventStore() - const events = eventStore.getEvents(sessionId) - - const contextResult = foldContextState(events, '') + const contextResult = foldContextState(eventStore.getContextWindowEvents(sessionId), '') return contextResult.currentContextWindowId || undefined } diff --git a/src/server/events/store.test.ts b/src/server/events/store.test.ts index e34dd6efe..5708794d2 100644 --- a/src/server/events/store.test.ts +++ b/src/server/events/store.test.ts @@ -11,6 +11,7 @@ import { mkdtempSync, rmSync, existsSync, statSync, writeFileSync, utimesSync } import { join } from 'node:path' import { tmpdir } from 'node:os' import { EventStore, initEventStore, getStaleRunningSessionIds } from './store.js' +import { foldContextState } from './folding.js' import { SETTINGS_KEYS } from '../db/settings.js' import type { TurnEvent, StoredEvent, SessionSnapshot } from './types.js' @@ -1323,6 +1324,199 @@ describe('EventStore - Event Cleanup', () => { }) }) + describe('countUserMessages', () => { + it('counts real user messages, ignoring system-generated ones', () => { + store.append('session-1', { type: 'message.start', data: { messageId: 'u1', role: 'user', content: 'hi' } }) + store.append('session-1', { + type: 'message.start', + data: { messageId: 'a1', role: 'assistant', content: 'hello' }, + }) + store.append('session-1', { + type: 'message.start', + data: { messageId: 's1', role: 'user', content: 'auto prompt', isSystemGenerated: true }, + }) + store.append('session-1', { + type: 'message.start', + data: { messageId: 'u2', role: 'user', content: 'again', isSystemGenerated: false }, + }) + store.append('session-1', { type: 'message.delta', data: { messageId: 'u1', content: 'x' } }) + + expect(store.countUserMessages('session-1')).toBe(2) + }) + + it('is scoped to its session', () => { + store.append('session-1', { type: 'message.start', data: { messageId: 'u1', role: 'user', content: 'hi' } }) + store.append('session-2', { type: 'message.start', data: { messageId: 'u2', role: 'user', content: 'yo' } }) + + expect(store.countUserMessages('session-1')).toBe(1) + }) + }) + + describe('getContextWindowEvents', () => { + const contextState = { + currentTokens: 10, + maxTokens: 200000, + compactionCount: 0, + dangerZone: false, + canCompact: false, + dynamicContextChanged: false, + } + const snapshotData = { + mode: 'planner' as const, + phase: 'plan' as const, + isRunning: false, + messages: [], + criteria: [], + metadataEntries: {}, + contextState, + currentContextWindowId: 'window-2', + todos: [], + readFiles: [], + snapshotSeq: 6, + snapshotAt: Date.now(), + } + + const seedEvents = (sessionId: string) => { + store.append(sessionId, { + type: 'session.initialized', + data: { projectId: 'p', workdir: '/tmp', contextWindowId: 'window-1' }, + }) + store.append(sessionId, { type: 'message.start', data: { messageId: 'm1', role: 'user', content: 'hi' } }) + store.append(sessionId, { type: 'context.state', data: contextState }) + store.append(sessionId, { + type: 'file.read', + data: { path: '/tmp/a.ts', tokenCount: 3, contextWindowId: 'window-1' }, + }) + store.append(sessionId, { + type: 'context.compacted', + data: { + closedWindowId: 'window-1', + newWindowId: 'window-2', + beforeTokens: 10, + afterTokens: 0, + summary: 'sum', + }, + }) + store.append(sessionId, { type: 'turn.snapshot', data: snapshotData }) + } + + it('folds to the same current window as the full event log', () => { + seedEvents('session-1') + + // The window id is a pure function of four event types. Reading (and + // JSON-parsing) the whole log to recover it is what pinned the main + // thread for ~6 s per call on a 322k-event session. + const filtered = store.getContextWindowEvents('session-1') + const full = store.getEvents('session-1') + + expect(foldContextState(filtered, '').currentContextWindowId).toBe( + foldContextState(full, '').currentContextWindowId, + ) + expect(filtered.map((event) => event.type)).toEqual([ + 'session.initialized', + 'context.state', + 'context.compacted', + 'turn.snapshot', + ]) + }) + + it('excludes tombstoned events', () => { + seedEvents('session-1') + + const compacted = store.getEvents('session-1').find((event) => event.type === 'context.compacted')! + store.tombstoneEvents('session-1', [compacted.seq]) + + const filtered = store.getContextWindowEvents('session-1') + expect(filtered.map((event) => event.type)).not.toContain('context.compacted') + // Tombstoned events must be invisible here exactly as they are in + // getEvents — the fold result has to match the remaining raw events. + const survivors = store.getEvents('session-1').filter((event) => event.seq !== compacted.seq) + expect(foldContextState(filtered, '').currentContextWindowId).toBe( + foldContextState(survivors, '').currentContextWindowId, + ) + }) + }) + + describe('events indexes', () => { + it('migrates a legacy database to the (session_id, event_type, seq) index', () => { + // Pre-migration schema: the snapshot lookup (type filter + ORDER BY seq + // DESC LIMIT 1) had no usable index, so SQLite walked every event of the + // session backwards and fetched each row (105 ms on a 322k-event session). + const legacy = new Database(':memory:') + legacy.exec(` + CREATE TABLE events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + session_id TEXT NOT NULL, + seq INTEGER NOT NULL, + timestamp INTEGER NOT NULL, + event_type TEXT NOT NULL, + payload TEXT NOT NULL, + UNIQUE(session_id, seq) + ) + `) + legacy.exec(`CREATE INDEX idx_events_session_seq ON events(session_id, seq)`) + legacy.exec(`CREATE INDEX idx_events_session_type ON events(session_id, event_type)`) + + const migrated = new EventStore(legacy) + + const indexNames = ( + legacy.prepare(`SELECT name FROM sqlite_master WHERE type = 'index' AND tbl_name = 'events'`).all() as { + name: string + }[] + ).map((row) => row.name) + + expect(indexNames).toContain('idx_events_session_type_seq') + // Redundant: (session_id, seq) is the UNIQUE constraint's auto-index and + // (session_id, event_type) is a prefix of the new covering index. Both + // only cost a write per appended event. + expect(indexNames).not.toContain('idx_events_session_seq') + expect(indexNames).not.toContain('idx_events_session_type') + // A migrated database must still resolve snapshots — the store queries + // name the covering index explicitly. + expect(migrated.getLatestSnapshotSeq('session-1')).toBe(0) + legacy.close() + }) + + it('serves the snapshot lookup from the covering index', () => { + store.append('session-1', { + type: 'message.delta', + data: { messageId: 'm1', content: 'x'.repeat(1000) }, + }) + store.append('session-1', { + type: 'turn.snapshot', + data: { + mode: 'planner', + phase: 'plan', + isRunning: false, + messages: [], + criteria: [], + metadataEntries: {}, + contextState: { + currentTokens: 0, + maxTokens: 200000, + compactionCount: 0, + dangerZone: false, + canCompact: false, + dynamicContextChanged: false, + }, + currentContextWindowId: 'window-1', + todos: [], + readFiles: [], + snapshotSeq: 2, + snapshotAt: Date.now(), + }, + }) + + const plan = db + .prepare( + `EXPLAIN QUERY PLAN SELECT * FROM events WHERE session_id = ? AND event_type = 'turn.snapshot' ORDER BY seq DESC LIMIT 1`, + ) + .all('session-1') as { detail: string }[] + + expect(plan.map((row) => row.detail).join(' ')).toContain('idx_events_session_type_seq') + }) + }) + describe('cleanup after snapshot', () => { it('should keep only snapshot and current window events after cleanup', () => { // Simulate a conversation with multiple turns @@ -1741,6 +1935,35 @@ describe('EventStore - Event Cleanup', () => { expect(second).not.toBe(first) }) + it('keeps the cached snapshot across non-snapshot appends', () => { + store.append('session-1', { type: 'turn.snapshot', data: snapshotData('planner', 1) }) + const first = store.getLatestSnapshot('session-1') + + // Streaming appends dominate a long turn (deltas, thinking, preparing) + // and cannot change the latest snapshot. Evicting it on every one of them + // makes the cache permanently cold: each state load then pays a full + // backward scan of the session to find the snapshot again. + store.append('session-1', { type: 'message.delta', data: { messageId: 'm1', content: 'chunk' } }) + store.append('session-1', { type: 'message.thinking', data: { messageId: 'm1', content: 'thought' } }) + store.appendBatch('session-1', [ + { type: 'message.delta', data: { messageId: 'm1', content: 'more' } }, + { type: 'message.delta', data: { messageId: 'm1', content: 'still more' } }, + ]) + + expect(store.getLatestSnapshot('session-1')).toBe(first) + }) + + it('drops the cached snapshot when its row is deleted', () => { + store.append('session-1', { type: 'turn.snapshot', data: snapshotData('planner', 1) }) + expect(store.getLatestSnapshot('session-1')).toBeDefined() + + // Deletion paths can remove the snapshot itself — the cache must not + // outlive the row, or every later read replays a deleted snapshot. + store.deleteEventsAfterSeq('session-1', 0) + + expect(store.getLatestSnapshot('session-1')).toBeUndefined() + }) + it('does not bake the first caller limit into the cached prompts', () => { const messages = Array.from({ length: 8 }, (_, i) => ({ id: `m${i}`, diff --git a/src/server/events/store.ts b/src/server/events/store.ts index 6d0d355f7..023a36d64 100644 --- a/src/server/events/store.ts +++ b/src/server/events/store.ts @@ -20,6 +20,7 @@ import type { TurnEvent, StoredEvent, SessionSnapshot, SnapshotMessage } from '. import { logger } from '../utils/logger.js' import { foldSessionState, buildSnapshot, trimSnapshotStreamingOutput } from './folding.js' import { SETTINGS_KEYS } from '../db/settings.js' +import { EVENTS_INDEX_DDL } from '../db/events-schema.js' // Rollback backups (.pre-de-dup.bak) are held for 10 days after creation, then // auto-pruned on the next migration invocation (startup auto-run or manual @@ -27,6 +28,18 @@ import { SETTINGS_KEYS } from '../db/settings.js' // 1GB+ copy is not left on disk forever. const SNAPSHOT_BACKUP_RETENTION_MS = 10 * 24 * 60 * 60 * 1000 +/** + * A state load slower than this blocks every HTTP request, WS push and the + * agent loop (better-sqlite3 is synchronous). The incident ran for 9 h at 98 % + * of a core with nothing in the log, because the only signal — a 12 s state + * load — was never reported. + */ +const SLOW_STATE_LOAD_WARN_MS = 1_000 +/** At most one slow-load warning per minute, process-wide. */ +const SLOW_LOAD_WARN_THROTTLE_MS = 60 * 1000 +/** Size at which a session database is worth warning about at startup. */ +const DB_SIZE_WARN_BYTES = 300 * 1024 * 1024 + /** * Detect a UNIQUE constraint violation (e.g. another openfox instance writing * to the same database concurrently). Used to retry appends with the next seq. @@ -161,6 +174,8 @@ export class EventStore { private snapshotCacheBytes = 0 private static readonly SNAPSHOT_CACHE_MAX_ENTRIES = 16 private static readonly SNAPSHOT_CACHE_MAX_BYTES = 128 * 1024 * 1024 + /** Process-wide throttle for the slow-state-load warning. */ + private static lastSlowLoadWarnAt = 0 // Recent user prompts per session (sidebar list). Computed once from the // snapshot + message.start events, then served from memory. Bounded — small // arrays, but a long-lived server must not accumulate one per listed @@ -194,15 +209,7 @@ export class EventStore { ) `) - this.db.exec(` - CREATE INDEX IF NOT EXISTS idx_events_session_seq - ON events(session_id, seq) - `) - - this.db.exec(` - CREATE INDEX IF NOT EXISTS idx_events_session_type - ON events(session_id, event_type) - `) + this.db.exec(EVENTS_INDEX_DDL) this.db.exec(` CREATE TABLE IF NOT EXISTS tombstones ( @@ -256,7 +263,10 @@ export class EventStore { } } - this.invalidateSessionCache(sessionId) + // Only a snapshot can change the latest snapshot. Evicting the parsed + // snapshot cache on every append keeps it permanently cold during a turn — + // deltas, thinking and tool.preparing events stream in continuously. + this.invalidateSessionCache(sessionId, { snapshot: event.type === 'turn.snapshot' }) const stored: StoredEvent = { seq, @@ -309,7 +319,7 @@ export class EventStore { }) transaction() - this.invalidateSessionCache(sessionId) + this.invalidateSessionCache(sessionId, { snapshot: events.some((event) => event.type === 'turn.snapshot') }) // Notify after transaction commits for (const stored of results) { @@ -346,7 +356,7 @@ export class EventStore { }) transaction() - this.invalidateSessionCache(sessionId) + this.invalidateSessionCache(sessionId, { snapshot: events.some((event) => event.type === 'turn.snapshot') }) const stored: StoredEvent[] = events.map((event) => ({ ...event, sessionId })) for (const event of stored) { @@ -450,7 +460,7 @@ export class EventStore { const row = this.db .prepare( - `SELECT * FROM events + `SELECT * FROM events INDEXED BY idx_events_session_type_seq WHERE session_id = ? AND event_type = 'turn.snapshot' ORDER BY seq DESC LIMIT 1`, ) @@ -485,11 +495,21 @@ export class EventStore { } } - private invalidateSessionCache(sessionId: string): void { - const entry = this.snapshotCache.get(sessionId) - if (entry) { - this.snapshotCacheBytes -= entry.bytes - this.snapshotCache.delete(sessionId) + /** + * Drop the in-memory caches for a session. + * + * `snapshot: false` keeps the parsed latest-snapshot entry: an append that is + * not a `turn.snapshot` cannot change which snapshot is latest. Deletion and + * cleanup paths must invalidate unconditionally — they can remove the row the + * cache entry was built from. + */ + private invalidateSessionCache(sessionId: string, opts: { snapshot?: boolean } = {}): void { + if (opts.snapshot !== false) { + const entry = this.snapshotCache.get(sessionId) + if (entry) { + this.snapshotCacheBytes -= entry.bytes + this.snapshotCache.delete(sessionId) + } } this.promptsCache.delete(sessionId) } @@ -581,11 +601,79 @@ export class EventStore { return prompts.slice(0, limit) } + /** + * Event count, payload bytes and in-flight counters for the events appended + * after `fromSeq`. + * + * The snapshot cadence measures this to decide when to snapshot: bounded by + * the last snapshot, so the check stays cheap as the log grows. Counting the + * whole log instead would make the check as expensive as the problem. + * + * `openMessages` (message.start without its message.done) and + * `pendingToolCalls` (tool.call without its tool.result) say whether the log + * is at a quiescent boundary: a snapshot taken mid-message freezes a partial + * message, and the context fold drops every later event carrying that + * messageId — including the tool results of the round in flight. + */ + getEventLogTail( + sessionId: string, + fromSeq: number, + ): { events: number; bytes: number; openMessages: number; pendingToolCalls: number } { + const row = this.db + .prepare( + `SELECT COUNT(*) AS events, + COALESCE(SUM(LENGTH(payload)), 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 > ?`, + ) + .get(sessionId, fromSeq) as { + events: number + bytes: number + openMessages: number + pendingToolCalls: number + } + + return row + } + + /** + * Count real (non system-generated) user messages in a session. + * + * Session naming only needs the count: materializing every event to count + * `message.start` events re-reads the whole log on a path that runs at + * session creation and on the sidebar. + */ + countUserMessages(sessionId: string): number { + const row = this.db + .prepare( + `SELECT COUNT(*) AS n FROM events + WHERE session_id = ? AND event_type = 'message.start' + AND json_extract(payload, '$.role') = 'user' + AND COALESCE(json_extract(payload, '$.isSystemGenerated'), 0) = 0`, + ) + .get(sessionId) as { n: number } + + return row.n + } + /** * Get the latest snapshot and all events since it * This is the primary method for loading a session efficiently */ getEventsSinceSnapshot(sessionId: string): { snapshot: SessionSnapshot | undefined; events: StoredEvent[] } { + const startedAt = Date.now() + const result = this.readEventsSinceSnapshot(sessionId) + this.warnIfSlowStateLoad(sessionId, Date.now() - startedAt) + return result + } + + private readEventsSinceSnapshot(sessionId: string): { + snapshot: SessionSnapshot | undefined + events: StoredEvent[] + } { const snapshotEvent = this.getLatestSnapshot(sessionId) if (!snapshotEvent) { @@ -605,6 +693,51 @@ export class EventStore { } } + /** + * The whole server is synchronous: a slow state load blocks every HTTP + * request, WS push and the agent loop. It used to be silent — the incident + * ran at 98 % CPU for 9 h with nothing in the log. Throttled process-wide so + * a degraded server cannot flood the log it is trying to report about. + */ + private warnIfSlowStateLoad(sessionId: string, elapsedMs: number): void { + if (elapsedMs < SLOW_STATE_LOAD_WARN_MS) return + const now = Date.now() + if (now - EventStore.lastSlowLoadWarnAt < SLOW_LOAD_WARN_THROTTLE_MS) return + EventStore.lastSlowLoadWarnAt = now + logger.warn('Slow session state load (blocks the main thread)', { + sessionId, + elapsedMs, + hint: 'event log is not being snapshotted/pruned', + }) + } + + /** + * Events that determine the current context window. + * + * `foldContextState` derives `currentContextWindowId` from exactly these four + * event types, so recovering it must not read (and JSON-parse) the whole log: + * on a 322k-event session the unbounded read cost ~6 s of synchronous work on + * the main thread, and it runs inside tool execution. + * + * `INDEXED BY` is required — the planner otherwise walks `(session_id, seq)` + * and filters row by row (measured: 109 ms vs 1.4 ms on the same database). + */ + getContextWindowEvents(sessionId: string): StoredEvent[] { + const rows = this.db + .prepare( + `SELECT e.* FROM events e INDEXED BY idx_events_session_type_seq + WHERE e.session_id = ? + AND e.event_type IN ('session.initialized', 'turn.snapshot', 'context.state', 'context.compacted') + AND NOT EXISTS ( + SELECT 1 FROM tombstones t WHERE t.session_id = e.session_id AND t.seq = e.seq + ) + ORDER BY e.seq`, + ) + .all(sessionId) as EventRow[] + + return rows.map((row) => this.rowToStoredEvent(row)) + } + /** * Get ALL events for a session, including synthetic events reconstructed * from the latest snapshot's messages. This provides a unified view of the @@ -892,7 +1025,7 @@ export class EventStore { const row = this.db .prepare( ` - SELECT seq FROM events + SELECT seq FROM events INDEXED BY idx_events_session_type_seq WHERE session_id = ? AND event_type = 'turn.snapshot' ORDER BY seq DESC LIMIT 1 `, @@ -1287,6 +1420,26 @@ export function initEventStore(db: Database.Database): EventStore { logger.info('Storage optimized', result) } + // Surface an oversized database at boot: it means pruning has not kept up + // with the writes, and state loads may block the main thread for seconds. + // Best-effort diagnostic — it must never block startup. + try { + const dbPath = db.name + if (dbPath && dbPath !== ':memory:') { + const { size } = statSync(dbPath) + if (size >= DB_SIZE_WARN_BYTES) { + logger.warn('Session database is large', { + path: dbPath, + sizeMB: Math.round(size / (1024 * 1024)), + freelistPages: db.pragma('freelist_count', { simple: true }), + hint: 'deleted events leave free pages — VACUUM reclaims them', + }) + } + } + } catch { + // Ignore: diagnostics only + } + // Snapshot stream de-dup, asynchronously (don't block startup). Automatic // and safe: the migration itself creates a rollback backup before its first // rewrite and is idempotent afterwards (settings-flag guarded). Only legacy diff --git a/src/server/queue/processor.test.ts b/src/server/queue/processor.test.ts index 9d6be1e78..61ae96c29 100644 --- a/src/server/queue/processor.test.ts +++ b/src/server/queue/processor.test.ts @@ -28,6 +28,7 @@ vi.mock('../events/store.js', () => ({ getEvents: vi.fn(() => []), getLatestSeq: vi.fn(() => undefined), getLatestSnapshot: vi.fn(() => undefined), + countUserMessages: vi.fn(() => 0), })), initEventStore: vi.fn(), })) diff --git a/src/server/sub-agents/manager.ts b/src/server/sub-agents/manager.ts index aeff83c7c..f252b554f 100644 --- a/src/server/sub-agents/manager.ts +++ b/src/server/sub-agents/manager.ts @@ -31,6 +31,7 @@ import { getEventStore, getCurrentContextWindowId } from '../events/index.js' import { createChatMessageMessage } from '../ws/protocol.js' import { logger } from '../utils/logger.js' import { getConversationMessages, processEventsForConversation } from '../chat/conversation-history.js' +import { createSnapshotCadence } from '../chat/snapshot-cadence.js' const RETURN_VALUE_INSTRUCTION = ` @@ -323,7 +324,16 @@ export async function executeSubAgent(options: SubAgentExecutionOptions): Promis loopResult = await runTopLevelAgentLoop( { mode: subAgentType, - append: (event) => eventStore.append(sessionId, event), + // Sub-agents append to the parent session's log directly. Without a + // cadence here a long sub-agent run would grow the log unbounded (the + // parent's own cadence only ticks on the parent's appends, and the + // parent is awaiting this sub-agent). The cadence state is shared per + // session, so this does not double the snapshot work. + append: createSnapshotCadence({ + sessionManager, + sessionId, + append: (event) => eventStore.append(sessionId, event), + }).append, sessionManager, sessionId, llmClient, diff --git a/src/server/utils/session-utils.ts b/src/server/utils/session-utils.ts index a4f3203aa..c7fa153d8 100644 --- a/src/server/utils/session-utils.ts +++ b/src/server/utils/session-utils.ts @@ -6,23 +6,9 @@ import { createContextStateMessage } from '../ws/protocol.js' import { getEventStore } from '../events/index.js' export function getSessionMessageCount(sessionId: string): number { - const eventStore = getEventStore() - const events = eventStore.getEvents(sessionId) - - let count = 0 - for (const event of events) { - if (event.type === 'message.start') { - const data = event.data as { role: string; isSystemGenerated?: boolean } - // System-generated messages (task reminders, auto prompts) are not - // "real" user input — they must not count as the first user message so - // auto session naming keys off the actual prompt. - if (data.role === 'user' && !data.isSystemGenerated) { - count++ - } - } - } - - return count + // Counted in SQL: the callers only need the number of real user messages, and + // materializing every event of the session to count them re-reads the log. + return getEventStore().countUserMessages(sessionId) } export function finalizeTurnCompletion( diff --git a/src/server/workflows/executor-mode.test.ts b/src/server/workflows/executor-mode.test.ts index c8b230d84..e29046f1e 100644 --- a/src/server/workflows/executor-mode.test.ts +++ b/src/server/workflows/executor-mode.test.ts @@ -15,9 +15,13 @@ vi.mock('../events/index.js', () => ({ getEventStore: () => ({ append: vi.fn(), getLatestSeq: vi.fn(() => 0), + getLatestSnapshotSeq: vi.fn(() => 0), + getEventsSinceSnapshot: vi.fn(() => ({ snapshot: undefined, events: [] })), + getEventLogTail: vi.fn(() => ({ events: 0, bytes: 0 })), getEvents: vi.fn(() => []), deleteEventsAfterSeq: vi.fn(), }), + combineEventsWithSnapshot: (_sessionId: string, _snapshot: unknown, events: unknown[]) => events, getCurrentContextWindowId: vi.fn(() => undefined), })) diff --git a/src/server/workflows/executor-resume-nudge.test.ts b/src/server/workflows/executor-resume-nudge.test.ts index a9feed2f2..0c82a5794 100644 --- a/src/server/workflows/executor-resume-nudge.test.ts +++ b/src/server/workflows/executor-resume-nudge.test.ts @@ -30,9 +30,13 @@ vi.mock('../events/index.js', () => ({ getEventStore: () => ({ append: mockAppend, getLatestSeq: vi.fn(() => 0), + getLatestSnapshotSeq: vi.fn(() => 0), + getEventsSinceSnapshot: vi.fn(() => ({ snapshot: undefined, events: [] })), + getEventLogTail: vi.fn(() => ({ events: 0, bytes: 0 })), getEvents: vi.fn(() => []), deleteEventsAfterSeq: vi.fn(), }), + combineEventsWithSnapshot: (_sessionId: string, _snapshot: unknown, events: unknown[]) => events, getCurrentContextWindowId: vi.fn(() => undefined), })) diff --git a/src/server/workflows/executor-retry.test.ts b/src/server/workflows/executor-retry.test.ts index 01fc2bd01..de0a7a060 100644 --- a/src/server/workflows/executor-retry.test.ts +++ b/src/server/workflows/executor-retry.test.ts @@ -31,7 +31,11 @@ vi.mock('../events/index.js', () => ({ getEventStore: () => ({ append: mockAppend, getLatestSeq: vi.fn(() => 0), + getLatestSnapshotSeq: vi.fn(() => 0), + getEventsSinceSnapshot: vi.fn(() => ({ snapshot: undefined, events: [] })), + getEventLogTail: vi.fn(() => ({ events: 0, bytes: 0 })), }), + combineEventsWithSnapshot: (_sessionId: string, _snapshot: unknown, events: unknown[]) => events, getCurrentContextWindowId: vi.fn(() => undefined), })) diff --git a/src/server/workflows/executor-subgroup.test.ts b/src/server/workflows/executor-subgroup.test.ts index 9a2ccd57f..6acd4c534 100644 --- a/src/server/workflows/executor-subgroup.test.ts +++ b/src/server/workflows/executor-subgroup.test.ts @@ -16,9 +16,13 @@ vi.mock('../events/index.js', () => ({ getEventStore: () => ({ append: vi.fn(), getLatestSeq: vi.fn(() => 0), + getLatestSnapshotSeq: vi.fn(() => 0), + getEventsSinceSnapshot: vi.fn(() => ({ snapshot: undefined, events: [] })), + getEventLogTail: vi.fn(() => ({ events: 0, bytes: 0 })), getEvents: vi.fn(() => []), deleteEventsAfterSeq: vi.fn(), }), + combineEventsWithSnapshot: (_sessionId: string, _snapshot: unknown, events: unknown[]) => events, getCurrentContextWindowId: vi.fn(() => undefined), })) diff --git a/src/server/workflows/executor.ts b/src/server/workflows/executor.ts index d6f9434f9..ba22727b6 100644 --- a/src/server/workflows/executor.ts +++ b/src/server/workflows/executor.ts @@ -25,6 +25,8 @@ import { TERMINAL_DONE, TERMINAL_BLOCKED } from './types.js' import { getEventStore, getCurrentContextWindowId } from '../events/index.js' import { createChatMessageMessage } from '../ws/protocol.js' import { runAgentTurn, TurnMetrics, createMessageStartEvent } from '../chat/orchestrator.js' +import { createSnapshotCadence } from '../chat/snapshot-cadence.js' +import type { TurnEvent } from '../events/types.js' import { loadAllAgentsDefault, resolveDefaultAgentId } from '../agents/registry.js' import { computeSessionStats } from '../../shared/stats.js' import { formatGitDiffFiles } from '../git/diff.js' @@ -621,7 +623,10 @@ export async function executeWorkflow( const turnMetrics = new TurnMetrics() const es = getEventStore() - const append = (event: import('../events/types.js').TurnEvent) => es.append(sessionId, event) + const append = (event: TurnEvent) => es.append(sessionId, event) + // Workflow steps call runAgentTurn directly, so without this a long step + // would never snapshot nor prune (the chat path's cadence never runs). + const cadence = createSnapshotCadence({ sessionManager, sessionId, append }) let stepDoneCalled = false @@ -669,7 +674,7 @@ export async function executeWorkflow( }, turnMetrics, stepAgentId, - append, + cadence.append, { ...(!firstEntryForStep.has(step.id) && !agentStep.prompt && !isResumingCurrentStep ? { injectKickoff: () => injectGenericKickoff(sessionId) } @@ -704,6 +709,9 @@ export async function executeWorkflow( return blockOnLLMFailure(agentResult.failed.error) } + // Snapshot + prune at the end of the step, mirroring runChatTurn. + cadence.flush() + firstEntryForStep.add(step.id) // After the first resumed turn completes, mark resume as consumed so // subsequent iterations of this step get the nudge if step_done wasn't called. From a662baeda99729c34a0eebe83240c0c62ea59e98 Mon Sep 17 00:00:00 2001 From: mossaab Date: Thu, 1 Oct 2026 13:42:25 +0000 Subject: [PATCH 2/6] fix(events): address review findings on the snapshot cadence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three issues raised in review of #401, all verified against the code. Byte accounting: LENGTH() on a TEXT value counts characters, not bytes, so a non-ASCII payload (accents, CJK, emoji) understated the un-snapshotted tail by up to 3x — measured 1.15x for accented French, 3.00x for CJK — and let the 48 MB cadence budget overshoot before a snapshot was taken. Measure the BLOB representation instead. User-message count: real user messages live in the latest snapshot as well as in the raw rows, because cleanupOldEvents() does not whitelist message.start (unlike mode.changed or context.state) and therefore deletes the rows once a snapshot absorbs them. Counting only the raw rows collapsed the count to ~0 after every prune, re-arming the "is this the first user message?" condition that drives session naming. Count the snapshot's messages plus the rows after its seq, and restore the tombstone exclusion the previous getEvents()-based implementation had. Mid-turn tool results: a mid-turn snapshot can capture a tool call whose result has not been emitted yet — a sub-agent's cadence fires while its parent turn is still awaiting that call (the parent's in-flight call became the nested cadence's baseline, so it does not defer the snapshot), and cleanupOldEvents() then deletes the parent's tool.call row. buildContextMessagesFromEventHistory() dropped the parent's later tool.result because its messageId was already in the snapshot, which left the call unfulfilled; stripOrphanedToolCalls() then removed the call from the message entirely, silently losing the sub-agent's return value from the parent's context. Keep an event that completes the snapshot rather than duplicating it. Regression tests added for all three. --- src/server/chat/conversation-history.test.ts | 67 ++++++++++++++ src/server/events/fold-messages.ts | 31 +++++-- src/server/events/store.test.ts | 94 ++++++++++++++++++++ src/server/events/store.ts | 33 +++++-- 4 files changed, 213 insertions(+), 12 deletions(-) diff --git a/src/server/chat/conversation-history.test.ts b/src/server/chat/conversation-history.test.ts index 8388ff977..2055e4177 100644 --- a/src/server/chat/conversation-history.test.ts +++ b/src/server/chat/conversation-history.test.ts @@ -242,6 +242,73 @@ describe('buildContextMessages', () => { expect(result[1]!.toolCallId).toBe('call-1') }) + it('keeps a tool result that completes a tool call the snapshot left unresolved', () => { + // A sub-agent's cadence can snapshot mid-parent-turn: the parent's tool + // call is still awaiting its result, so the snapshot embeds it without + // one, and cleanupOldEvents() then deletes the parent's raw `tool.call`. + // Dropping the parent's later `tool.result` would leave the call + // unfulfilled, and stripOrphanedToolCalls would remove the call from the + // message entirely — losing the sub-agent's return value from context. + const snapshotSeq = 10 + const events: StoredEvent[] = [ + makeEvent({ + seq: 1, + type: 'session.initialized', + data: { projectId: 'p1', workdir: '/tmp', contextWindowId: 'window-1' }, + }), + makeEvent({ + seq: snapshotSeq, + type: 'turn.snapshot', + data: { + mode: 'planner', + phase: 'plan', + isRunning: true, + messages: [ + { + id: 'm1', + role: 'assistant', + content: 'delegating to a sub-agent', + timestamp: 5, + contextWindowId: 'window-1', + toolCalls: [{ id: 'call-1', name: 'task', arguments: { prompt: 'do it' } }], + }, + ], + criteria: [], + metadataEntries: {}, + contextState: { + currentTokens: 0, + maxTokens: 200000, + compactionCount: 0, + dangerZone: false, + canCompact: false, + dynamicContextChanged: false, + }, + currentContextWindowId: 'window-1', + todos: [], + snapshotSeq, + snapshotAt: 5, + }, + }), + makeEvent({ + seq: snapshotSeq + 1, + type: 'tool.result', + data: { + messageId: 'm1', + toolCallId: 'call-1', + result: { success: true, output: 'sub-agent done', durationMs: 10, truncated: false }, + }, + }), + ] + + const result = buildContextMessagesFromEventHistory(events, 'window-1', { includeVerifier: false }) + + const assistant = result.find((message) => message.role === 'assistant' && message.toolCalls?.length) + expect(assistant?.toolCalls?.[0]?.id).toBe('call-1') + const toolMessage = result.find((message) => message.role === 'tool') + expect(toolMessage?.toolCallId).toBe('call-1') + expect(toolMessage?.content).toContain('sub-agent done') + }) + it('produces the same output as buildContextMessagesFromEventHistory for top-level scope', () => { const events: StoredEvent[] = [ makeEvent({ diff --git a/src/server/events/fold-messages.ts b/src/server/events/fold-messages.ts index ed7a4adb2..0bc17af3a 100644 --- a/src/server/events/fold-messages.ts +++ b/src/server/events/fold-messages.ts @@ -480,12 +480,33 @@ export function buildContextMessagesFromEventHistory( // abort-snapshot) are dropped, exactly as the pre-unification fold did — the // snapshot content stays authoritative and synthetic events can never be // double-appended by a later delta/thinking/tool.result. + // + // Exception: an event that *completes* the snapshot rather than duplicating it + // must be kept. A mid-turn snapshot can capture a tool call whose result has + // not been emitted yet — a sub-agent's cadence fires while its parent turn is + // still awaiting that call (the parent's in-flight call became the nested + // cadence's baseline, so it does not defer the snapshot), and + // `cleanupOldEvents()` then deletes the parent's `tool.call` row. Dropping the + // parent's later `tool.result` would leave the snapshot's tool call + // unfulfilled, and `stripOrphanedToolCalls` would remove the call from the + // message entirely — silently losing the sub-agent's return value from the + // parent's context. const snapshotMessageIds = new Set(snapshot.messages.map((message) => message.id)) - const laterEvents = events.filter( - (event) => - event.seq > snapshotEvent.seq && - !('messageId' in event.data && snapshotMessageIds.has((event.data as { messageId: string }).messageId)), - ) + const unresolvedToolCallIds = new Set() + for (const message of snapshot.messages) { + for (const toolCall of message.toolCalls ?? []) { + if (toolCall.result === undefined) unresolvedToolCallIds.add(toolCall.id) + } + } + const laterEvents = events.filter((event) => { + if (event.seq <= snapshotEvent.seq) return false + if (!('messageId' in event.data)) return true + if (!snapshotMessageIds.has((event.data as { messageId: string }).messageId)) return true + // The snapshot already holds this message. Keep only what completes it. + if (event.type !== 'tool.result') return false + const toolCallId = (event.data as { toolCallId?: string }).toolCallId + return toolCallId !== undefined && unresolvedToolCallIds.has(toolCallId) + }) return buildContextMessagesFromStoredEvents( [...snapshotMessagesToEvents(snapshot.messages, snapshotEvent.sessionId), ...laterEvents], diff --git a/src/server/events/store.test.ts b/src/server/events/store.test.ts index 5708794d2..94eb5fd64 100644 --- a/src/server/events/store.test.ts +++ b/src/server/events/store.test.ts @@ -1350,6 +1350,100 @@ describe('EventStore - Event Cleanup', () => { expect(store.countUserMessages('session-1')).toBe(1) }) + + it('counts user messages folded into the latest snapshot, whose raw rows were pruned', () => { + // Realistic seq layout: session.initialized takes seq 1, which + // cleanupOldEvents() deliberately preserves. + store.append('session-1', { + type: 'session.initialized', + data: { projectId: 'p', workdir: '/tmp', contextWindowId: 'window-1' }, + }) + store.append('session-1', { type: 'message.start', data: { messageId: 'u1', role: 'user', content: 'first' } }) + store.append('session-1', { type: 'message.start', data: { messageId: 'u2', role: 'user', content: 'second' } }) + store.append('session-1', { + type: 'turn.snapshot', + data: { + mode: 'planner', + phase: 'plan', + isRunning: false, + messages: [ + { id: 'u1', role: 'user', content: 'first', timestamp: 1 }, + { id: 'u2', role: 'user', content: 'second', timestamp: 2 }, + ], + criteria: [], + metadataEntries: {}, + contextState: { + currentTokens: 0, + maxTokens: 200000, + compactionCount: 0, + dangerZone: false, + canCompact: false, + dynamicContextChanged: false, + }, + currentContextWindowId: 'window-1', + todos: [], + readFiles: [], + snapshotSeq: 3, + snapshotAt: Date.now(), + }, + }) + expect(store.cleanupOldEvents('session-1')).toBeGreaterThan(0) + + // Counting only the raw rows now reports 0, which re-arms the + // "is this the first user message?" condition driving session naming. + expect(store.countUserMessages('session-1')).toBe(2) + }) + + it('still counts user messages appended after the snapshot', () => { + store.append('session-1', { + type: 'session.initialized', + data: { projectId: 'p', workdir: '/tmp', contextWindowId: 'window-1' }, + }) + store.append('session-1', { type: 'message.start', data: { messageId: 'u1', role: 'user', content: 'first' } }) + store.append('session-1', { + type: 'turn.snapshot', + data: { + mode: 'planner', + phase: 'plan', + isRunning: false, + messages: [{ id: 'u1', role: 'user', content: 'first', timestamp: 1 }], + criteria: [], + metadataEntries: {}, + contextState: { + currentTokens: 0, + maxTokens: 200000, + compactionCount: 0, + dangerZone: false, + canCompact: false, + dynamicContextChanged: false, + }, + currentContextWindowId: 'window-1', + todos: [], + readFiles: [], + snapshotSeq: 2, + snapshotAt: Date.now(), + }, + }) + store.cleanupOldEvents('session-1') + store.append('session-1', { type: 'message.start', data: { messageId: 'u2', role: 'user', content: 'second' } }) + + expect(store.countUserMessages('session-1')).toBe(2) + }) + }) + + describe('getEventLogTail', () => { + it('measures UTF-8 bytes, not characters, for the snapshot byte budget', () => { + // SQLite's LENGTH() on a TEXT value counts characters, so a non-ASCII + // payload (accents, CJK, emoji) would understate the tail by up to 3x + // and let the 48 MB budget overshoot before a snapshot is taken. + const content = '修复事件日志无限增长的问题 🦊 déjà réglé' + store.append('session-1', { type: 'message.delta', data: { messageId: 'm1', content } }) + + const tail = store.getEventLogTail('session-1', 0) + + expect(tail.bytes).toBeGreaterThan(content.length) + expect(tail.bytes).toBeGreaterThanOrEqual(Buffer.byteLength(content, 'utf8')) + }) }) describe('getContextWindowEvents', () => { diff --git a/src/server/events/store.ts b/src/server/events/store.ts index 023a36d64..a1869b5d3 100644 --- a/src/server/events/store.ts +++ b/src/server/events/store.ts @@ -622,7 +622,10 @@ export class EventStore { const row = this.db .prepare( `SELECT COUNT(*) AS events, - COALESCE(SUM(LENGTH(payload)), 0) AS bytes, + -- CAST to BLOB: LENGTH() on a TEXT value counts characters, not + -- 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) @@ -642,21 +645,37 @@ export class EventStore { /** * Count real (non system-generated) user messages in a session. * + * Real user messages live in two places: the ones already folded into the + * latest snapshot, and the `message.start` rows appended since. Their raw rows + * are deleted by `cleanupOldEvents()` (which does not whitelist them, unlike + * `mode.changed` or `context.state`), so counting only the raw rows makes the + * count collapse to ~0 after every prune — which re-arms the "is this the + * first user message?" condition that drives session naming. + * * Session naming only needs the count: materializing every event to count * `message.start` events re-reads the whole log on a path that runs at * session creation and on the sidebar. */ countUserMessages(sessionId: string): number { + const snapshot = this.getLatestSnapshot(sessionId) + let count = 0 + for (const message of snapshot?.data.messages ?? []) { + if (message.role === 'user' && message.isSystemGenerated !== true) count++ + } + const row = this.db .prepare( - `SELECT COUNT(*) AS n FROM events - WHERE session_id = ? AND event_type = 'message.start' - AND json_extract(payload, '$.role') = 'user' - AND COALESCE(json_extract(payload, '$.isSystemGenerated'), 0) = 0`, + `SELECT COUNT(*) AS n FROM events e + WHERE e.session_id = ? AND e.seq > ? AND e.event_type = 'message.start' + AND json_extract(e.payload, '$.role') = 'user' + AND COALESCE(json_extract(e.payload, '$.isSystemGenerated'), 0) = 0 + AND NOT EXISTS ( + SELECT 1 FROM tombstones t WHERE t.session_id = e.session_id AND t.seq = e.seq + )`, ) - .get(sessionId) as { n: number } + .get(sessionId, snapshot?.seq ?? 0) as { n: number } - return row.n + return count + row.n } /** From f2ab29bc9afee5643d8c1e2d7f44f2ba7c0d2ae2 Mon Sep 17 00:00:00 2001 From: vv7r Date: Sun, 4 Oct 2026 17:46:33 +0000 Subject: [PATCH 3/6] fix(events): keep one history entry per compaction across snapshots A snapshot is built from the previous one plus the events since (combineEventsWithSnapshot). That stream also replays the previous snapshot's compactions and format retries as synthetic events, just before it, for the consumers that count them; the fold then seeded its history from the snapshot on top of them, so each snapshot doubled the history and every later one doubled it again. With the mid-turn cadence snapshotting often, it exploded: seen live, 7 compactions stored as 2.3 million entries in a 433 MB snapshot, rewritten at every turn end (8 s per write, a prompt took that long to show) until the disk was full. The fold keeps one entry per compaction or retry (same windows or attempt, same timestamp) when it seeds from a snapshot. This also shrinks a snapshot already inflated that way on the next snapshot. Co-Authored-By: Claude Opus 5.5 --- src/server/chat/snapshot-cadence.test.ts | 61 ++++++++++++++++++++++++ src/server/events/fold-state.ts | 27 ++++++++++- 2 files changed, 86 insertions(+), 2 deletions(-) diff --git a/src/server/chat/snapshot-cadence.test.ts b/src/server/chat/snapshot-cadence.test.ts index 6e5c9e635..153907fe7 100644 --- a/src/server/chat/snapshot-cadence.test.ts +++ b/src/server/chat/snapshot-cadence.test.ts @@ -212,6 +212,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/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 + }) +} From fe06e8aeb4cbf7b7285de69d55eeefef715573a8 Mon Sep 17 00:00:00 2001 From: vv7r Date: Sun, 4 Oct 2026 23:19:53 +0000 Subject: [PATCH 4/6] fix(events): count open messages and pending tool calls as sets A successful compaction starts its summary attempt's message again with the summary as content, then closes it once. The tail counted starts minus dones, so every compaction left one message open for good: the snapshot cadence then found the turn busy at every check and never took a mid-turn snapshot for the rest of the turn (main turn or sub-agent). The tail now counts the ids started and not done in it (and tool calls without a result), instead of subtracting counts: a message started twice counts once, and a done or result whose start is before the tail can no longer hide another open message or pending call. The counts read only their own event types through the session/type/seq index; the check measured 30 ms on a 48 MB tail (the default budget), 64 ms on 100 MB, 269 ms on 400 MB, dominated by summing the tail's payload bytes. Co-Authored-By: Claude Opus 5.5 --- src/server/events/store.test.ts | 44 +++++++++++++++++++++++++++++++++ src/server/events/store.ts | 24 +++++++++++++----- 2 files changed, 62 insertions(+), 6 deletions(-) diff --git a/src/server/events/store.test.ts b/src/server/events/store.test.ts index 94eb5fd64..cf7546945 100644 --- a/src/server/events/store.test.ts +++ b/src/server/events/store.test.ts @@ -1444,6 +1444,50 @@ 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('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..9f43dedb9 100644 --- a/src/server/events/store.ts +++ b/src/server/events/store.ts @@ -626,13 +626,25 @@ 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. 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)) 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)) 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 From 542a8330d91113ba36089dafc6168ce8d8829fc5 Mon Sep 17 00:00:00 2001 From: vv7r Date: Sat, 3 Oct 2026 11:01:21 +0000 Subject: [PATCH 5/6] fix(events): snapshot at the next quiet point once a check was deferred The cadence measures the tail at most every 30 s, on whatever event comes in, and only snapshots at a quiet point (no message streaming, no tool round in flight). A deferred check still restarted the 30 s period, and a sub-agent is nearly always streaming or running a tool: the check kept landing on a busy moment and the snapshot was put off to the end of the turn. Seen live on 7 and 12 minute sub-agent runs, the tail over its budget and no mid-turn snapshot (together with the open-message miscount fixed in the next commit). A check that finds the tail over budget while the turn is busy now marks the cadence deferred, and the next quiet point (a message.done or a tool.result) checks again without waiting for the period, at most every 5 s: each check scans the tail (~0.1 s at 46 MB), and parallel tools or sub-agents can keep the turn busy over many quiet points in a row. The budget is evaluated before the in-flight test, so only an over-budget tail defers. Co-Authored-By: Claude Opus 5.5 --- src/server/chat/snapshot-cadence.test.ts | 73 +++++++++++++++++++++++- src/server/chat/snapshot-cadence.ts | 39 ++++++++++--- 2 files changed, 104 insertions(+), 8 deletions(-) diff --git a/src/server/chat/snapshot-cadence.test.ts b/src/server/chat/snapshot-cadence.test.ts index 153907fe7..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. 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(), } From 72d614ebade1e86fec9fcaea768c151effbea224 Mon Sep 17 00:00:00 2001 From: vv7r Date: Mon, 5 Oct 2026 17:13:34 +0000 Subject: [PATCH 6/6] fix(events): skip NULL ids in the tail's open counts `x NOT IN (subquery)` is never true when the subquery returns a NULL: a single message.done or tool.result without an id in the tail would read openMessages and pendingToolCalls as 0, and the cadence would snapshot mid-message. The subqueries now skip NULL ids. Co-Authored-By: Claude Opus 5.5 --- src/server/events/store.test.ts | 17 +++++++++++++++++ src/server/events/store.ts | 12 ++++++++---- 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/src/server/events/store.test.ts b/src/server/events/store.test.ts index cf7546945..aca2bd471 100644 --- a/src/server/events/store.test.ts +++ b/src/server/events/store.test.ts @@ -1467,6 +1467,23 @@ describe('EventStore - Event Cleanup', () => { 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', diff --git a/src/server/events/store.ts b/src/server/events/store.ts index 9f43dedb9..09a446477 100644 --- a/src/server/events/store.ts +++ b/src/server/events/store.ts @@ -630,18 +630,22 @@ export class EventStore { -- 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. Subqueries so they only read their own event - -- types (session/type/seq index), not every tail row. + -- 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)) AS openMessages, + 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)) AS pendingToolCalls + 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 {