From 16439fd499799eb93dabe4c1a78bcf78da9583b3 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Mon, 6 Jul 2026 15:13:00 -0700 Subject: [PATCH] perf(ai-vault): extend incremental transcript parsing to all append-only JSONL agents (STA-1417) (#7593) Co-authored-by: Orca --- .../ai-vault/session-scanner-accumulator.ts | 31 +- .../session-scanner-codex-fixtures.ts | 92 ++++ .../ai-vault/session-scanner-codex-parser.ts | 439 +++++++++--------- .../session-scanner-codex-title-index.ts | 93 ++++ .../ai-vault/session-scanner-droid-parser.ts | 88 ++-- .../ai-vault/session-scanner-graph-parsers.ts | 95 ++-- .../session-scanner-incremental-fixtures.ts | 240 ++++++++++ ...session-scanner-parse-cache-agents.test.ts | 250 ++++++++++ .../ai-vault/session-scanner-parse-cache.ts | 101 ++-- .../session-scanner-primary-parsers.ts | 81 +++- .../session-scanner-secondary-parsers.ts | 181 ++++---- .../ai-vault/session-scanner-token-values.ts | 13 + src/main/ai-vault/session-scanner-types.ts | 20 + src/main/ai-vault/session-scanner-values.ts | 1 + 14 files changed, 1294 insertions(+), 431 deletions(-) create mode 100644 src/main/ai-vault/session-scanner-codex-fixtures.ts create mode 100644 src/main/ai-vault/session-scanner-codex-title-index.ts create mode 100644 src/main/ai-vault/session-scanner-incremental-fixtures.ts create mode 100644 src/main/ai-vault/session-scanner-parse-cache-agents.test.ts diff --git a/src/main/ai-vault/session-scanner-accumulator.ts b/src/main/ai-vault/session-scanner-accumulator.ts index 85b33a962..c31408a84 100644 --- a/src/main/ai-vault/session-scanner-accumulator.ts +++ b/src/main/ai-vault/session-scanner-accumulator.ts @@ -7,7 +7,11 @@ import { type AiVaultSessionPreviewMessage } from '../../shared/ai-vault-types' import { LOCAL_EXECUTION_HOST_ID, type ExecutionHostId } from '../../shared/execution-host' -import type { FileWithMtime, SessionAccumulator } from './session-scanner-types' +import type { + FileWithMtime, + ResumableSessionParseState, + SessionAccumulator +} from './session-scanner-types' import { extractPreviewContentText, extractString, @@ -41,6 +45,31 @@ export function createAccumulator(args: { } } +export function cloneSessionAccumulator(accumulator: SessionAccumulator): SessionAccumulator { + return { ...accumulator, previewMessages: [...accumulator.previewMessages] } +} + +// Resumable fold for parsers whose only parse state is the accumulator itself +// (cursor, copilot, droid, openclaw/pi, gemini-jsonl). Parsers with extra +// closure state (claude, codex) build their own ResumableSessionParseState. +export function accumulatorFoldResumeState( + accumulator: SessionAccumulator, + consumeRecordLine: (accumulator: SessionAccumulator, line: string) => void +): ResumableSessionParseState { + return { + consumeLine: (line) => consumeRecordLine(accumulator, line), + clone: () => + accumulatorFoldResumeState(cloneSessionAccumulator(accumulator), consumeRecordLine), + touchFile: (file) => { + accumulator.modifiedAt = file.modifiedAt + }, + // Finalize a snapshot: the live accumulator (and its preview array) keeps + // accumulating appended lines after this session object is handed out. + finalize: (platform, options) => + finalizeSession(cloneSessionAccumulator(accumulator), platform, options) + } +} + export function finalizeSession( accumulator: SessionAccumulator, platform: NodeJS.Platform, diff --git a/src/main/ai-vault/session-scanner-codex-fixtures.ts b/src/main/ai-vault/session-scanner-codex-fixtures.ts new file mode 100644 index 000000000..ffdab7f1d --- /dev/null +++ b/src/main/ai-vault/session-scanner-codex-fixtures.ts @@ -0,0 +1,92 @@ +import type { IncrementalAgentFixture } from './session-scanner-incremental-fixtures' + +// Codex fixture lines for the incremental-parse differential tests, split out +// of session-scanner-incremental-fixtures.ts to respect the max-lines budget. + +const CODEX_SESSION_ID = '019f0000-1111-7222-8333-444444444444' + +function codexLine(record: Record): string { + return JSON.stringify(record) +} + +export function codexFixture(): IncrementalAgentFixture { + return { + agent: 'codex', + fileName: `rollout-2026-05-01T10-00-00-${CODEX_SESSION_ID}.jsonl`, + seedLines: [ + codexLine({ + timestamp: '2026-05-01T10:00:00.000Z', + type: 'session_meta', + payload: { id: CODEX_SESSION_ID, cwd: '/repo/app', git: { branch: 'feature/vault' } } + }), + codexLine({ + timestamp: '2026-05-01T10:00:05.000Z', + type: 'response_item', + payload: { type: 'message', role: 'user', content: 'codex seed question' } + }), + codexLine({ + timestamp: '2026-05-01T10:00:10.000Z', + type: 'response_item', + payload: { type: 'message', role: 'assistant', content: 'codex seed answer' } + }), + codexLine({ + timestamp: '2026-05-01T10:00:11.000Z', + type: 'event_msg', + payload: { + type: 'token_count', + info: { total_token_usage: { input_tokens: 100, output_tokens: 40, total_tokens: 140 } }, + model: 'gpt-5.1-codex' + } + }) + ], + appendLines: [ + codexLine({ + timestamp: '2026-05-01T10:05:00.000Z', + type: 'event_msg', + payload: { type: 'user_message', message: 'codex follow-up' } + }), + codexLine({ + timestamp: '2026-05-01T10:05:20.000Z', + type: 'event_msg', + payload: { type: 'agent_message', message: 'codex incremental answer' } + }), + codexLine({ + timestamp: '2026-05-01T10:05:21.000Z', + type: 'event_msg', + payload: { + type: 'token_count', + info: { total_token_usage: { input_tokens: 220, output_tokens: 90, total_tokens: 310 } } + } + }) + ], + truncatedLines: [ + codexLine({ + timestamp: '2026-05-01T10:00:00.000Z', + type: 'session_meta', + payload: { id: CODEX_SESSION_ID, cwd: '/repo/app' } + }), + codexLine({ + timestamp: '2026-05-01T10:00:05.000Z', + type: 'event_msg', + payload: { type: 'user_message', message: 'rewritten only turn' } + }) + ] + } +} + +export function codexWorkerFixtureLines(): string[] { + return [ + codexLine({ + timestamp: '2026-05-01T10:00:00.000Z', + type: 'session_meta', + payload: { id: CODEX_SESSION_ID, cwd: '/repo/app', thread_source: 'subagent' } + }), + codexLine({ + timestamp: '2026-05-01T10:00:05.000Z', + type: 'event_msg', + payload: { type: 'user_message', message: 'worker turn' } + }) + ] +} + +export const CODEX_FIXTURE_SESSION_ID = CODEX_SESSION_ID diff --git a/src/main/ai-vault/session-scanner-codex-parser.ts b/src/main/ai-vault/session-scanner-codex-parser.ts index b18721a82..5127720b1 100644 --- a/src/main/ai-vault/session-scanner-codex-parser.ts +++ b/src/main/ai-vault/session-scanner-codex-parser.ts @@ -1,18 +1,25 @@ import { createReadStream } from 'node:fs' -import { stat } from 'node:fs/promises' -import { basename, dirname, join } from 'node:path' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' +import { readCodexSessionIndexTitle } from './session-scanner-codex-title-index' import type { ExecutionHostId } from '../../shared/execution-host' import { addPreviewContent, + cloneSessionAccumulator, createAccumulator, finalizeSession, sessionIdFromFileName, updateTimeline } from './session-scanner-accumulator' -import type { CodexUsageSnapshot, FileWithMtime } from './session-scanner-types' +import type { + CodexUsageSnapshot, + FileWithMtime, + ResumableParseFinalizeOptions, + ResumableSessionParseState, + SessionAccumulator +} from './session-scanner-types' import { + addCodexUsage, asRecord, extractContentText, extractGitBranch, @@ -24,15 +31,6 @@ import { subtractCodexUsage } from './session-scanner-values' -const CODEX_SESSION_INDEX_FILE = 'session_index.jsonl' - -type CodexSessionIndexTitleCacheEntry = { - signature: string - titles: Map -} - -const codexSessionIndexTitleCache = new Map>() - export async function parseCodexSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform, @@ -74,6 +72,213 @@ export async function parseCodexSessionContent(args: { }) } +type CodexSessionParseState = { + accumulator: SessionAccumulator + previousTotals: CodexUsageSnapshot | null + rejectedWorkerSession: boolean + sawSessionMeta: boolean + // Which source set the current title; an index-file title outranks the raw + // first user prompt, so finalize must know whether 'meta' already won. + titleSource: 'meta' | 'user' | null +} + +function createCodexParseState(file: FileWithMtime): CodexSessionParseState { + return { + accumulator: createAccumulator({ + agent: 'codex', + file, + sessionId: sessionIdFromFileName(file.path) + }), + previousTotals: null, + rejectedWorkerSession: false, + sawSessionMeta: false, + titleSource: null + } +} + +function cloneCodexParseState(state: CodexSessionParseState): CodexSessionParseState { + return { + // previousTotals snapshots are replaced, never mutated, so sharing is safe. + ...state, + accumulator: cloneSessionAccumulator(state.accumulator) + } +} + +function consumeCodexRecordLine(state: CodexSessionParseState, line: string): void { + if (state.rejectedWorkerSession) { + return + } + const record = parseJsonObject(line) + if (!record) { + return + } + const { accumulator } = state + + updateTimeline(accumulator, extractString(record.timestamp)) + + const payload = asRecord(record.payload) + if (record.type === 'session_meta' && payload) { + if (isCodexWorkerSession(payload)) { + // Why: Codex writes internal worker/sub-agent transcripts into the same + // history tree; AI Vault should show user-started sessions only. + state.rejectedWorkerSession = true + return + } + state.sawSessionMeta = true + const sessionId = extractString(payload.id) + if (sessionId) { + accumulator.sessionId = sessionId + } + const metadataTitle = extractCodexSessionMetadataTitle(payload) + if (metadataTitle) { + accumulator.title = metadataTitle + state.titleSource = 'meta' + } + const cwd = extractString(payload.cwd) + if (cwd) { + accumulator.cwd = cwd + } + accumulator.branch = extractGitBranch(payload.git) ?? accumulator.branch + return + } + + if (record.type === 'turn_context' && payload) { + const cwd = extractString(payload.cwd) + if (cwd) { + accumulator.cwd = cwd + } + const model = extractModel(payload) + if (model) { + accumulator.model = model + } + return + } + + if (!payload) { + return + } + + if (record.type === 'response_item' && payload.type === 'message') { + accumulator.messageCount++ + if (payload.role === 'user' && !accumulator.title) { + accumulator.title = extractContentText(payload.content) + state.titleSource = accumulator.title ? 'user' : state.titleSource + } + addPreviewContent( + accumulator, + payload.role === 'assistant' ? 'assistant' : payload.role === 'user' ? 'user' : 'unknown', + payload.content, + record.timestamp + ) + return + } + + if (record.type !== 'event_msg') { + return + } + + if (payload.type === 'user_message') { + accumulator.messageCount++ + if (!accumulator.title) { + accumulator.title = extractContentText(payload.message) + state.titleSource = accumulator.title ? 'user' : state.titleSource + } + addPreviewContent(accumulator, 'user', payload.message, record.timestamp) + return + } + + if (payload.type === 'agent_message') { + accumulator.messageCount++ + addPreviewContent(accumulator, 'assistant', payload.message, record.timestamp) + return + } + + if (payload.type !== 'token_count') { + return + } + + const info = asRecord(payload.info) + if (!info) { + return + } + const totalUsage = normalizeCodexUsage(info.total_token_usage) + const lastUsage = normalizeCodexUsage(info.last_token_usage) + let delta: CodexUsageSnapshot | null = null + if (totalUsage) { + delta = subtractCodexUsage(totalUsage, state.previousTotals) + state.previousTotals = totalUsage + } else if (lastUsage) { + delta = lastUsage + state.previousTotals = state.previousTotals + ? addCodexUsage(state.previousTotals, lastUsage) + : lastUsage + } + if (delta) { + accumulator.totalTokens += delta.totalTokens + } + const model = extractModel(payload) + if (model) { + accumulator.model = model + } +} + +async function finalizeCodexParseState( + state: CodexSessionParseState, + platform: NodeJS.Platform, + args: { + codexHome: string | null + titleReader?: (sessionId: string) => Promise + executionHostId?: ExecutionHostId + executionHostPlatform?: NodeJS.Platform | null + } +): Promise { + if (state.rejectedWorkerSession) { + return null + } + // Finalize a snapshot: the live state keeps accumulating appended lines. + const snapshot = cloneCodexParseState(state) + // Why: Codex names threads lazily in session_index.jsonl, so the lookup runs + // per finalize (the index read is signature-cached) — a title that appears + // after the transcript was first parsed must still replace the raw prompt. + if (snapshot.sawSessionMeta && snapshot.titleSource !== 'meta') { + const indexedTitle = await args.titleReader?.(snapshot.accumulator.sessionId) + if (indexedTitle) { + snapshot.accumulator.title = indexedTitle + } + } + return finalizeSession(snapshot.accumulator, platform, { + codexHome: args.codexHome, + executionHostId: args.executionHostId, + executionHostPlatform: args.executionHostPlatform + }) +} + +export function createCodexSessionResumeState( + file: FileWithMtime, + codexHome: string | null +): ResumableSessionParseState { + return codexResumeStateFromParseState(createCodexParseState(file), codexHome, (sessionId) => + readCodexSessionIndexTitle(file.path, codexHome, sessionId) + ) +} + +function codexResumeStateFromParseState( + state: CodexSessionParseState, + codexHome: string | null, + titleReader: (sessionId: string) => Promise +): ResumableSessionParseState { + return { + consumeLine: (line) => consumeCodexRecordLine(state, line), + clone: () => + codexResumeStateFromParseState(cloneCodexParseState(state), codexHome, titleReader), + touchFile: (file) => { + state.accumulator.modifiedAt = file.modifiedAt + }, + finalize: (platform, options?: ResumableParseFinalizeOptions) => + finalizeCodexParseState(state, platform, { codexHome, titleReader, ...options }) + } +} + async function parseCodexSessionLines(args: { file: FileWithMtime lines: AsyncIterable | Iterable @@ -83,142 +288,22 @@ async function parseCodexSessionLines(args: { executionHostPlatform?: NodeJS.Platform | null titleReader?: (sessionId: string) => Promise }): Promise { - const accumulator = createAccumulator({ - agent: 'codex', - file: args.file, - sessionId: sessionIdFromFileName(args.file.path) - }) - let previousTotals: CodexUsageSnapshot | null = null - + const state = createCodexParseState(args.file) for await (const line of args.lines) { - const record = parseJsonObject(line) - if (!record) { - continue - } - - updateTimeline(accumulator, extractString(record.timestamp)) - - const payload = asRecord(record.payload) - if (record.type === 'session_meta' && payload) { - if (isCodexWorkerSession(payload)) { - // Why: Codex writes internal worker/sub-agent transcripts into the same - // history tree; AI Vault should show user-started sessions only. - return null - } - const sessionId = extractString(payload.id) - if (sessionId) { - accumulator.sessionId = sessionId - } - const indexedTitle = - extractCodexSessionMetadataTitle(payload) ?? - (await args.titleReader?.(accumulator.sessionId)) - if (indexedTitle) { - accumulator.title = indexedTitle - } - const cwd = extractString(payload.cwd) - if (cwd) { - accumulator.cwd = cwd - } - accumulator.branch = extractGitBranch(payload.git) ?? accumulator.branch - continue - } - - if (record.type === 'turn_context' && payload) { - const cwd = extractString(payload.cwd) - if (cwd) { - accumulator.cwd = cwd - } - const model = extractModel(payload) - if (model) { - accumulator.model = model - } - continue - } - - if (!payload) { - continue - } - - if (record.type === 'response_item' && payload.type === 'message') { - accumulator.messageCount++ - if (payload.role === 'user' && !accumulator.title) { - accumulator.title = extractContentText(payload.content) - } - addPreviewContent( - accumulator, - payload.role === 'assistant' ? 'assistant' : payload.role === 'user' ? 'user' : 'unknown', - payload.content, - record.timestamp - ) - continue - } - - if (record.type !== 'event_msg') { - continue - } - - if (payload.type === 'user_message') { - accumulator.messageCount++ - if (!accumulator.title) { - accumulator.title = extractContentText(payload.message) - } - addPreviewContent(accumulator, 'user', payload.message, record.timestamp) - continue - } - - if (payload.type === 'agent_message') { - accumulator.messageCount++ - addPreviewContent(accumulator, 'assistant', payload.message, record.timestamp) - continue - } - - if (payload.type !== 'token_count') { - continue - } - - const info = asRecord(payload.info) - if (!info) { - continue - } - const totalUsage = normalizeCodexUsage(info.total_token_usage) - const lastUsage = normalizeCodexUsage(info.last_token_usage) - let delta: CodexUsageSnapshot | null = null - if (totalUsage) { - delta = subtractCodexUsage(totalUsage, previousTotals) - previousTotals = totalUsage - } else if (lastUsage) { - delta = lastUsage - previousTotals = previousTotals ? addCodexUsage(previousTotals, lastUsage) : lastUsage - } - if (delta) { - accumulator.totalTokens += delta.totalTokens - } - const model = extractModel(payload) - if (model) { - accumulator.model = model + consumeCodexRecordLine(state, line) + if (state.rejectedWorkerSession) { + // Worker transcripts are excluded outright; stop reading early. + return null } } - - return finalizeSession(accumulator, args.platform, { + return finalizeCodexParseState(state, args.platform, { codexHome: args.codexHome, + titleReader: args.titleReader, executionHostId: args.executionHostId, executionHostPlatform: args.executionHostPlatform }) } -function addCodexUsage( - base: CodexUsageSnapshot, - increment: CodexUsageSnapshot -): CodexUsageSnapshot { - return { - inputTokens: base.inputTokens + increment.inputTokens, - cachedInputTokens: base.cachedInputTokens + increment.cachedInputTokens, - outputTokens: base.outputTokens + increment.outputTokens, - reasoningOutputTokens: base.reasoningOutputTokens + increment.reasoningOutputTokens, - totalTokens: base.totalTokens + increment.totalTokens - } -} - function extractCodexThreadSource(payload: Record): string | null { return extractString(payload.thread_source) ?? extractString(payload.threadSource) } @@ -240,79 +325,3 @@ function extractCodexSessionMetadataTitle(payload: Record): str normalizeTitleText(extractString(payload.threadName) ?? '') ) } - -async function readCodexSessionIndexTitle( - sessionFilePath: string, - codexHome: string | null, - sessionId: string -): Promise { - const resolvedCodexHome = codexHome ?? codexHomeFromSessionFilePath(sessionFilePath) - if (!resolvedCodexHome) { - return null - } - const titleBySessionId = await readCodexSessionIndexTitles(resolvedCodexHome) - return titleBySessionId.get(sessionId) ?? null -} - -function codexHomeFromSessionFilePath(sessionFilePath: string): string | null { - let currentDir = dirname(sessionFilePath) - while (currentDir && dirname(currentDir) !== currentDir) { - if (basename(currentDir) === 'sessions') { - return dirname(currentDir) - } - currentDir = dirname(currentDir) - } - return null -} - -async function readCodexSessionIndexTitles(codexHome: string): Promise> { - const indexPath = join(codexHome, CODEX_SESSION_INDEX_FILE) - let signature: string - try { - const indexStat = await stat(indexPath) - signature = `${indexStat.size}:${indexStat.mtimeMs}` - } catch { - return new Map() - } - - const cached = codexSessionIndexTitleCache.get(codexHome) - if (cached) { - const entry = await cached - if (entry.signature === signature) { - return entry.titles - } - } - - const pending = readCodexSessionIndexTitlesFromDisk(indexPath).then((titles) => ({ - signature, - titles - })) - codexSessionIndexTitleCache.set(codexHome, pending) - return (await pending).titles -} - -async function readCodexSessionIndexTitlesFromDisk( - indexPath: string -): Promise> { - const titleBySessionId = new Map() - try { - const lines = createInterface({ - input: createReadStream(indexPath, { encoding: 'utf-8' }), - crlfDelay: Infinity - }) - for await (const line of lines) { - const record = parseJsonObject(line) - if (!record) { - continue - } - const sessionId = extractString(record.id) - const title = normalizeTitleText(extractString(record.thread_name) ?? '') - if (sessionId && title) { - titleBySessionId.set(sessionId, title) - } - } - } catch { - // Codex creates the index opportunistically; older homes may only have raw transcripts. - } - return titleBySessionId -} diff --git a/src/main/ai-vault/session-scanner-codex-title-index.ts b/src/main/ai-vault/session-scanner-codex-title-index.ts new file mode 100644 index 000000000..d22f964f5 --- /dev/null +++ b/src/main/ai-vault/session-scanner-codex-title-index.ts @@ -0,0 +1,93 @@ +import { createReadStream } from 'node:fs' +import { stat } from 'node:fs/promises' +import { basename, dirname, join } from 'node:path' +import { createInterface } from 'node:readline' +import { extractString, normalizeTitleText, parseJsonObject } from './session-scanner-values' + +// Codex names threads lazily in /session_index.jsonl; transcripts +// carry no title of their own, so parsers look the thread name up here. + +const CODEX_SESSION_INDEX_FILE = 'session_index.jsonl' + +type CodexSessionIndexTitleCacheEntry = { + signature: string + titles: Map +} + +const codexSessionIndexTitleCache = new Map>() + +export async function readCodexSessionIndexTitle( + sessionFilePath: string, + codexHome: string | null, + sessionId: string +): Promise { + const resolvedCodexHome = codexHome ?? codexHomeFromSessionFilePath(sessionFilePath) + if (!resolvedCodexHome) { + return null + } + const titleBySessionId = await readCodexSessionIndexTitles(resolvedCodexHome) + return titleBySessionId.get(sessionId) ?? null +} + +function codexHomeFromSessionFilePath(sessionFilePath: string): string | null { + let currentDir = dirname(sessionFilePath) + while (currentDir && dirname(currentDir) !== currentDir) { + if (basename(currentDir) === 'sessions') { + return dirname(currentDir) + } + currentDir = dirname(currentDir) + } + return null +} + +async function readCodexSessionIndexTitles(codexHome: string): Promise> { + const indexPath = join(codexHome, CODEX_SESSION_INDEX_FILE) + let signature: string + try { + const indexStat = await stat(indexPath) + signature = `${indexStat.size}:${indexStat.mtimeMs}` + } catch { + return new Map() + } + + const cached = codexSessionIndexTitleCache.get(codexHome) + if (cached) { + const entry = await cached + if (entry.signature === signature) { + return entry.titles + } + } + + const pending = readCodexSessionIndexTitlesFromDisk(indexPath).then((titles) => ({ + signature, + titles + })) + codexSessionIndexTitleCache.set(codexHome, pending) + return (await pending).titles +} + +async function readCodexSessionIndexTitlesFromDisk( + indexPath: string +): Promise> { + const titleBySessionId = new Map() + try { + const lines = createInterface({ + input: createReadStream(indexPath, { encoding: 'utf-8' }), + crlfDelay: Infinity + }) + for await (const line of lines) { + const record = parseJsonObject(line) + if (!record) { + continue + } + const sessionId = extractString(record.id) + const title = normalizeTitleText(extractString(record.thread_name) ?? '') + if (sessionId && title) { + titleBySessionId.set(sessionId, title) + } + } + } catch { + // Codex creates the index opportunistically; older homes may only have raw transcripts. + } + return titleBySessionId +} diff --git a/src/main/ai-vault/session-scanner-droid-parser.ts b/src/main/ai-vault/session-scanner-droid-parser.ts index d99795bbb..2abaa7b81 100644 --- a/src/main/ai-vault/session-scanner-droid-parser.ts +++ b/src/main/ai-vault/session-scanner-droid-parser.ts @@ -2,11 +2,15 @@ import { createReadStream } from 'node:fs' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' -import type { FileWithMtime, SessionAccumulator } from './session-scanner-types' +import type { + FileWithMtime, + ResumableSessionParseState, + SessionAccumulator +} from './session-scanner-types' import { + accumulatorFoldResumeState, addPreviewMessage, createAccumulator, - finalizeSession, sessionIdFromFileName, updateTimeline } from './session-scanner-accumulator' @@ -50,51 +54,57 @@ export async function parseDroidSessionContent( }) } +function consumeDroidRecordLine(accumulator: SessionAccumulator, line: string): void { + const record = parseJsonObject(line) + if (!record) { + return + } + updateTimeline(accumulator, record.timestamp) + if (record.type === 'session_start') { + accumulator.sessionId = extractString(record.id) ?? accumulator.sessionId + accumulator.title = normalizeTitleText(extractString(record.title) ?? '') + accumulator.cwd = extractString(record.cwd) ?? accumulator.cwd + return + } + if (record.type === 'system') { + accumulator.cwd = extractString(record.cwd) ?? accumulator.cwd + accumulator.model = extractString(record.model) ?? accumulator.model + } + const streamSessionId = extractString(record.session_id) ?? extractString(record.sessionId) + if (streamSessionId) { + accumulator.sessionId = streamSessionId + } + if (record.type === 'message') { + consumeDroidMessage(accumulator, record) + } else if (record.type === 'completion') { + accumulator.messageCount++ + accumulator.totalTokens += tokenTotal(record.usage) + addPreviewMessage(accumulator, { + role: 'assistant', + text: extractString(record.finalText), + timestamp: record.timestamp + }) + } +} + +export function createDroidSessionResumeState(file: FileWithMtime): ResumableSessionParseState { + return accumulatorFoldResumeState( + createAccumulator({ agent: 'droid', file, sessionId: sessionIdFromFileName(file.path) }), + consumeDroidRecordLine + ) +} + async function parseDroidSessionLines(args: { file: FileWithMtime lines: AsyncIterable | Iterable platform: NodeJS.Platform options?: ParserSessionOptions }): Promise { - const accumulator = createAccumulator({ - agent: 'droid', - file: args.file, - sessionId: sessionIdFromFileName(args.file.path) - }) - + const state = createDroidSessionResumeState(args.file) for await (const line of args.lines) { - const record = parseJsonObject(line) - if (!record) { - continue - } - updateTimeline(accumulator, record.timestamp) - if (record.type === 'session_start') { - accumulator.sessionId = extractString(record.id) ?? accumulator.sessionId - accumulator.title = normalizeTitleText(extractString(record.title) ?? '') - accumulator.cwd = extractString(record.cwd) ?? accumulator.cwd - continue - } - if (record.type === 'system') { - accumulator.cwd = extractString(record.cwd) ?? accumulator.cwd - accumulator.model = extractString(record.model) ?? accumulator.model - } - const streamSessionId = extractString(record.session_id) ?? extractString(record.sessionId) - if (streamSessionId) { - accumulator.sessionId = streamSessionId - } - if (record.type === 'message') { - consumeDroidMessage(accumulator, record) - } else if (record.type === 'completion') { - accumulator.messageCount++ - accumulator.totalTokens += tokenTotal(record.usage) - addPreviewMessage(accumulator, { - role: 'assistant', - text: extractString(record.finalText), - timestamp: record.timestamp - }) - } + state.consumeLine(line) } - return finalizeSession(accumulator, args.platform, args.options) + return state.finalize(args.platform, args.options) } function consumeDroidMessage( diff --git a/src/main/ai-vault/session-scanner-graph-parsers.ts b/src/main/ai-vault/session-scanner-graph-parsers.ts index a52c83b32..01d84ddfa 100644 --- a/src/main/ai-vault/session-scanner-graph-parsers.ts +++ b/src/main/ai-vault/session-scanner-graph-parsers.ts @@ -4,8 +4,13 @@ import { basename, dirname, join } from 'node:path' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' -import type { FileWithMtime, SessionAccumulator } from './session-scanner-types' +import type { + FileWithMtime, + ResumableSessionParseState, + SessionAccumulator +} from './session-scanner-types' import { + accumulatorFoldResumeState, addPreviewContent, addPreviewMessage, createAccumulator, @@ -186,6 +191,51 @@ export async function parseMessageGraphSessionContent( }) } +function consumeMessageGraphRecordLine(accumulator: SessionAccumulator, line: string): void { + const record = parseJsonObject(line) + if (!record) { + return + } + updateTimeline(accumulator, extractString(record.timestamp)) + if (record.type === 'session') { + const sessionId = extractString(record.id) + if (sessionId) { + accumulator.sessionId = sessionId + } + accumulator.cwd = extractString(record.cwd) ?? accumulator.cwd + return + } + if (record.type === 'model_change') { + accumulator.model = extractString(record.modelId) ?? accumulator.model + return + } + if (record.type !== 'message') { + return + } + const message = asRecord(record.message) + const role = extractString(message?.role) + if (role === 'user' || role === 'assistant') { + accumulator.messageCount++ + if (role === 'user') { + accumulator.title ??= extractMessageText(message) + } else { + accumulator.model = extractString(message?.model) ?? accumulator.model + accumulator.totalTokens += tokenTotal(message?.usage) + } + addPreviewContent(accumulator, role, message?.content, record.timestamp) + } +} + +export function createMessageGraphSessionResumeState( + agent: 'openclaw' | 'pi', + file: FileWithMtime +): ResumableSessionParseState { + return accumulatorFoldResumeState( + createAccumulator({ agent, file, sessionId: sessionIdFromFileName(file.path) }), + consumeMessageGraphRecordLine + ) +} + async function parseMessageGraphSessionLines(args: { agent: 'openclaw' | 'pi' file: FileWithMtime @@ -193,46 +243,9 @@ async function parseMessageGraphSessionLines(args: { platform: NodeJS.Platform options?: ParserSessionOptions }): Promise { - const accumulator = createAccumulator({ - agent: args.agent, - file: args.file, - sessionId: sessionIdFromFileName(args.file.path) - }) - + const state = createMessageGraphSessionResumeState(args.agent, args.file) for await (const line of args.lines) { - const record = parseJsonObject(line) - if (!record) { - continue - } - updateTimeline(accumulator, extractString(record.timestamp)) - if (record.type === 'session') { - const sessionId = extractString(record.id) - if (sessionId) { - accumulator.sessionId = sessionId - } - accumulator.cwd = extractString(record.cwd) ?? accumulator.cwd - continue - } - if (record.type === 'model_change') { - accumulator.model = extractString(record.modelId) ?? accumulator.model - continue - } - if (record.type !== 'message') { - continue - } - const message = asRecord(record.message) - const role = extractString(message?.role) - if (role === 'user' || role === 'assistant') { - accumulator.messageCount++ - if (role === 'user') { - accumulator.title ??= extractMessageText(message) - } else { - accumulator.model = extractString(message?.model) ?? accumulator.model - accumulator.totalTokens += tokenTotal(message?.usage) - } - addPreviewContent(accumulator, role, message?.content, record.timestamp) - } + state.consumeLine(line) } - - return finalizeSession(accumulator, args.platform, args.options) + return state.finalize(args.platform, args.options) } diff --git a/src/main/ai-vault/session-scanner-incremental-fixtures.ts b/src/main/ai-vault/session-scanner-incremental-fixtures.ts new file mode 100644 index 000000000..01219836a --- /dev/null +++ b/src/main/ai-vault/session-scanner-incremental-fixtures.ts @@ -0,0 +1,240 @@ +import type { AiVaultAgent } from '../../shared/ai-vault-types' + +import { codexFixture } from './session-scanner-codex-fixtures' + +// Line builders for the incremental-parse differential tests: each agent gets +// a seed transcript, an appended continuation, and a truncated rewrite, all in +// that agent's real on-disk JSONL record shapes. + +export type IncrementalAgentFixture = { + agent: AiVaultAgent + fileName: string + seedLines: string[] + appendLines: string[] + truncatedLines: string[] +} + +export function cursorFixture(): IncrementalAgentFixture { + const line = (role: string, text: string, at: string) => + JSON.stringify({ role, message: { content: text }, timestamp: at }) + return { + agent: 'cursor', + fileName: 'agent-transcripts-aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee.jsonl', + seedLines: [ + line('user', 'cursor seed question', '2026-05-01T10:00:00.000Z'), + line('assistant', 'cursor seed answer', '2026-05-01T10:01:00.000Z') + ], + appendLines: [ + line('user', 'cursor follow-up', '2026-05-01T10:02:00.000Z'), + line('assistant', 'cursor incremental answer', '2026-05-01T10:03:00.000Z') + ], + truncatedLines: [line('user', 'cursor rewritten', '2026-05-01T10:00:00.000Z')] + } +} + +export function copilotFixture(): IncrementalAgentFixture { + const line = (type: string, data: Record, at: string) => + JSON.stringify({ type, data, timestamp: at }) + return { + agent: 'copilot', + fileName: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee.jsonl', + seedLines: [ + line( + 'session.start', + { sessionId: 'copilot-session-1', startTime: '2026-05-01T10:00:00.000Z' }, + '2026-05-01T10:00:00.000Z' + ), + line('user.message', { content: 'copilot seed question' }, '2026-05-01T10:00:05.000Z'), + line('assistant.message', { content: 'copilot seed answer' }, '2026-05-01T10:00:30.000Z') + ], + appendLines: [ + line('user.message', { content: 'copilot follow-up' }, '2026-05-01T10:05:00.000Z'), + line( + 'assistant.message', + { content: 'copilot incremental answer' }, + '2026-05-01T10:05:30.000Z' + ), + line( + 'session.shutdown', + { currentModel: 'gpt-5.1', currentTokens: 340 }, + '2026-05-01T10:06:00.000Z' + ) + ], + truncatedLines: [ + line( + 'session.start', + { sessionId: 'copilot-session-1', startTime: '2026-05-01T10:00:00.000Z' }, + '2026-05-01T10:00:00.000Z' + ) + ] + } +} + +export function droidFixture(): IncrementalAgentFixture { + return { + agent: 'droid', + fileName: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee.jsonl', + seedLines: [ + JSON.stringify({ + type: 'session_start', + id: 'droid-session-1', + title: 'Droid seed task', + cwd: '/repo/app', + timestamp: '2026-05-01T10:00:00.000Z' + }), + JSON.stringify({ + type: 'message', + role: 'user', + text: 'droid seed question', + timestamp: '2026-05-01T10:00:05.000Z' + }) + ], + appendLines: [ + JSON.stringify({ + type: 'completion', + finalText: 'droid incremental answer', + usage: { input_tokens: 50, output_tokens: 25 }, + timestamp: '2026-05-01T10:01:00.000Z' + }) + ], + truncatedLines: [ + JSON.stringify({ + type: 'session_start', + id: 'droid-session-1', + title: 'Droid rewritten', + timestamp: '2026-05-01T10:00:00.000Z' + }) + ] + } +} + +export function openclawFixture(): IncrementalAgentFixture { + return { + agent: 'openclaw', + fileName: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee.jsonl', + seedLines: [ + JSON.stringify({ + type: 'session', + id: 'openclaw-session-1', + cwd: '/repo/app', + timestamp: '2026-05-01T10:00:00.000Z' + }), + JSON.stringify({ + type: 'message', + message: { role: 'user', content: 'openclaw seed question' }, + timestamp: '2026-05-01T10:00:05.000Z' + }) + ], + appendLines: [ + JSON.stringify({ + type: 'message', + message: { + role: 'assistant', + content: 'openclaw incremental answer', + model: 'claw-1', + usage: { input_tokens: 40, output_tokens: 20 } + }, + timestamp: '2026-05-01T10:01:00.000Z' + }) + ], + truncatedLines: [ + JSON.stringify({ + type: 'session', + id: 'openclaw-session-1', + timestamp: '2026-05-01T10:00:00.000Z' + }) + ] + } +} + +// Pi shares OpenClaw's message-graph format and factory, but gets its own +// fixture so the registry's 'pi' branch is exercised explicitly. +export function piFixture(): IncrementalAgentFixture { + return { + agent: 'pi', + fileName: 'bbbbbbbb-cccc-4ddd-8eee-ffffffffffff.jsonl', + seedLines: [ + JSON.stringify({ + type: 'session', + id: 'pi-session-1', + cwd: '/repo/app', + timestamp: '2026-05-01T10:00:00.000Z' + }), + JSON.stringify({ + type: 'message', + message: { role: 'user', content: 'pi seed question' }, + timestamp: '2026-05-01T10:00:05.000Z' + }) + ], + appendLines: [ + JSON.stringify({ + type: 'model_change', + modelId: 'pi-2', + timestamp: '2026-05-01T10:00:30.000Z' + }), + JSON.stringify({ + type: 'message', + message: { + role: 'assistant', + content: 'pi incremental answer', + usage: { input_tokens: 30, output_tokens: 10 } + }, + timestamp: '2026-05-01T10:01:00.000Z' + }) + ], + truncatedLines: [ + JSON.stringify({ type: 'session', id: 'pi-session-1', timestamp: '2026-05-01T10:00:00.000Z' }) + ] + } +} + +export function geminiJsonlFixture(): IncrementalAgentFixture { + return { + agent: 'gemini', + fileName: 'aaaaaaaa-bbbb-4ccc-8ddd-eeeeeeeeeeee.jsonl', + seedLines: [ + JSON.stringify({ + sessionId: 'gemini-session-1', + startTime: '2026-05-01T10:00:00.000Z', + type: 'user', + content: 'gemini seed question', + timestamp: '2026-05-01T10:00:00.000Z' + }), + JSON.stringify({ + type: 'gemini', + content: 'gemini seed answer', + model: 'gemini-3-pro', + tokens: { input: 80, output: 30 }, + timestamp: '2026-05-01T10:00:30.000Z' + }) + ], + appendLines: [ + JSON.stringify({ + type: 'user', + content: 'gemini follow-up', + timestamp: '2026-05-01T10:01:00.000Z' + }), + JSON.stringify({ $set: { lastUpdated: '2026-05-01T10:01:05.000Z' } }) + ], + truncatedLines: [ + JSON.stringify({ + sessionId: 'gemini-session-1', + type: 'user', + content: 'gemini rewritten', + timestamp: '2026-05-01T10:00:00.000Z' + }) + ] + } +} + +export function allIncrementalAgentFixtures(): IncrementalAgentFixture[] { + return [ + codexFixture(), + cursorFixture(), + copilotFixture(), + droidFixture(), + openclawFixture(), + piFixture(), + geminiJsonlFixture() + ] +} diff --git a/src/main/ai-vault/session-scanner-parse-cache-agents.test.ts b/src/main/ai-vault/session-scanner-parse-cache-agents.test.ts new file mode 100644 index 000000000..61348ad7c --- /dev/null +++ b/src/main/ai-vault/session-scanner-parse-cache-agents.test.ts @@ -0,0 +1,250 @@ +import { appendFile, mkdir, mkdtemp, rm, stat, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { parseAgentSessionFile } from './session-scanner-agent-parser' +import { + CODEX_FIXTURE_SESSION_ID, + codexFixture, + codexWorkerFixtureLines +} from './session-scanner-codex-fixtures' +import { allIncrementalAgentFixtures } from './session-scanner-incremental-fixtures' +import { + createSessionParseStats, + parseAgentSessionFileCached, + resetSessionParseCacheForTests +} from './session-scanner-parse-cache' +import type { SessionFileCandidate } from './session-scanner-types' + +let tempRoots: string[] = [] + +beforeEach(() => { + resetSessionParseCacheForTests() +}) + +afterEach(async () => { + await Promise.all(tempRoots.map((root) => rm(root, { recursive: true, force: true }))) + tempRoots = [] +}) + +async function makeTempDir(): Promise { + const root = await mkdtemp(join(tmpdir(), 'orca-parse-cache-agents-')) + tempRoots.push(root) + return root +} + +async function candidateFor( + agent: SessionFileCandidate['agent'], + path: string, + codexHome: string | null = null +): Promise { + const fileStat = await stat(path) + return { + agent, + file: { + path, + mtimeMs: fileStat.mtimeMs, + modifiedAt: fileStat.mtime.toISOString(), + sizeBytes: fileStat.size + }, + codexHome + } +} + +describe.each(allIncrementalAgentFixtures())('incremental parse parity: $agent', (fixture) => { + it('reuses unchanged files, resumes appends, and matches cold parses exactly', async () => { + const root = await makeTempDir() + const path = join(root, fixture.fileName) + await writeFile(path, `${fixture.seedLines.join('\n')}\n`) + + const stats = createSessionParseStats() + const seedCandidate = await candidateFor(fixture.agent, path) + const seeded = await parseAgentSessionFileCached(seedCandidate, process.platform, stats) + expect(stats.fullParses).toBe(1) + expect(seeded).toEqual(await parseAgentSessionFile(seedCandidate, process.platform)) + + // Unchanged rescan returns the identical cached object. + const reused = await parseAgentSessionFileCached(seedCandidate, process.platform, stats) + expect(reused).toBe(seeded) + expect(stats.reused).toBe(1) + + // Appended lines resume from the stored byte offset and must equal a + // cold parse of the grown file. + await appendFile(path, `${fixture.appendLines.join('\n')}\n`) + const grownCandidate = await candidateFor(fixture.agent, path) + const incremental = await parseAgentSessionFileCached(grownCandidate, process.platform, stats) + expect(stats.incremental).toBe(1) + expect(incremental).toEqual(await parseAgentSessionFile(grownCandidate, process.platform)) + + // A truncated rewrite falls back to a full parse. + await writeFile(path, `${fixture.truncatedLines.join('\n')}\n`) + const truncatedCandidate = await candidateFor(fixture.agent, path) + const reparsed = await parseAgentSessionFileCached(truncatedCandidate, process.platform, stats) + expect(stats.fullParses).toBe(2) + expect(reparsed).toEqual(await parseAgentSessionFile(truncatedCandidate, process.platform)) + }) + + it('includes a trailing unterminated line without double-counting it later', async () => { + const root = await makeTempDir() + const path = join(root, fixture.fileName) + const lastSeedLine = fixture.seedLines.at(-1) + const headLines = fixture.seedLines.slice(0, -1) + await writeFile(path, `${[...headLines, ''].join('\n')}${lastSeedLine}`) + + const partialCandidate = await candidateFor(fixture.agent, path) + const shown = await parseAgentSessionFileCached(partialCandidate, process.platform) + expect(shown).toEqual(await parseAgentSessionFile(partialCandidate, process.platform)) + + await appendFile(path, `\n${fixture.appendLines.join('\n')}\n`) + const grownCandidate = await candidateFor(fixture.agent, path) + const stats = createSessionParseStats() + const completed = await parseAgentSessionFileCached(grownCandidate, process.platform, stats) + expect(stats.incremental).toBe(1) + expect(completed).toEqual(await parseAgentSessionFile(grownCandidate, process.platform)) + }) + + it('tolerates a mid-write truncated trailing line and never double-counts it', async () => { + const root = await makeTempDir() + const path = join(root, fixture.fileName) + // A writer caught mid-record: the trailing line is invalid JSON. + await writeFile(path, `${fixture.seedLines.join('\n')}\n{"type":"user","mess`) + + const shown = await parseAgentSessionFileCached( + await candidateFor(fixture.agent, path), + process.platform + ) + expect(shown).toEqual( + await parseAgentSessionFile(await candidateFor(fixture.agent, path), process.platform) + ) + + // The writer "finishes" the interrupted record as unparseable junk (both + // the fold and a cold parse must skip it identically) and appends more. + await appendFile(path, `age": }\n${fixture.appendLines.join('\n')}\n`) + const stats = createSessionParseStats() + const completed = await parseAgentSessionFileCached( + await candidateFor(fixture.agent, path), + process.platform, + stats + ) + expect(stats.incremental).toBe(1) + expect(completed).toEqual( + await parseAgentSessionFile(await candidateFor(fixture.agent, path), process.platform) + ) + }) +}) + +describe('codex-specific resume behavior', () => { + it('keeps rejecting worker sessions across incremental appends', async () => { + const root = await makeTempDir() + const path = join(root, codexFixture().fileName) + await writeFile(path, `${codexWorkerFixtureLines().join('\n')}\n`) + + const stats = createSessionParseStats() + const seeded = await parseAgentSessionFileCached( + await candidateFor('codex', path), + process.platform, + stats + ) + expect(seeded).toBeNull() + + await appendFile( + path, + `${JSON.stringify({ + timestamp: '2026-05-01T10:10:00.000Z', + type: 'event_msg', + payload: { type: 'agent_message', message: 'worker keeps writing' } + })}\n` + ) + const grown = await parseAgentSessionFileCached( + await candidateFor('codex', path), + process.platform, + stats + ) + expect(stats.incremental).toBe(1) + expect(grown).toBeNull() + }) + + it('picks up a session_index title that appears after the transcript was cached', async () => { + const root = await makeTempDir() + const codexHome = join(root, 'codex-home') + const sessionsDir = join(codexHome, 'sessions', '2026', '05', '01') + await mkdir(sessionsDir, { recursive: true }) + const fixture = codexFixture() + const path = join(sessionsDir, fixture.fileName) + await writeFile(path, `${fixture.seedLines.join('\n')}\n`) + + // No index yet: the title falls back to the first user prompt. + const seeded = await parseAgentSessionFileCached( + await candidateFor('codex', path, codexHome), + process.platform + ) + expect(seeded?.title).toBe('codex seed question') + + // Codex names the thread lazily; the next (incremental) parse must adopt it. + await writeFile( + join(codexHome, 'session_index.jsonl'), + `${JSON.stringify({ id: CODEX_FIXTURE_SESSION_ID, thread_name: 'Indexed thread title' })}\n` + ) + await appendFile(path, `${fixture.appendLines.join('\n')}\n`) + const stats = createSessionParseStats() + const renamed = await parseAgentSessionFileCached( + await candidateFor('codex', path, codexHome), + process.platform, + stats + ) + expect(stats.incremental).toBe(1) + expect(renamed?.title).toBe('Indexed thread title') + expect(renamed).toEqual( + await parseAgentSessionFile(await candidateFor('codex', path, codexHome), process.platform) + ) + }) +}) + +describe('non-resumable formats keep reuse-only caching', () => { + it('re-parses a changed grok summary fully and reuses it when unchanged', async () => { + const root = await makeTempDir() + const sessionDir = join(root, 'session-1') + await mkdir(sessionDir, { recursive: true }) + const path = join(sessionDir, 'summary.json') + await writeFile( + path, + JSON.stringify({ + session_id: 'grok-1', + title: 'Grok seed', + updated_at: '2026-05-01T10:00:00Z' + }) + ) + + const stats = createSessionParseStats() + const seeded = await parseAgentSessionFileCached( + await candidateFor('grok', path), + process.platform, + stats + ) + const reused = await parseAgentSessionFileCached( + await candidateFor('grok', path), + process.platform, + stats + ) + expect(reused).toBe(seeded) + expect(stats).toMatchObject({ fullParses: 1, reused: 1, incremental: 0 }) + + await writeFile( + path, + JSON.stringify({ + session_id: 'grok-1', + title: 'Grok rewritten with a longer title', + updated_at: '2026-05-01T11:00:00Z' + }) + ) + const rewritten = await parseAgentSessionFileCached( + await candidateFor('grok', path), + process.platform, + stats + ) + expect(stats).toMatchObject({ fullParses: 2, incremental: 0 }) + expect(rewritten).toEqual( + await parseAgentSessionFile(await candidateFor('grok', path), process.platform) + ) + }) +}) diff --git a/src/main/ai-vault/session-scanner-parse-cache.ts b/src/main/ai-vault/session-scanner-parse-cache.ts index f8e20f92a..dcc7fc649 100644 --- a/src/main/ai-vault/session-scanner-parse-cache.ts +++ b/src/main/ai-vault/session-scanner-parse-cache.ts @@ -2,14 +2,18 @@ import { createReadStream } from 'node:fs' import { open } from 'node:fs/promises' import type { AiVaultSession } from '../../shared/ai-vault-types' import { parseAgentSessionFile } from './session-scanner-agent-parser' +import { createCodexSessionResumeState } from './session-scanner-codex-parser' +import { createDroidSessionResumeState } from './session-scanner-droid-parser' +import { createMessageGraphSessionResumeState } from './session-scanner-graph-parsers' import { - cloneClaudeSessionParseState, - consumeClaudeSessionLine, - createClaudeSessionParseState, - finalizeClaudeSessionParseState, - type ClaudeSessionParseState + createClaudeSessionResumeState, + createGeminiJsonlSessionResumeState } from './session-scanner-primary-parsers' -import type { SessionFileCandidate } from './session-scanner-types' +import { + createCopilotSessionResumeState, + createCursorSessionResumeState +} from './session-scanner-secondary-parsers' +import type { ResumableSessionParseState, SessionFileCandidate } from './session-scanner-types' // Sized past the default recency cap (1000) plus the in-scope cap (2000) so a // full steady-state result set stays resident between forced rescans. @@ -18,8 +22,8 @@ const MAX_CACHE_ENTRIES = 4096 const NEWLINE_BYTE = 0x0a const CARRIAGE_RETURN_BYTE = 0x0d -type ClaudeResumePoint = { - state: ClaudeSessionParseState +type ResumePoint = { + state: ResumableSessionParseState // Byte offset just past the last complete ('\n'-terminated) line consumed; // a trailing unterminated line is deliberately left before this point. byteOffset: number @@ -30,7 +34,42 @@ type SessionParseCacheEntry = { sizeBytes: number | null platform: NodeJS.Platform session: AiVaultSession | null - claudeResume: ClaudeResumePoint | null + resume: ResumePoint | null +} + +// Incremental append-parsing applies only to transcripts that are append-only +// JSONL line-folds. Whole-JSON documents (grok/rovo/devin/hermes/gemini-json) +// are rewritten in place, Kimi reads a state doc plus a sibling wire file, and +// OpenCode reads SQLite rows or a doc plus a message dir — those formats keep +// unchanged-file reuse only and re-parse whole when they change. +// Returns a factory (not a state) so steady-state resumes, which clone the +// cached state instead, never pay for a throwaway accumulator. +function resumableStateFactoryFor( + candidate: SessionFileCandidate +): (() => ResumableSessionParseState) | null { + switch (candidate.agent) { + case 'claude': + return () => createClaudeSessionResumeState(candidate.file) + case 'codex': + return () => createCodexSessionResumeState(candidate.file, candidate.codexHome) + case 'cursor': + return () => createCursorSessionResumeState(candidate.file) + case 'copilot': + return () => createCopilotSessionResumeState(candidate.file) + case 'droid': + return () => createDroidSessionResumeState(candidate.file) + case 'openclaw': + case 'pi': { + const agent = candidate.agent + return () => createMessageGraphSessionResumeState(agent, candidate.file) + } + case 'gemini': + return candidate.file.path.endsWith('.jsonl') + ? () => createGeminiJsonlSessionResumeState(candidate.file) + : null + default: + return null + } } export type SessionParseStats = { @@ -63,10 +102,12 @@ function storeEntry(path: string, entry: SessionParseCacheEntry): void { /** * Parse a session file, reusing prior work where the file is provably - * unchanged (mtime+size) and, for Claude transcripts, resuming the parse from - * the last consumed byte when the file only grew. This is what keeps the - * renderer's ~5s forced rescans from re-reading gigabytes of transcripts - * (STA-1278: main process pegging one core during multi-agent workloads). + * unchanged (mtime+size) and, for append-only JSONL transcripts (Claude, + * Codex, Cursor, Copilot, Droid, OpenClaw/Pi, Gemini-JSONL), resuming the + * parse from the last consumed byte when the file only grew. This is what + * keeps the renderer's ~5s forced rescans from re-reading gigabytes of + * transcripts (STA-1278/STA-1417: main process pegging one core during + * multi-agent workloads). */ export async function parseAgentSessionFileCached( candidate: SessionFileCandidate, @@ -89,8 +130,15 @@ export async function parseAgentSessionFileCached( return entry.session } - if (candidate.agent === 'claude') { - const parsed = await parseClaudeCandidateWithResume({ candidate, platform, entry, stats }) + const stateFactory = resumableStateFactoryFor(candidate) + if (stateFactory) { + const parsed = await parseResumableCandidate({ + candidate, + platform, + entry, + stats, + stateFactory + }) storeEntry(file.path, parsed) return parsed.session } @@ -105,19 +153,20 @@ export async function parseAgentSessionFileCached( sizeBytes: file.sizeBytes ?? null, platform, session, - claudeResume: null + resume: null }) return session } -async function parseClaudeCandidateWithResume(args: { +async function parseResumableCandidate(args: { candidate: SessionFileCandidate platform: NodeJS.Platform entry: SessionParseCacheEntry | undefined stats?: SessionParseStats + stateFactory: () => ResumableSessionParseState }): Promise { const { file } = args.candidate - const resume = args.entry?.platform === args.platform ? args.entry.claudeResume : null + const resume = args.entry?.platform === args.platform ? args.entry.resume : null const canResume = resume !== null && resume !== undefined && @@ -127,9 +176,7 @@ async function parseClaudeCandidateWithResume(args: { // Clone before consuming: a failed read must not corrupt the cached state, // or the next resume would double-count the lines applied before the error. - const state = canResume - ? cloneClaudeSessionParseState(resume.state) - : createClaudeSessionParseState(file) + const state = canResume ? resume.state.clone() : args.stateFactory() const startOffset = canResume ? resume.byteOffset : 0 if (args.stats) { if (canResume) { @@ -142,30 +189,30 @@ async function parseClaudeCandidateWithResume(args: { const readResult = await consumeCompleteJsonlLines({ path: file.path, start: startOffset, - onLine: (line) => consumeClaudeSessionLine(state, line) + onLine: (line) => state.consumeLine(line) }) if (args.stats) { args.stats.bytesRead += readResult.bytesRead } // The stat this scan displays is current even when nothing new was consumed. - state.accumulator.modifiedAt = file.modifiedAt + state.touchFile(file) // Keep parity with the one-shot parser: a final unterminated line is shown, // but stays out of the resumable state so the (possibly still-growing) line // is re-read once complete instead of being half-counted. let displayState = state if (readResult.trailingPartialLine !== null) { - displayState = cloneClaudeSessionParseState(state) - consumeClaudeSessionLine(displayState, readResult.trailingPartialLine) + displayState = state.clone() + displayState.consumeLine(readResult.trailingPartialLine) } return { mtimeMs: file.mtimeMs, sizeBytes: file.sizeBytes ?? null, platform: args.platform, - session: finalizeClaudeSessionParseState(displayState, args.platform), - claudeResume: { state, byteOffset: readResult.consumedThrough } + session: await displayState.finalize(args.platform), + resume: { state, byteOffset: readResult.consumedThrough } } } diff --git a/src/main/ai-vault/session-scanner-primary-parsers.ts b/src/main/ai-vault/session-scanner-primary-parsers.ts index eeedee917..e07329a30 100644 --- a/src/main/ai-vault/session-scanner-primary-parsers.ts +++ b/src/main/ai-vault/session-scanner-primary-parsers.ts @@ -3,8 +3,13 @@ import { readFile } from 'node:fs/promises' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' -import type { FileWithMtime, SessionAccumulator } from './session-scanner-types' +import type { + FileWithMtime, + ResumableSessionParseState, + SessionAccumulator +} from './session-scanner-types' import { + accumulatorFoldResumeState, addPreviewContent, createAccumulator, finalizeSession, @@ -139,6 +144,23 @@ export function finalizeClaudeSessionParseState( return finalizeSession(snapshot.accumulator, platform, options) } +export function createClaudeSessionResumeState(file: FileWithMtime): ResumableSessionParseState { + return claudeResumeStateFromParseState(createClaudeSessionParseState(file)) +} + +function claudeResumeStateFromParseState( + state: ClaudeSessionParseState +): ResumableSessionParseState { + return { + consumeLine: (line) => consumeClaudeSessionLine(state, line), + clone: () => claudeResumeStateFromParseState(cloneClaudeSessionParseState(state)), + touchFile: (file) => { + state.accumulator.modifiedAt = file.modifiedAt + }, + finalize: (platform, options) => finalizeClaudeSessionParseState(state, platform, options) + } +} + export async function parseClaudeSessionFile( file: FileWithMtime, platform: NodeJS.Platform = process.platform @@ -239,38 +261,47 @@ export async function parseGeminiJsonlSessionFile( return parseGeminiJsonlSessionLines({ file, lines, platform }) } +function consumeGeminiJsonlRecordLine(accumulator: SessionAccumulator, line: string): void { + const record = parseJsonObject(line) + if (!record) { + return + } + const setRecord = asRecord(record.$set) + if (setRecord) { + updateTimeline(accumulator, extractString(setRecord.lastUpdated)) + return + } + const sessionId = extractString(record.sessionId) + if (sessionId) { + accumulator.sessionId = sessionId + } + updateTimeline(accumulator, extractString(record.startTime)) + updateTimeline(accumulator, extractString(record.lastUpdated)) + consumeGeminiMessage(accumulator, record) +} + +// Resumable only for the JSONL log format; Gemini's legacy single-JSON +// session documents are rewritten in place and must be re-read whole. +export function createGeminiJsonlSessionResumeState( + file: FileWithMtime +): ResumableSessionParseState { + return accumulatorFoldResumeState( + createAccumulator({ agent: 'gemini', file, sessionId: sessionIdFromFileName(file.path) }), + consumeGeminiJsonlRecordLine + ) +} + async function parseGeminiJsonlSessionLines(args: { file: FileWithMtime lines: AsyncIterable | Iterable platform: NodeJS.Platform options?: ParserSessionOptions }): Promise { - const accumulator = createAccumulator({ - agent: 'gemini', - file: args.file, - sessionId: sessionIdFromFileName(args.file.path) - }) - + const state = createGeminiJsonlSessionResumeState(args.file) for await (const line of args.lines) { - const record = parseJsonObject(line) - if (!record) { - continue - } - const setRecord = asRecord(record.$set) - if (setRecord) { - updateTimeline(accumulator, extractString(setRecord.lastUpdated)) - continue - } - const sessionId = extractString(record.sessionId) - if (sessionId) { - accumulator.sessionId = sessionId - } - updateTimeline(accumulator, extractString(record.startTime)) - updateTimeline(accumulator, extractString(record.lastUpdated)) - consumeGeminiMessage(accumulator, record) + state.consumeLine(line) } - - return finalizeSession(accumulator, args.platform, args.options) + return state.finalize(args.platform, args.options) } export function consumeGeminiMessage( diff --git a/src/main/ai-vault/session-scanner-secondary-parsers.ts b/src/main/ai-vault/session-scanner-secondary-parsers.ts index 1edcf6292..364c25b5d 100644 --- a/src/main/ai-vault/session-scanner-secondary-parsers.ts +++ b/src/main/ai-vault/session-scanner-secondary-parsers.ts @@ -4,8 +4,13 @@ import { join } from 'node:path' import { createInterface } from 'node:readline' import type { AiVaultSession } from '../../shared/ai-vault-types' import type { ExecutionHostId } from '../../shared/execution-host' -import type { FileWithMtime, SessionAccumulator } from './session-scanner-types' +import type { + FileWithMtime, + ResumableSessionParseState, + SessionAccumulator +} from './session-scanner-types' import { + accumulatorFoldResumeState, addPreviewContent, addPreviewMessage, createAccumulator, @@ -60,70 +65,75 @@ export async function parseCopilotSessionContent( }) } +function consumeCopilotRecordLine(accumulator: SessionAccumulator, line: string): void { + const record = parseJsonObject(line) + if (!record) { + return + } + updateTimeline(accumulator, extractString(record.timestamp)) + const data = asRecord(record.data) + if (record.type === 'session.start' && data) { + const sessionId = extractString(data.sessionId) + if (sessionId) { + accumulator.sessionId = sessionId + } + updateTimeline(accumulator, extractString(data.startTime)) + return + } + if (record.type === 'session.model_change' && data) { + accumulator.model = extractString(data.newModel) ?? accumulator.model + return + } + if (record.type === 'session.info' && data) { + accumulator.cwd = extractTrustedFolder(data.message) ?? accumulator.cwd + return + } + if (record.type === 'user.message' && data) { + accumulator.messageCount++ + accumulator.title ??= normalizeTitleText( + extractString(data.transformedContent) ?? extractString(data.content) ?? '' + ) + addPreviewMessage(accumulator, { + role: 'user', + text: extractString(data.transformedContent) ?? extractString(data.content), + timestamp: record.timestamp + }) + return + } + if (record.type === 'assistant.message' && data) { + accumulator.messageCount++ + addPreviewMessage(accumulator, { + role: 'assistant', + text: extractString(data.content), + timestamp: record.timestamp + }) + return + } + if (record.type === 'session.shutdown' && data) { + accumulator.model = extractString(data.currentModel) ?? accumulator.model + accumulator.totalTokens += numberValue(data.currentTokens) + accumulator.totalTokens += copilotModelMetricsTotal(data.modelMetrics) + } +} + +export function createCopilotSessionResumeState(file: FileWithMtime): ResumableSessionParseState { + return accumulatorFoldResumeState( + createAccumulator({ agent: 'copilot', file, sessionId: sessionIdFromFileName(file.path) }), + consumeCopilotRecordLine + ) +} + async function parseCopilotSessionLines(args: { file: FileWithMtime lines: AsyncIterable | Iterable platform: NodeJS.Platform options?: ParserSessionOptions }): Promise { - const accumulator = createAccumulator({ - agent: 'copilot', - file: args.file, - sessionId: sessionIdFromFileName(args.file.path) - }) - + const state = createCopilotSessionResumeState(args.file) for await (const line of args.lines) { - const record = parseJsonObject(line) - if (!record) { - continue - } - updateTimeline(accumulator, extractString(record.timestamp)) - const data = asRecord(record.data) - if (record.type === 'session.start' && data) { - const sessionId = extractString(data.sessionId) - if (sessionId) { - accumulator.sessionId = sessionId - } - updateTimeline(accumulator, extractString(data.startTime)) - continue - } - if (record.type === 'session.model_change' && data) { - accumulator.model = extractString(data.newModel) ?? accumulator.model - continue - } - if (record.type === 'session.info' && data) { - accumulator.cwd = extractTrustedFolder(data.message) ?? accumulator.cwd - continue - } - if (record.type === 'user.message' && data) { - accumulator.messageCount++ - accumulator.title ??= normalizeTitleText( - extractString(data.transformedContent) ?? extractString(data.content) ?? '' - ) - addPreviewMessage(accumulator, { - role: 'user', - text: extractString(data.transformedContent) ?? extractString(data.content), - timestamp: record.timestamp - }) - continue - } - if (record.type === 'assistant.message' && data) { - accumulator.messageCount++ - addPreviewMessage(accumulator, { - role: 'assistant', - text: extractString(data.content), - timestamp: record.timestamp - }) - continue - } - if (record.type === 'session.shutdown' && data) { - accumulator.model = extractString(data.currentModel) ?? accumulator.model - accumulator.totalTokens += numberValue(data.currentTokens) - accumulator.totalTokens += copilotModelMetricsTotal(data.modelMetrics) - } + state.consumeLine(line) } - - return finalizeSession(accumulator, args.platform, args.options) + return state.finalize(args.platform, args.options) } export async function parseCursorSessionFile( @@ -151,40 +161,45 @@ export async function parseCursorSessionContent( }) } +function consumeCursorRecordLine(accumulator: SessionAccumulator, line: string): void { + const record = parseJsonObject(line) + if (!record) { + return + } + updateTimeline(accumulator, extractString(record.timestamp)) + const role = extractString(record.role) + if (role === 'user' || role === 'assistant') { + accumulator.messageCount++ + if (role === 'user') { + accumulator.title ??= extractMessageText(record.message) ?? extractContentText(record.content) + } + addPreviewContent( + accumulator, + role, + asRecord(record.message)?.content ?? record.content, + record.timestamp + ) + } +} + +export function createCursorSessionResumeState(file: FileWithMtime): ResumableSessionParseState { + return accumulatorFoldResumeState( + createAccumulator({ agent: 'cursor', file, sessionId: sessionIdFromFileName(file.path) }), + consumeCursorRecordLine + ) +} + async function parseCursorSessionLines(args: { file: FileWithMtime lines: AsyncIterable | Iterable platform: NodeJS.Platform options?: ParserSessionOptions }): Promise { - const accumulator = createAccumulator({ - agent: 'cursor', - file: args.file, - sessionId: sessionIdFromFileName(args.file.path) - }) - + const state = createCursorSessionResumeState(args.file) for await (const line of args.lines) { - const record = parseJsonObject(line) - if (!record) { - continue - } - updateTimeline(accumulator, extractString(record.timestamp)) - const role = extractString(record.role) - if (role === 'user' || role === 'assistant') { - accumulator.messageCount++ - if (role === 'user') { - accumulator.title ??= - extractMessageText(record.message) ?? extractContentText(record.content) - } - addPreviewContent( - accumulator, - role, - asRecord(record.message)?.content ?? record.content, - record.timestamp - ) - } + state.consumeLine(line) } - return finalizeSession(accumulator, args.platform, args.options) + return state.finalize(args.platform, args.options) } export async function parseOpenCodeSessionFile( diff --git a/src/main/ai-vault/session-scanner-token-values.ts b/src/main/ai-vault/session-scanner-token-values.ts index 83b288140..e2621b9f4 100644 --- a/src/main/ai-vault/session-scanner-token-values.ts +++ b/src/main/ai-vault/session-scanner-token-values.ts @@ -104,3 +104,16 @@ export function subtractCodexUsage( export function numberValue(value: unknown): number { return typeof value === 'number' && Number.isFinite(value) ? value : 0 } + +export function addCodexUsage( + base: CodexUsageSnapshot, + increment: CodexUsageSnapshot +): CodexUsageSnapshot { + return { + inputTokens: base.inputTokens + increment.inputTokens, + cachedInputTokens: base.cachedInputTokens + increment.cachedInputTokens, + outputTokens: base.outputTokens + increment.outputTokens, + reasoningOutputTokens: base.reasoningOutputTokens + increment.reasoningOutputTokens, + totalTokens: base.totalTokens + increment.totalTokens + } +} diff --git a/src/main/ai-vault/session-scanner-types.ts b/src/main/ai-vault/session-scanner-types.ts index 56ec69717..7c6c38a0b 100644 --- a/src/main/ai-vault/session-scanner-types.ts +++ b/src/main/ai-vault/session-scanner-types.ts @@ -64,6 +64,26 @@ export type SessionParseResult = { issue: AiVaultScanIssue | null } +export type ResumableParseFinalizeOptions = { + executionHostId?: ExecutionHostId + executionHostPlatform?: NodeJS.Platform | null +} + +// One in-progress parse of an append-only transcript, resumable across scans. +// The parse cache stores a state per file and feeds it only newly appended +// lines; `clone` must deep-copy anything `consumeLine` mutates so a failed +// read or a display-only trailing line can never corrupt the cached fold. +export type ResumableSessionParseState = { + consumeLine(line: string): void + clone(): ResumableSessionParseState + // Refresh per-scan file metadata (mtime display string) without re-parsing. + touchFile(file: FileWithMtime): void + finalize( + platform: NodeJS.Platform, + options?: ResumableParseFinalizeOptions + ): Promise | AiVaultSession | null +} + export type SessionAccumulator = { agent: AiVaultAgent sessionId: string diff --git a/src/main/ai-vault/session-scanner-values.ts b/src/main/ai-vault/session-scanner-values.ts index 8a5a76306..9733175e3 100644 --- a/src/main/ai-vault/session-scanner-values.ts +++ b/src/main/ai-vault/session-scanner-values.ts @@ -154,6 +154,7 @@ export function errorMessage(err: unknown): string { } export { + addCodexUsage, claudeUsageTotal, copilotModelMetricsTotal, normalizeCodexUsage,