fix(events): keep the event log bounded during long turns

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
 

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

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

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({

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<StoredEvent[]> {
   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, {

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 })),
   }
 }
 

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<void> {
   const { sessionManager, sessionId } = options
   const eventStore = getEventStore()
-  const statsIdentity = resolveStatsIdentity(options)
 
   const session = sessionManager.requireSession(sessionId)
   const mode = session.mode
@@ -189,31 +188,32 @@ export async function runChatTurn(options: OrchestratorOptions): Promise<void> {
   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 {
       // Session may have been deleted (e.g. during abort) — skip
     }
   }
 
+  // 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()
 
   try {
     // 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<void> {
     }
 
     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 } : {}),
-  })
-}