xiaoxue_admin
/
orca
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

perf(ai-vault): extend incremental transcript parsing to all append-only JSONL agents (STA-1417) (#7593) 

Co-authored-by: Orca <help@stably.ai>
This commit is contained in:
Brennan Benson 2026-07-06 15:13:00 -07:00 committed by GitHub
parent 417723411e
commit 16439fd499
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
14 changed files with 1294 additions and 431 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
src/main/ai-vault/session-scanner-accumulator.ts
View File

@ -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,

92
src/main/ai-vault/session-scanner-codex-fixtures.ts Normal file
View File

@ -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, unknown>): 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

439
src/main/ai-vault/session-scanner-codex-parser.ts
View File

@ -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<string, string>
}
const codexSessionIndexTitleCache = new Map<string, Promise<CodexSessionIndexTitleCacheEntry>>()
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<string | null>
executionHostId?: ExecutionHostId
executionHostPlatform?: NodeJS.Platform | null
}
): Promise<AiVaultSession | null> {
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<string | null>
): 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<string> | Iterable<string>
@ -83,142 +288,22 @@ async function parseCodexSessionLines(args: {
executionHostPlatform?: NodeJS.Platform | null
titleReader?: (sessionId: string) => Promise<string | null>
}): Promise<AiVaultSession | null> {
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, unknown>): string | null {
return extractString(payload.thread_source) ?? extractString(payload.threadSource)
}
@ -240,79 +325,3 @@ function extractCodexSessionMetadataTitle(payload: Record<string, unknown>): str
normalizeTitleText(extractString(payload.threadName) ?? '')
)
}
async function readCodexSessionIndexTitle(
sessionFilePath: string,
codexHome: string | null,
sessionId: string
): Promise<string | null> {
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<Map<string, string>> {
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<Map<string, string>> {
const titleBySessionId = new Map<string, string>()
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
}

93
src/main/ai-vault/session-scanner-codex-title-index.ts Normal file
View File

@ -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 <CODEX_HOME>/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<string, string>
}
const codexSessionIndexTitleCache = new Map<string, Promise<CodexSessionIndexTitleCacheEntry>>()
export async function readCodexSessionIndexTitle(
sessionFilePath: string,
codexHome: string | null,
sessionId: string
): Promise<string | null> {
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<Map<string, string>> {
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<Map<string, string>> {
const titleBySessionId = new Map<string, string>()
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
}

88
src/main/ai-vault/session-scanner-droid-parser.ts
View File

@ -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<string> | Iterable<string>
platform: NodeJS.Platform
options?: ParserSessionOptions
}): Promise<AiVaultSession | null> {
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(

95
src/main/ai-vault/session-scanner-graph-parsers.ts
View File

@ -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<AiVaultSession | null> {
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)
}

240
src/main/ai-vault/session-scanner-incremental-fixtures.ts Normal file
View File

@ -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<string, unknown>, 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()
]
}

250
src/main/ai-vault/session-scanner-parse-cache-agents.test.ts Normal file
View File

@ -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<string> {
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<SessionFileCandidate> {
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)
)
})
})

101
src/main/ai-vault/session-scanner-parse-cache.ts
View File

@ -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<SessionParseCacheEntry> {
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 }
}
}

81
src/main/ai-vault/session-scanner-primary-parsers.ts
View File

@ -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<string> | Iterable<string>
platform: NodeJS.Platform
options?: ParserSessionOptions
}): Promise<AiVaultSession | null> {
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(

181
src/main/ai-vault/session-scanner-secondary-parsers.ts
View File

@ -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<string> | Iterable<string>
platform: NodeJS.Platform
options?: ParserSessionOptions
}): Promise<AiVaultSession | null> {
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<string> | Iterable<string>
platform: NodeJS.Platform
options?: ParserSessionOptions
}): Promise<AiVaultSession | null> {
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(

13
src/main/ai-vault/session-scanner-token-values.ts
View File

@ -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
}
}

20
src/main/ai-vault/session-scanner-types.ts
View File

@ -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> | AiVaultSession | null
}
export type SessionAccumulator = {
agent: AiVaultAgent
sessionId: string

1
src/main/ai-vault/session-scanner-values.ts
View File

@ -154,6 +154,7 @@ export function errorMessage(err: unknown): string {
}
export {
addCodexUsage,
claudeUsageTotal,
copilotModelMetricsTotal,
normalizeCodexUsage,