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
2026-10-30 16:47:34 +00:00 · 2026-10-15 13:45:27 -07:00
parent 4fdccda051
commit 7f492dc15b
9 changed files with 530 additions and 249 deletions
@@ -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)
+  })
+})
@@ -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<typeof FsPromises>('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)
-  })
-})
@@ -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 () => {
@@ -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<JournalReplacementItem[]> {
  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 }
 }

 /**
@@ -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<T>(
+  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<void>
+  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<void>((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<void> => {
+        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<void>((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<string>()
-  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)]
-}
@@ -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<void>((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) {
@@ -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')
      }
    }
  )
@@ -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()
@@ -103,6 +103,7 @@ export const SQLITE_SCHEMA_FILE_DATABASES: Record<string, string> = {
    '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<string, WslMoveFieldDisposition>
      ['${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',
      [