From f2ab29bc9afee5643d8c1e2d7f44f2ba7c0d2ae2 Mon Sep 17 00:00:00 2001 From: vv7r Date: Sun, 4 Oct 2026 17:46:33 +0000 Subject: [PATCH 1/4] 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 2/4] 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 3/4] 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 4/4] 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 {