From 7f492dc15b4bde2383697b596a835d8dae1008ec Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 15 Oct 2026 13:45:27 -0700 Subject: [PATCH] fix(native-chat): resume large terminal conversations in a new native chat (#29056) * fix(native-chat): resume conversations of any size in a new native chat Resume in New Native Chat refused any saved conversation over 16 MiB, so long Claude and Codex sessions could not move into a chat. The import now streams the file, bounds each message as it is read, and keeps the newest messages within the same 16 MiB budget; one line is still bounded at 16 MiB. The agent resumes the full conversation itself, so only the chat's earliest messages can fall outside the window. * test(native-chat): adopt preflight refuses one line past the record bound, not a large file * fix(native-chat): bound retained Pi import state and identities * fix(native-chat): keep Pi branch projection responsive * fix(native-chat): classify temporary Pi import table * fix(native-chat): keep Pi resume independent of import scratch cleanup * test(native-chat): match legacy Pi identity by shape --- ...journal-legacy-import-record-bound.test.ts | 41 +++ ...journal-legacy-import-source-bound.test.ts | 60 ---- .../journal-legacy-import.test.ts | 88 ++++-- .../journal-legacy-import.ts | 182 +++++++----- .../journal-pi-active-branch.ts | 265 ++++++++++++------ .../journal-pi-context-import.spec.ts | 119 +++++++- ...tured-agent-session-adopted-import.test.ts | 6 +- .../native-chat/transcript-stream-lines.ts | 12 +- .../wsl-move/wsl-move-storage-dispositions.ts | 6 + 9 files changed, 530 insertions(+), 249 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-legacy-import-record-bound.test.ts delete mode 100644 src/main/native-chat/agent-session-journal/journal-legacy-import-source-bound.test.ts diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import-record-bound.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import-record-bound.test.ts new file mode 100644 index 00000000000..515429eea1e --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import-record-bound.test.ts @@ -0,0 +1,41 @@ +import { createReadStream } from 'node:fs' +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { prepareLegacyTranscriptImport } from './journal-legacy-import' + +vi.mock(import('node:fs'), async (importOriginal) => { + const original = await importOriginal() + return { ...original, createReadStream: vi.fn(original.createReadStream) } +}) + +const RECORD_LIMIT_BYTES = 16 * 1024 * 1024 +const roots: string[] = [] + +afterEach(async () => { + vi.clearAllMocks() + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))) +}) + +describe('legacy import record byte bound', () => { + it('refuses a single line past the limit and closes the stream', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-legacy-import-bound-')) + roots.push(root) + const filePath = join(root, 'oversized-line.jsonl') + await writeFile(filePath, ' '.repeat(RECORD_LIMIT_BYTES + 1)) + + const result = await prepareLegacyTranscriptImport({ + agent: 'claude', + sessionId: 'record-bound', + options: { filePath } + }) + + expect(result).toEqual({ + ok: false, + error: expect.stringContaining(`${RECORD_LIMIT_BYTES} byte limit`) + }) + expect(createReadStream).toHaveBeenCalledOnce() + expect(vi.mocked(createReadStream).mock.results[0]?.value.destroyed).toBe(true) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import-source-bound.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import-source-bound.test.ts deleted file mode 100644 index 1ffc846e10e..00000000000 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import-source-bound.test.ts +++ /dev/null @@ -1,60 +0,0 @@ -import { createReadStream } from 'node:fs' -import { appendFile, mkdtemp, rm, stat, writeFile } from 'node:fs/promises' -import type * as FsPromises from 'node:fs/promises' -import { tmpdir } from 'node:os' -import { join } from 'node:path' -import { afterEach, describe, expect, it, vi } from 'vitest' -import { prepareLegacyTranscriptImport } from './journal-legacy-import' - -vi.mock(import('node:fs'), async (importOriginal) => { - const original = await importOriginal() - return { ...original, createReadStream: vi.fn(original.createReadStream) } -}) - -vi.mock(import('node:fs/promises'), async (importOriginal) => { - const original = await importOriginal() - return { ...original, stat: vi.fn() } -}) - -const SOURCE_LIMIT_BYTES = 16 * 1024 * 1024 -const roots: string[] = [] - -afterEach(async () => { - vi.clearAllMocks() - await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))) -}) - -describe('legacy import source byte bound', () => { - it('refuses a source that grows past the limit after stat and closes the stream', async () => { - const root = await mkdtemp(join(tmpdir(), 'orca-legacy-import-bound-')) - roots.push(root) - const filePath = join(root, 'growing.jsonl') - await writeFile( - filePath, - `${JSON.stringify({ - type: 'assistant', - uuid: 'first-message', - message: { role: 'assistant', content: [{ type: 'text', text: 'Keep the prior journal' }] } - })}\n` - ) - const actualFs = await vi.importActual('node:fs/promises') - const beforeGrowth = await actualFs.stat(filePath) - vi.mocked(stat).mockImplementationOnce(async () => { - await appendFile(filePath, Buffer.alloc(SOURCE_LIMIT_BYTES, 0x20)) - return beforeGrowth - }) - - const result = await prepareLegacyTranscriptImport({ - agent: 'claude', - sessionId: 'source-bound', - options: { filePath } - }) - - expect(result).toEqual({ - ok: false, - error: expect.stringContaining(`${SOURCE_LIMIT_BYTES} byte limit`) - }) - expect(createReadStream).toHaveBeenCalledOnce() - expect(vi.mocked(createReadStream).mock.results[0]?.value.destroyed).toBe(true) - }) -}) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts index 9da96daf490..993c1531303 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts @@ -13,7 +13,10 @@ import type { AgentSessionProviderHandle } from '../../../shared/agent-session-p import type { NativeChatMessage } from '../../../shared/native-chat-types' import type * as TranscriptLineDecoders from '../transcript-line-decoders' import { createLegacyIdentityTracker } from './journal-legacy-identity' -import { importLegacyTranscriptIntoJournal } from './journal-legacy-import' +import { + importLegacyTranscriptIntoJournal, + prepareLegacyTranscriptImport +} from './journal-legacy-import' import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' import { openAgentSessionJournal } from './journal-store-factory' import type { AgentSessionJournal } from './journal-store' @@ -486,29 +489,72 @@ describe('payload bounds on import', () => { }) describe('import failures', () => { - it('rejects a legacy source above the fixed 16 MiB import cap before decoding', async () => { - const journalDir = join(root, 'oversized-source-journal') - const journal = await open('claude', CLAUDE_SESSION, { - database: openTestJournalHostDatabase(journalDir) - }) - const filePath = join(root, 'oversized-source.jsonl') - await writeFile(filePath, 'x'.repeat(16 * 1024 * 1024 + 1), 'utf8') - const epoch = journal.epoch - - await expect( - importLegacyTranscriptIntoJournal({ - journal, - agent: 'claude', + it('charges retained identity metadata against the history budget', async () => { + const filePath = join(root, 'large-identities.jsonl') + const longId = 'i'.repeat(80 * 1024) + const lines = Array.from({ length: 250 }, (_, index) => + JSON.stringify({ + type: 'user', + uuid: `${index}-${longId}`, sessionId: CLAUDE_SESSION, - fence: 1, - options: { filePath } + message: { role: 'user', content: 'small' } }) - ).resolves.toMatchObject({ - ok: false, - error: `Legacy transcript exceeds the ${16 * 1024 * 1024}-byte import bound` + ) + await writeFile(filePath, `${lines.join('\n')}\n`, 'utf8') + + const result = await prepareLegacyTranscriptImport({ + agent: 'claude', + sessionId: CLAUDE_SESSION, + options: { filePath } }) - expect(journal.epoch).toBe(epoch) - expect(journal.snapshot().items).toEqual([]) + expect(result.ok).toBe(true) + if (!result.ok) { + throw new Error(result.error) + } + expect(result.items.length).toBeLessThan(lines.length) + expect( + result.items.reduce((total, item) => total + Buffer.byteLength(JSON.stringify(item)), 0) + ).toBeLessThanOrEqual(16 * 1024 * 1024) + expect(result.items.at(-1)?.identity).toMatchObject({ + provider: 'claude', + uuid: `249-${longId}` + }) + }) + + it('imports a source past 16 MiB, keeping its newest messages', async () => { + const journal = await open('claude', CLAUDE_SESSION, { + database: openTestJournalHostDatabase(join(root, 'large-source-journal')) + }) + const filePath = join(root, 'large-source.jsonl') + const filler = 'x'.repeat(17 * 1024) + const lines = Array.from({ length: 1_100 }, (_, index) => + JSON.stringify({ + type: 'assistant', + uuid: `message-${index}`, + message: { role: 'assistant', content: [{ type: 'text', text: `${index} ${filler}` }] } + }) + ) + await writeFile(filePath, `${lines.join('\n')}\n`, 'utf8') + + const result = await importLegacyTranscriptIntoJournal({ + journal, + agent: 'claude', + sessionId: CLAUDE_SESSION, + fence: 1, + options: { filePath } + }) + + expect(result).toMatchObject({ ok: true, replaced: true }) + const texts = journal + .snapshot() + .items.map((item) => + item.body.kind === 'message' && item.body.blocks[0]?.type === 'text' + ? item.body.blocks[0].text.split(' ')[0] + : null + ) + expect(texts.at(-1)).toBe('1099') + expect(texts.length).toBeGreaterThan(900) + expect(texts.length).toBeLessThan(lines.length) }) it('bounds oversized legacy tool-call input before journal publication', async () => { diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts index e69abd28a67..389ad083f9b 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts @@ -11,7 +11,6 @@ // writing; a later structured resume rolls the epoch again and rebuilds. import { createReadStream } from 'node:fs' -import { stat } from 'node:fs/promises' import type { AgentType } from '../../../shared/agent-status-types' import type { AgentJournalCursor, @@ -48,7 +47,11 @@ export type LegacyImportOptions = ResolveSessionFileOptions & { decodedMessageIdentities?: true } -const MAX_LEGACY_IMPORT_SOURCE_BYTES = 16 * 1024 * 1024 +/** One transcript line; the resume reconciler reads the same file under the same bound. */ +const MAX_LEGACY_IMPORT_RECORD_BYTES = 16 * 1024 * 1024 +/** Bounded history kept from a source of any size. The agent resumes the whole conversation + * itself, so only the chat's earliest messages fall outside this. */ +const MAX_LEGACY_IMPORT_RETAINED_BYTES = 16 * 1024 * 1024 /** A roster is a status list; an imported one is as untrusted as any other block. */ const MAX_LEGACY_IMPORT_SUBAGENTS = 64 @@ -105,38 +108,21 @@ export async function prepareLegacyTranscriptImport(input: { return { ok: false, error: `No transcript found for ${input.agent} session ${input.sessionId}` } } - // Refuse an oversized source before decoding any prefix. Importing a prefix - // would make the restored timeline look complete while silently omitting - // later records; callers can retry after reducing the source or quota. try { - const sourceBytes = (await stat(filePath)).size - if (sourceBytes > MAX_LEGACY_IMPORT_SOURCE_BYTES) { - return { - ok: false, - error: `Legacy transcript exceeds the ${MAX_LEGACY_IMPORT_SOURCE_BYTES}-byte import bound` - } + return { + ok: true, + items: await decodeReplacementItems({ + filePath, + transcriptAgent, + agent: input.agent, + sessionId: input.sessionId, + decodedMessageIdentities: options.decodedMessageIdentities, + limits + }) } } catch (err) { return { ok: false, error: err instanceof Error ? err.message : String(err) } } - - let decoded: { messages: NativeChatMessage[]; identities: AgentJournalItemIdentity[] } - try { - decoded = await decodeWithIdentities({ - filePath, - transcriptAgent, - agent: input.agent, - sessionId: input.sessionId, - decodedMessageIdentities: options.decodedMessageIdentities - }) - } catch (err) { - return { ok: false, error: err instanceof Error ? err.message : String(err) } - } - - if (decoded.identities.length !== decoded.messages.length) { - return { ok: false, error: 'Legacy transcript identity coverage is incomplete' } - } - return { ok: true, items: replacementItems(decoded.messages, decoded.identities, limits) } } /** OpenCode keeps conversations in a database (`filePath` names it), read by the reader its @@ -172,20 +158,27 @@ function replacementItems( const replacement: JournalReplacementItem[] = [] for (const [index, message] of messages.entries()) { const identity = identities[index] - if (!identity) { - continue + if (identity) { + replacement.push(replacementItem(message, identity, limits)) } - replacement.push({ - identity, - body: legacyItemBody(message, limits), - observedAt: message.timestamp ?? undefined, - // Imported history is no generation's work: never current, whatever it says still runs. - ownerFence: 0 - }) } return replacement } +function replacementItem( + message: NativeChatMessage, + identity: AgentJournalItemIdentity, + limits: JournalPayloadLimits +): JournalReplacementItem { + return { + identity, + body: legacyItemBody(message, limits), + observedAt: message.timestamp ?? undefined, + // Imported history is no generation's work: never current, whatever it says still runs. + ownerFence: 0 + } +} + const TRANSCRIPT_DECODERS = { claude: decodeClaudeTranscriptLine, codex: decodeCodexTranscriptLine, @@ -193,60 +186,99 @@ const TRANSCRIPT_DECODERS = { omp: decodeOmpTranscriptLine } as const -/** Run the real decoder while recording an identity anchor per emitted message, - * index-aligned with `messages`. */ -async function decodeWithIdentities(input: { +/** Run the real decoder while recording an identity anchor per emitted message. Each message is + * bounded as it is read and only the newest fit the retained budget, so memory tracks what is + * kept, never the source's size. */ +async function decodeReplacementItems(input: { filePath: string transcriptAgent: keyof typeof TRANSCRIPT_DECODERS agent: AgentType sessionId: string decodedMessageIdentities?: true -}): Promise<{ messages: NativeChatMessage[]; identities: AgentJournalItemIdentity[] }> { + limits: JournalPayloadLimits +}): Promise { const tracker = createLegacyIdentityTracker({ transcriptAgent: input.transcriptAgent, agent: input.agent, sessionId: input.sessionId }) const decode = TRANSCRIPT_DECODERS[input.transcriptAgent] - const identities: AgentJournalItemIdentity[] = [] - const pi = input.agent === 'pi' ? createPiTranscriptCollector() : null + const toItem = (message: NativeChatMessage, identity: AgentJournalItemIdentity) => + replacementItem(message, identity, input.limits) + const retained = createNewestItemWindow(MAX_LEGACY_IMPORT_RETAINED_BYTES) + const pi = + input.agent === 'pi' + ? createPiTranscriptCollector(toItem, (line, lineIndex) => tracker.identify(line, lineIndex)) + : null let lineIndex = 0 - - // Count raw bytes while reading: the source can grow after the stat check. - const stream = createReadStream(input.filePath) - const { messages } = await decodeTranscriptStream( - stream, - input.filePath, - 0, - (line, fallbackId) => { - const trackedIdentity = tracker.identify(line, lineIndex) - lineIndex += 1 - if (pi) { - pi.record(line, fallbackId, trackedIdentity) + try { + await decodeTranscriptStream( + createReadStream(input.filePath), + input.filePath, + 0, + (line, fallbackId) => { + const currentLineIndex = lineIndex + lineIndex += 1 + if (pi) { + pi.record(line, fallbackId, currentLineIndex) + return null + } + const message = decode(line, fallbackId) + if (message) { + retained.push( + toItem( + message, + input.decodedMessageIdentities + ? { + provider: 'legacy', + agent: input.agent, + sessionId: input.sessionId, + recordId: message.id + } + : tracker.identify(line, currentLineIndex) + ) + ) + } return null + }, + true, + MAX_LEGACY_IMPORT_RECORD_BYTES + ) + if (pi) { + await pi.project(retained.push) + } + return retained.items() + } finally { + pi?.close() + } +} + +/** The newest complete items that fit `maxBytes`; the latest item is always kept. */ +function createNewestItemWindow(maxBytes: number): { + push(item: JournalReplacementItem): void + items(): JournalReplacementItem[] +} { + let entries: ({ item: JournalReplacementItem; bytes: number } | undefined)[] = [] + let head = 0 + let bytes = 0 + return { + push(item) { + const itemBytes = Buffer.byteLength(JSON.stringify(item), 'utf8') + entries.push({ item, bytes: itemBytes }) + bytes += itemBytes + while (bytes > maxBytes && head < entries.length - 1) { + bytes -= entries[head]?.bytes ?? 0 + entries[head] = undefined + head += 1 } - const message = decode(line, fallbackId) - if (message) { - identities.push( - input.decodedMessageIdentities - ? { - provider: 'legacy', - agent: input.agent, - sessionId: input.sessionId, - recordId: message.id - } - : trackedIdentity - ) + // Drop evicted entries in batches so a long source stays linear. + if (head > 1024 && head * 2 > entries.length) { + entries = entries.slice(head) + head = 0 } - return message }, - true, - MAX_LEGACY_IMPORT_SOURCE_BYTES - ) - if (pi) { - return pi.project() + items: () => entries.slice(head).flatMap((entry) => (entry ? [entry.item] : [])) } - return { messages, identities } } /** diff --git a/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts b/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts index 7940fa70f9c..fb0e723f41e 100644 --- a/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts +++ b/src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts @@ -1,62 +1,213 @@ +import { performance } from 'node:perf_hooks' import type { AgentJournalItemIdentity } from '../../../shared/agent-session-journal-types' import type { NativeChatMessage } from '../../../shared/native-chat-types' import { asRecord, parseJsonObject, timestampMs } from '../../ai-vault/session-scanner-values' +import Database from '../../sqlite/sync-database' import { decodeOmpTranscriptLine } from '../transcript-line-decoders-omp' -type PiTranscriptBranchEntry = { - id: string - parentId: string | null - type: unknown - firstKeptEntryId: string | null - message: NativeChatMessage | null - identity: AgentJournalItemIdentity +function closePiImportDatabase(db: Database.Database | null): void { + try { + db?.close() + } catch (error) { + console.warn('[journal-pi-import] temporary database close failed:', error) + } } -export function createPiTranscriptCollector(): { - record(line: string, fallbackId: string, identity: AgentJournalItemIdentity): void - project(): { messages: NativeChatMessage[]; identities: AgentJournalItemIdentity[] } +/** Keep the complete Pi tree on disk until its saved leaf identifies the active branch. */ +export function createPiTranscriptCollector( + toItem: (message: NativeChatMessage, identity: AgentJournalItemIdentity) => T, + identify: (line: string, lineIndex: number) => AgentJournalItemIdentity +): { + record(line: string, fallbackId: string, lineIndex: number): void + project(keep: (item: T) => void): Promise + close(): void } { - const entries: PiTranscriptBranchEntry[] = [] - const fileEntryIds: (string | null)[] = [] + let db: Database.Database + let opened: Database.Database | null = null + let insert: Database.Statement + let byFileIndex: Database.Statement + try { + opened = new Database('') + db = opened + db.exec(` + PRAGMA journal_mode = OFF; + PRAGMA synchronous = OFF; + PRAGMA temp_store = FILE; + PRAGMA cache_size = -4096; + CREATE TABLE entries ( + id TEXT PRIMARY KEY, + file_index INTEGER NOT NULL UNIQUE, + parent_id TEXT, + type TEXT, + first_kept_id TEXT, + raw_line TEXT NOT NULL, + fallback_id TEXT NOT NULL, + line_index INTEGER NOT NULL, + branch_depth INTEGER + ); + CREATE INDEX entries_branch_depth ON entries(branch_depth) WHERE branch_depth IS NOT NULL; + `) + insert = db.prepare(`INSERT INTO entries + (id, file_index, parent_id, type, first_kept_id, raw_line, fallback_id, line_index) + VALUES (?, ?, ?, ?, ?, ?, ?, ?)`) + byFileIndex = db.prepare('SELECT id FROM entries WHERE file_index = ?') + db.exec('BEGIN') + } catch (error) { + closePiImportDatabase(opened) + throw error + } + let fileIndex = 0 let version = 1 + let leafId: string | null = null + let closed = false return { - record(line, fallbackId, identity) { + record(line, fallbackId, lineIndex) { const record = parseJsonObject(line) if (!record) { return } if (record.type === 'session') { version = typeof record.version === 'number' ? record.version : 1 - fileEntryIds.push(null) + fileIndex += 1 return } - const id = version < 2 ? `#${fileEntryIds.length}` : record.id - const parentId = version < 2 ? (entries.at(-1)?.id ?? null) : record.parentId + const id = version < 2 ? `#${fileIndex}` : record.id + const parentId = version < 2 ? leafId : record.parentId if (typeof id !== 'string' || !(parentId === null || typeof parentId === 'string')) { throw new Error('Pi session entry has no valid branch identity') } - // Pi v1 indexes include the header and parsed metadata, but not malformed lines. - const firstKeptEntryId = + // Pi v1 indexes include headers and parsed metadata, but not malformed lines. + const keptByIndex = version < 2 && typeof record.firstKeptEntryIndex === 'number' - ? fileEntryIds[record.firstKeptEntryIndex] - : record.firstKeptEntryId - const message = - record.type === 'compaction' || record.type === 'branch_summary' - ? piSummaryMessage(record, fallbackId) - : asRecord(record.message)?.role === 'system' + ? byFileIndex.get(record.firstKeptEntryIndex) + : undefined + const firstKeptId = + keptByIndex && typeof keptByIndex.id === 'string' + ? keptByIndex.id + : version < 2 && typeof record.firstKeptEntryIndex === 'number' ? null - : decodeOmpTranscriptLine(line, fallbackId) - entries.push({ - id, - parentId, - type: record.type, - firstKeptEntryId: typeof firstKeptEntryId === 'string' ? firstKeptEntryId : null, - message, - identity - }) - fileEntryIds.push(id) + : record.firstKeptEntryId + try { + insert.run( + id, + fileIndex, + parentId, + typeof record.type === 'string' ? record.type : null, + typeof firstKeptId === 'string' ? firstKeptId : null, + line, + fallbackId, + lineIndex + ) + } catch (error) { + if ( + error instanceof Error && + error.message.includes('UNIQUE constraint failed: entries.id') + ) { + throw new Error('Pi session has duplicate branch identities') + } + throw error + } + leafId = id + fileIndex += 1 + if (fileIndex % 1024 === 0) { + db.exec('COMMIT; BEGIN') + } }, - project: () => selectActivePiBranch(entries) + async project(keep) { + db.exec('COMMIT') + if (leafId === null) { + return + } + const byId = db.prepare( + 'SELECT parent_id, type, first_kept_id, branch_depth FROM entries WHERE id = ?' + ) + const mark = db.prepare('UPDATE entries SET branch_depth = ? WHERE id = ?') + let currentId: string | null = leafId + let depth = 0 + let compactionDepth: number | null = null + let firstKeptId: string | null = null + db.exec('BEGIN') + while (currentId !== null) { + const row = byId.get(currentId) + if (!row) { + break + } + if (row.branch_depth !== null) { + throw new Error('Pi session branch contains a cycle') + } + mark.run(depth, currentId) + if (compactionDepth === null && row.type === 'compaction') { + compactionDepth = depth + firstKeptId = typeof row.first_kept_id === 'string' ? row.first_kept_id : null + } + currentId = typeof row.parent_id === 'string' && row.parent_id ? row.parent_id : null + depth += 1 + if (depth % 1024 === 0) { + await new Promise((resolve) => setImmediate(resolve)) + } + } + db.exec('COMMIT') + const selected = db.prepare(`SELECT raw_line, fallback_id, line_index FROM entries + WHERE branch_depth IS NOT NULL AND branch_depth >= ? AND branch_depth <= ? + ORDER BY branch_depth DESC`) + const emit = async (from: number, through: number): Promise => { + let processedRows = 0 + let processedBytes = 0 + let sliceStartedAt = performance.now() + for (const row of selected.iterate(from, through)) { + if ( + typeof row.raw_line !== 'string' || + typeof row.fallback_id !== 'string' || + typeof row.line_index !== 'number' + ) { + throw new Error('Pi session branch index is unreadable') + } + processedRows += 1 + processedBytes += Buffer.byteLength(row.raw_line, 'utf8') + const record = parseJsonObject(row.raw_line) + if (record) { + const message = + record.type === 'compaction' || record.type === 'branch_summary' + ? piSummaryMessage(record, row.fallback_id) + : asRecord(record.message)?.role === 'system' + ? null + : decodeOmpTranscriptLine(row.raw_line, row.fallback_id) + if (message) { + keep(toItem(message, identify(row.raw_line, row.line_index))) + } + } + if ( + processedBytes >= 1024 * 1024 || + performance.now() - sliceStartedAt >= 8 || + processedRows >= 256 + ) { + await new Promise((resolve) => setImmediate(resolve)) + processedRows = 0 + processedBytes = 0 + sliceStartedAt = performance.now() + } + } + } + if (compactionDepth === null) { + await emit(0, depth - 1) + return + } + await emit(compactionDepth, compactionDepth) + const kept = firstKeptId === null ? undefined : byId.get(firstKeptId) + if (typeof kept?.branch_depth === 'number' && kept.branch_depth > compactionDepth) { + await emit(compactionDepth + 1, kept.branch_depth) + } + if (compactionDepth > 0) { + await emit(0, compactionDepth - 1) + } + }, + close() { + if (closed) { + return + } + closed = true + closePiImportDatabase(db) + } } } @@ -77,47 +228,3 @@ function piSummaryMessage( source: 'transcript' } } - -/** Match Pi's fresh-session TUI: persisted leaf ancestry, then buildContextEntries. */ -function selectActivePiBranch(entries: readonly PiTranscriptBranchEntry[]): { - messages: NativeChatMessage[] - identities: AgentJournalItemIdentity[] -} { - const byId = new Map(entries.map((entry) => [entry.id, entry])) - if (byId.size !== entries.length) { - throw new Error('Pi session has duplicate branch identities') - } - const branch: PiTranscriptBranchEntry[] = [] - const visited = new Set() - let current = entries.at(-1) - while (current) { - if (visited.has(current.id)) { - throw new Error('Pi session branch contains a cycle') - } - visited.add(current.id) - branch.push(current) - current = current.parentId ? byId.get(current.parentId) : undefined - } - const messages: NativeChatMessage[] = [] - const identities: AgentJournalItemIdentity[] = [] - for (const entry of compactedPiBranch(branch.toReversed())) { - if (entry.message) { - messages.push(entry.message) - identities.push(entry.identity) - } - } - return { messages, identities } -} - -function compactedPiBranch(branch: PiTranscriptBranchEntry[]): PiTranscriptBranchEntry[] { - const compactionIndex = branch.findLastIndex((entry) => entry.type === 'compaction') - const compaction = branch[compactionIndex] - if (!compaction) { - return branch - } - const keptIndex = branch.findIndex((entry) => entry.id === compaction.firstKeptEntryId) - // Pi searches for firstKeptEntryId only before the latest compaction. - const kept = - keptIndex !== -1 && keptIndex < compactionIndex ? branch.slice(keptIndex, compactionIndex) : [] - return [compaction, ...kept, ...branch.slice(compactionIndex + 1)] -} diff --git a/src/main/native-chat/agent-session-journal/journal-pi-context-import.spec.ts b/src/main/native-chat/agent-session-journal/journal-pi-context-import.spec.ts index c99edd2f99d..8ec20876e76 100644 --- a/src/main/native-chat/agent-session-journal/journal-pi-context-import.spec.ts +++ b/src/main/native-chat/agent-session-journal/journal-pi-context-import.spec.ts @@ -1,12 +1,14 @@ import { mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' -import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import Database from '../../sqlite/sync-database' import { importLegacyTranscriptIntoJournal, prepareLegacyTranscriptImport } from './journal-legacy-import' import { createTrackedJournalOpener } from './journal-host-database-test-support' +import { createPiTranscriptCollector } from './journal-pi-active-branch' import { PI_CONTEXT_FIXTURES } from './journal-pi-context-fixtures' export function registerJournalPiContextImportTests(): void { @@ -42,6 +44,78 @@ export function registerJournalPiContextImportTests(): void { } describe('saved Pi context import', () => { + it('keeps prepared history when temporary database teardown reports an error', async () => { + const close = Database.prototype.close + const closeSpy = vi.spyOn(Database.prototype, 'close').mockImplementation(function ( + this: Database.Database + ) { + close.call(this) + throw new Error('test teardown failure') + }) + const warn = vi.spyOn(console, 'warn').mockImplementation(() => undefined) + try { + const items = await prepare([ + { + type: 'message', + id: 'first', + parentId: null, + message: { role: 'user', content: 'hello' } + } + ]) + expect(items).toHaveLength(1) + expect(items[0]?.identity).toMatchObject({ provider: 'legacy', recordId: 'first' }) + expect(warn).toHaveBeenCalledWith( + '[journal-pi-import] temporary database close failed:', + expect.objectContaining({ message: 'test teardown failure' }) + ) + } finally { + closeSpy.mockRestore() + warn.mockRestore() + } + }) + + it('lets the event loop run while projecting a large active branch', async () => { + const collector = createPiTranscriptCollector( + (message) => message.id, + (_line, lineIndex) => ({ + provider: 'legacy', + agent: 'pi', + sessionId: 'session', + recordId: `active-${lineIndex}` + }) + ) + const selected: string[] = [] + let markerAt = -1 + try { + collector.record(JSON.stringify({ type: 'session', version: 3 }), 'session', 0) + for (let index = 0; index < 12; index += 1) { + collector.record( + JSON.stringify({ + type: 'message', + id: `active-${index}`, + parentId: index === 0 ? null : `active-${index - 1}`, + message: { role: 'user', content: 'x'.repeat(128 * 1024) } + }), + `active-${index}`, + index + 1 + ) + } + const marker = new Promise((resolve) => + setImmediate(() => { + markerAt = selected.length + resolve() + }) + ) + await collector.project((id) => selected.push(id)) + await marker + expect(selected).toEqual(Array.from({ length: 12 }, (_, index) => `active-${index}`)) + expect(markerAt).toBeGreaterThanOrEqual(0) + expect(markerAt).toBeLessThan(12) + } finally { + collector.close() + } + }) + it.each(PI_CONTEXT_FIXTURES)('$name', async (fixture) => { const items = await prepare(fixture.entries, 'version' in fixture ? fixture.version : 3) expect( @@ -53,6 +127,49 @@ export function registerJournalPiContextImportTests(): void { ).toEqual(fixture.texts.map((text) => [text])) }) + it('selects saved compaction ancestry after a large inactive branch', async () => { + const timestamp = '2026-10-12T12:00:00.000Z' + const inactive = Array.from({ length: 950 }, (_, index) => ({ + type: 'message', + id: `inactive-${index}`, + parentId: index === 0 ? 'root' : `inactive-${index - 1}`, + timestamp, + message: { role: 'user', content: 'discarded '.concat('x'.repeat(18 * 1024)) } + })) + const items = await prepare([ + { + type: 'message', + id: 'root', + parentId: null, + timestamp, + message: { role: 'user', content: 'root' } + }, + ...inactive, + { + type: 'compaction', + id: 'summary', + parentId: 'root', + firstKeptEntryId: 'root', + summary: 'summary', + timestamp + }, + { + type: 'message', + id: 'latest', + parentId: 'summary', + timestamp, + message: { role: 'assistant', content: 'latest' } + } + ]) + expect( + items.flatMap((item) => + item.body.kind === 'message' + ? item.body.blocks.flatMap((block) => (block.type === 'text' ? [block.text] : [])) + : [] + ) + ).toEqual(['summary', 'root', 'latest']) + }) + it('persists compaction and branch summaries in the adopted journal across reopen', async () => { const fixture = PI_CONTEXT_FIXTURES[0] if (!fixture) { diff --git a/src/main/native-chat/agent-session-wire/structured-agent-session-adopted-import.test.ts b/src/main/native-chat/agent-session-wire/structured-agent-session-adopted-import.test.ts index 92eed86e5e7..dd52d0f525d 100644 --- a/src/main/native-chat/agent-session-wire/structured-agent-session-adopted-import.test.ts +++ b/src/main/native-chat/agent-session-wire/structured-agent-session-adopted-import.test.ts @@ -1,7 +1,7 @@ import { AGENT_JOURNAL_THREAD_SCOPE } from '../../../shared/agent-session-journal-types' // Source validation must finish before a new session claims the provider conversation. -import { mkdtemp, rm, writeFile, truncate } from 'node:fs/promises' +import { appendFile, mkdtemp, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' import { join } from 'node:path' import { afterEach, describe, expect, it, vi } from 'vitest' @@ -303,7 +303,7 @@ describe('adopting a provider conversation on create', () => { const transcriptPath = join(root, 'rollout.jsonl') if (kind === 'oversized') { await writeCodexRollout(transcriptPath, 'original turn') - await truncate(transcriptPath, 16 * 1024 * 1024 + 1) + await appendFile(transcriptPath, 'x'.repeat(16 * 1024 * 1024 + 1)) } else if (kind === 'empty' || kind === 'invalid') { await writeFile(transcriptPath, kind === 'empty' ? '' : 'not json\n') } @@ -324,7 +324,7 @@ describe('adopting a provider conversation on create', () => { expect(store?.getRecord(SESSION)).toBeNull() expect(store?.listOperationRows()).toEqual([]) if (kind === 'oversized') { - expect(JSON.stringify(result)).toContain('import bound') + expect(JSON.stringify(result)).toContain('byte limit') } } ) diff --git a/src/main/native-chat/transcript-stream-lines.ts b/src/main/native-chat/transcript-stream-lines.ts index a3542665aab..64de2adb625 100644 --- a/src/main/native-chat/transcript-stream-lines.ts +++ b/src/main/native-chat/transcript-stream-lines.ts @@ -1,7 +1,6 @@ import type { Readable } from 'node:stream' import { StringDecoder } from 'node:string_decoder' import type { NativeChatMessage } from '../../shared/native-chat-types' -import { NodeReadableTextTooLargeError } from '../../shared/node-readable-text' import { transcriptFallbackId } from './transcript-fallback-id' type TranscriptDecoder = (line: string, fallbackId: string) => NativeChatMessage | null @@ -12,24 +11,17 @@ export async function decodeTranscriptStream( start: number, decode: TranscriptDecoder, includeTrailingLine: boolean, - maxSourceBytes = Infinity + maxRecordBytes = Infinity ): Promise<{ messages: NativeChatMessage[]; consumedBytes: number }> { const messages: NativeChatMessage[] = [] let consumedBytes = 0 - let sourceBytes = 0 const framer = createTranscriptLineFramer((line, byteLength, terminated) => { if (terminated || includeTrailingLine) { decodeLine(line, consumedBytes) consumedBytes += byteLength } - }) + }, maxRecordBytes) for await (const chunk of stream) { - if (maxSourceBytes !== Infinity) { - sourceBytes += Buffer.isBuffer(chunk) ? chunk.byteLength : Buffer.byteLength(chunk, 'utf8') - if (sourceBytes > maxSourceBytes) { - throw new NodeReadableTextTooLargeError(sourceBytes, maxSourceBytes) - } - } framer.write(chunk) } framer.end() diff --git a/src/shared/wsl-move/wsl-move-storage-dispositions.ts b/src/shared/wsl-move/wsl-move-storage-dispositions.ts index d50d1535b74..72d84d1d8c4 100644 --- a/src/shared/wsl-move/wsl-move-storage-dispositions.ts +++ b/src/shared/wsl-move/wsl-move-storage-dispositions.ts @@ -103,6 +103,7 @@ export const SQLITE_SCHEMA_FILE_DATABASES: Record = { 'agent-session-journal', 'src/main/native-chat/agent-session-journal/command-receipt-schema.ts': 'agent-session-journal', 'src/main/native-chat/agent-session-journal/journal-database-schema.ts': 'agent-session-journal', + 'src/main/native-chat/agent-session-journal/journal-pi-active-branch.ts': 'pi-branch-import', 'src/main/native-chat/agent-session-journal/queued-message-schema.ts': 'agent-session-journal', 'src/main/persistence/profile-state/profile-state-database-schema.ts': 'profile-state', 'src/main/runtime/orchestration/db/schema/create-core-tables-sql.ts': 'orchestration', @@ -221,6 +222,11 @@ export const SQLITE_TABLE_DISPOSITIONS: Record ['${IMPORT_SCOPE_TABLE}'], notApplicableDisposition(58, 'Cookie import staging; the partition is bound, not copied.') ), + ...tables( + 'pi-branch-import', + ['entries'], + notApplicableDisposition(52, 'Operation-local Pi import scratch, outside workspace state.') + ), ...tables( 'test-fixture', [