fix(native-chat): bound legacy retirement proof work

Persist source-intrinsic refusals for the exact copy, source and destination descriptor. Resume completed proof work after transient failures, and let the existing conversion owner finish pending imports.

Use covering-index key pages for history membership and exit terminal retirement states before writer admission. Preserve originals and prepared unlink manifests.

src/main/native-chat/agent-session-journal/journal-legacy-file-lock.ts

@@ -1,5 +1,8 @@
 import { signalProcessTree } from '@orca/process-host/process-tree-termination'
-import { JournalLegacyRetentionError } from './journal-legacy-retention-error'
+import {
+  JournalLegacyRetentionError,
+  JournalLegacyIneligibleError
+} from './journal-legacy-retention-error'
 import type { PipedProcessSpawner } from '@orca/process-host/process-spec'
 import { spawnProcess } from '@orca/process-host'
 import { createRequire } from 'node:module'
@@ -74,7 +77,7 @@ async function startLegacyHelper(
       } else {
         const error =
           code === 4
-            ? new JournalLegacyRetentionError('legacy storage or file identity is unsupported')
+            ? new JournalLegacyIneligibleError('legacy storage or file identity is unsupported')
             : new Error('legacy journal lock is unavailable')
         ready.reject(error)
         reject(error)

src/main/native-chat/agent-session-journal/journal-legacy-proof-frontier.ts

@@ -0,0 +1,91 @@
+import { JournalLegacyRetentionError } from './journal-legacy-retention-error'
+import type { LegacyProofCursor, LegacyProofPage } from './journal-legacy-source-proof'
+import type { JournalSqlReaderPool } from './journal-sql-reader-pool'
+import type { JournalSqlConversions } from './journal-sql-conversions'
+import type { JournalSqlOwner } from './journal-sql-owner'
+
+export type LegacyProofFrontier = {
+  cursor: LegacyProofCursor | null
+  page?: LegacyProofPage
+  session: number
+  pending: boolean
+  converted?: boolean
+}
+
+export function createLegacyProofFrontier(): LegacyProofFrontier {
+  return { cursor: { table: 0, after: null }, session: 0, pending: false }
+}
+
+export async function completeJournalLegacyProof(
+  readers: Pick<JournalSqlReaderPool, 'legacyProof' | 'importState'>,
+  conversions: Pick<JournalSqlConversions, 'ensure'>,
+  writer: Pick<JournalSqlOwner, 'maintain'>,
+  boundary: () => Promise<boolean>,
+  signal: AbortSignal,
+  rest: (started: number) => Promise<void>,
+  frontier = createLegacyProofFrontier()
+): Promise<void> {
+  const admit = async () => {
+    signal.throwIfAborted()
+    if (!(await boundary())) {
+      throw new Error('legacy proof was interrupted')
+    }
+    signal.throwIfAborted()
+  }
+  while (frontier.cursor) {
+    await admit()
+    if (!frontier.page) {
+      const started = performance.now()
+      frontier.page = await readers.legacyProof(frontier.cursor)
+      await rest(started)
+    }
+    const page = frontier.page
+    if (page.retained) {
+      throw new JournalLegacyRetentionError(page.retained, page.reason)
+    }
+    while (frontier.session < page.sessions.length) {
+      await admit()
+      const source = page.sessions[frontier.session]!
+      if (!frontier.pending) {
+        frontier.pending = true
+        await conversions.ensure(source.sessionId, false)
+        frontier.converted = true
+      }
+      let state = await readers.importState(source.sessionId)
+      if (state.state !== 'ready' && source.empty && frontier.converted) {
+        const epoch = await writer.maintain({
+          kind: 'import-finish',
+          header: {
+            sessionId: source.sessionId,
+            workspaceId: source.workspaceId,
+            epoch: source.epoch,
+            head: {
+              lastSequence: 0,
+              highestFence: 0,
+              contextBoundary: 0,
+              lastActivityAt: 0,
+              derivedTurnItemId: null,
+              historyOrigin: null,
+              queueFailureBoundary: 0,
+              queuePauseMarks: { latestStop: null, resumedSequence: 0, reopenedSequence: 0 }
+            }
+          }
+        })
+        if (typeof epoch !== 'string') {
+          throw new Error('legacy proof received no committed empty-chat epoch')
+        }
+        state = { state: 'ready', epoch }
+      }
+      if (state.state !== 'ready') {
+        throw new Error('legacy proof awaits committed conversion')
+      }
+      frontier.session++
+      frontier.pending = false
+      frontier.converted = false
+    }
+    frontier.cursor = page.next
+    frontier.page = undefined
+    frontier.session = 0
+  }
+  signal.throwIfAborted()
+}

src/main/native-chat/agent-session-journal/journal-legacy-proof.ts

@@ -0,0 +1,166 @@
+import {
+  JournalLegacyRetentionError,
+  JournalLegacyIneligibleError
+} from './journal-legacy-retention-error'
+import {
+  completeJournalLegacyProof,
+  createLegacyProofFrontier
+} from './journal-legacy-proof-frontier'
+import {
+  decodeLegacySourceRefusal,
+  type LegacySourceRefusal,
+  type Retirement
+} from './journal-legacy-retirement-state'
+import { journalDatabasePath } from './database/journal-host-database'
+import { journalDomainDatabasePath } from './database/journal-domain-database'
+import { readJournalLegacySourceReceipt } from './database/journal-legacy-source-receipt'
+import { legacyDestinationMain } from './journal-legacy-retirement-files'
+import type { createJournalLegacyFileEligibility } from './journal-legacy-file-lock'
+import type { JournalSqlOwner } from './journal-sql-owner'
+import type { JournalSqlReaderPool } from './journal-sql-reader-pool'
+import type { JournalSqlConversions } from './journal-sql-conversions'
+
+type Observation = { proof: string; observed: string; destination: string }
+
+/** One descriptor owns verified work, negative evidence and pending conversion progress. */
+export class JournalLegacyProof {
+  private descriptor?: string
+  private receipt = ''
+  private refusal?: LegacySourceRefusal
+  private refusalSaved = false
+  private ineligible = false
+  private eligible = false
+  private proven = false
+  private restamping = false
+  private frontier = createLegacyProofFrontier()
+
+  constructor(
+    private readonly directory: string,
+    private readonly writer: Pick<JournalSqlOwner, 'maintain'>,
+    private readonly readers: Pick<JournalSqlReaderPool, 'legacyProof' | 'importState'>,
+    private readonly conversions: Pick<JournalSqlConversions, 'ensure'>,
+    private readonly boundary: () => Promise<boolean>,
+    private readonly signal: AbortSignal,
+    private readonly rest: (started: number) => Promise<void>,
+    private readonly eligibility: ReturnType<typeof createJournalLegacyFileEligibility>
+  ) {}
+
+  get terminal(): boolean {
+    return !!this.refusal || this.ineligible
+  }
+
+  async retained(): Promise<boolean> {
+    if (!this.terminal) {
+      return false
+    }
+    await this.observe(this.receipt)
+    if (!this.terminal) {
+      return false
+    }
+    await this.saveRefusal()
+    return true
+  }
+
+  async observe(proof: string): Promise<Observation> {
+    const observed = JSON.stringify(
+      await readJournalLegacySourceReceipt(journalDatabasePath(this.directory))
+    )
+    const destination = await legacyDestinationMain(journalDomainDatabasePath(this.directory))
+    const descriptor = `${proof}\n${observed}\n${destination}`
+    if (descriptor !== this.descriptor) {
+      this.descriptor = descriptor
+      this.receipt = proof
+      this.refusal = undefined
+      this.refusalSaved = this.ineligible = this.eligible = this.proven = this.restamping = false
+      this.frontier = createLegacyProofFrontier()
+    }
+    return { proof, observed, destination }
+  }
+
+  restore(value: string | null, observation: Observation): boolean {
+    const refusal = decodeLegacySourceRefusal(value)
+    if (
+      !refusal ||
+      refusal.proof !== observation.proof ||
+      refusal.observed !== observation.observed ||
+      refusal.destination !== observation.destination
+    ) {
+      return false
+    }
+    this.refusal = refusal
+    this.refusalSaved = true
+    return true
+  }
+
+  async checkEligibility(): Promise<void> {
+    if (this.eligible) {
+      return
+    }
+    try {
+      await this.eligibility.check(journalDatabasePath(this.directory), this.signal)
+    } catch (error) {
+      this.ineligible = error instanceof JournalLegacyIneligibleError
+      throw error
+    }
+    this.eligible = true
+  }
+
+  restamped(): void {
+    this.restamping = false
+  }
+
+  async run(observation: Observation, state: Retirement | null, fresh = false): Promise<void> {
+    if (fresh && !this.restamping) {
+      this.proven = false
+      this.frontier = createLegacyProofFrontier()
+      this.restamping = true
+    }
+    if (this.proven) {
+      return
+    }
+    try {
+      await completeJournalLegacyProof(
+        this.readers,
+        this.conversions,
+        this.writer,
+        this.boundary,
+        this.signal,
+        this.rest,
+        this.frontier
+      )
+    } catch (error) {
+      if (error instanceof JournalLegacyRetentionError && error.reason) {
+        this.refusal = {
+          version: 1,
+          phase: 'refused',
+          policy: 1,
+          proof: observation.proof,
+          observed: observation.observed,
+          destination: observation.destination,
+          reason: error.reason
+        }
+        // Partial unlink still owes its original manifest, even when a later proof refuses.
+        this.refusalSaved = state?.phase === 'prepared' || state?.phase === 'complete'
+        await this.saveRefusal()
+      }
+      throw error
+    }
+    this.proven = true
+  }
+
+  private async saveRefusal(): Promise<void> {
+    if (!this.refusal || this.refusalSaved) {
+      return
+    }
+    if (
+      (await this.writer.maintain({
+        kind: 'legacy-retirement-save',
+        value: JSON.stringify(this.refusal),
+        durable: false
+      })) !== true
+    ) {
+      throw new Error('legacy refusal commit was not acknowledged')
+    }
+    this.refusalSaved = true
+  }
+}

src/main/native-chat/agent-session-journal/journal-legacy-retention-error.ts

@@ -1,2 +1,35 @@
-/** Only refusals from the frozen-source proof are permanent for that descriptor. */
-export class JournalLegacyRetentionError extends Error {}
+import { isSqliteCorruption } from '../../sqlite/sqlite-read-failure'
+
+export const LEGACY_SOURCE_REFUSALS = [
+  'source-inventory',
+  'source-recovery',
+  'source-orphan',
+  'source-queue',
+  'source-queued-only-v1',
+  'source-corrupt'
+] as const
+export type LegacySourceRefusalReason = (typeof LEGACY_SOURCE_REFUSALS)[number]
+
+/** Only proven frozen-source facts can survive a database owner. */
+export class JournalLegacyRetentionError extends Error {
+  constructor(
+    message: string,
+    readonly reason?: LegacySourceRefusalReason
+  ) {
+    super(message)
+  }
+}
+
+/** Unsupported storage is stable for this owner/descriptor, not across opens. */
+export class JournalLegacyIneligibleError extends JournalLegacyRetentionError {}
+
+export function readJournalLegacySource<T>(read: () => T): T {
+  try {
+    return read()
+  } catch (error) {
+    if (isSqliteCorruption(error)) {
+      throw new JournalLegacyRetentionError('legacy source is corrupt', 'source-corrupt')
+    }
+    throw error
+  }
+}

src/main/native-chat/agent-session-journal/journal-legacy-retirement-cost.test.ts

@@ -0,0 +1,693 @@
+import { existsSync, writeFileSync, readFileSync, statSync, unlinkSync } from 'node:fs'
+import { join } from 'node:path'
+import { ChildProcess } from 'node:child_process'
+import { PassThrough } from 'node:stream'
+import { spawnProcess } from '@orca/process-host'
+import { expect, it, vi } from 'vitest'
+import Database from '../../sqlite/sync-database'
+import { journalDomainDatabasePath } from './database/journal-domain-database'
+import { JournalSqlBackgroundMaintenance } from './journal-sql-background-maintenance'
+import { JournalSqlActivity } from './journal-sql-activity'
+import * as fileLock from './journal-legacy-file-lock'
+import * as profileLock from '../../orcad/instance-lock/orcad-instance-lock-hold'
+import * as retirementFiles from './journal-legacy-retirement-files'
+import * as sourceReceipt from './database/journal-legacy-source-receipt'
+import { readJournalSqlDatabase } from './journal-sql-database-read'
+import { journalSqlReaderPath } from './journal-sql-worker-bundle.test-fixture'
+import { JournalLegacyRetirement } from './journal-legacy-retirement'
+import { readJournalLegacyProofPage, type LegacyProofCursor } from './journal-legacy-source-proof'
+import { JournalSqlReaderPool } from './journal-sql-reader-pool'
+import { prepareJournalDomainImport } from './journal-domain-import'
+import { journalSqlRetryRows } from './journal-sql-retry-rows.test-fixture'
+import { insertJournalRow } from './durable-rows/journal-row-table'
+import { profile, cleanup } from './journal-legacy-retirement.test-fixture'
+
+vi.mock('@orca/process-host', { spy: true })
+
+it('persists one late source refusal across 1 and 10 owners at N and 10N history size', async () => {
+  for (const count of [128, 1280]) {
+    const check = vi.fn(async () => {})
+    const test = await profile(
+      (legacy) => {
+        const row = journalSqlRetryRows(true).find((row) => row.kind === 'item')!
+        for (let n = 0; n < count; n++) {
+          insertJournalRow(legacy, 'unopened', {
+            ...row,
+            seq: journalSqlRetryRows(true).length + n + 1,
+            itemId: `preceding-${n}`
+          })
+        }
+        legacy.exec("INSERT INTO journal_rows VALUES('z-orphan','old',1,1,'saved orphan history')")
+      },
+      { present: () => {}, check }
+    )
+    const bytes = readFileSync(test.filename)
+    const proof = vi.spyOn(test.readers, 'legacyProof')
+    const ensure = vi.spyOn(test.conversions, 'ensure')
+    await test.retirement.run()
+    expect(test.receiptState()).toMatchObject({
+      phase: 'refused',
+      reason: 'source-orphan',
+      policy: 1
+    })
+    test.retirement.stop()
+    await test.retirement.flushCredit()
+    proof.mockClear()
+    ensure.mockClear()
+    check.mockClear()
+    for (let owners = 0; owners < 10; owners++) {
+      const owner = test.reopen()
+      await owner.run()
+      owner.stop()
+      await owner.flushCredit()
+      expect(proof).not.toHaveBeenCalled()
+      expect(ensure).not.toHaveBeenCalled()
+      expect(check).not.toHaveBeenCalled()
+      expect(test.receiptState().phase).toBe('refused')
+    }
+    expect(readFileSync(test.filename)).toEqual(bytes)
+  }
+})
+
+it('invalidates every saved refusal descriptor component and rejects unsupported refusal policy', async () => {
+  const test = await profile((legacy) =>
+    legacy.exec("INSERT INTO journal_outcomes VALUES(901,1,'orphan','key',1,'0','orphan text',0)")
+  )
+  await test.retirement.run()
+  const refused = test.receiptState()
+  const proof = vi.spyOn(test.readers, 'legacyProof')
+  for (const changed of ['proof', 'observed', 'destination', 'policy', 'reason']) {
+    test.saveState({ ...refused, [changed]: changed === 'policy' ? 2 : 'unsupported-or-old' })
+    proof.mockClear()
+    await test.reopen().run()
+    expect(proof.mock.calls.length).toBeGreaterThan(0)
+    expect(test.receiptState()).toMatchObject({ phase: 'refused', reason: 'source-orphan' })
+  }
+  const receipt = String(
+    test.db
+      .prepare(
+        "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-source-copy.v1'"
+      )
+      .get()!.value
+  )
+  test.db
+    .prepare(
+      "UPDATE agent_session_store_meta SET value=? WHERE key='journal.legacy-source-copy.v1'"
+    )
+    .run(`${receipt}\n`)
+  proof.mockClear()
+  await test.reopen().run()
+  expect(proof).not.toHaveBeenCalled()
+  expect(existsSync(test.filename)).toBe(true)
+  test.db
+    .prepare(
+      "UPDATE agent_session_store_meta SET value=? WHERE key='journal.legacy-source-copy.v1'"
+    )
+    .run(receipt)
+  const fresh = await profile()
+  fresh.db
+    .prepare("INSERT INTO agent_session_store_meta VALUES('journal.legacy-retirement.v1',?)")
+    .run(JSON.stringify(refused))
+  await fresh.retirement.run()
+  expect(fresh.receiptState().phase).toBe('grace')
+  const changed = new Database(test.filename)
+  changed.prepare("INSERT INTO journal_sessions VALUES('downgrade','folder','new-epoch')").run()
+  changed.close()
+  proof.mockClear()
+  await test.reopen().run()
+  expect(proof).not.toHaveBeenCalled()
+  expect(test.errors.at(-1)).toMatchObject({
+    message: 'legacy source no longer matches its copy receipt'
+  })
+  expect(existsSync(test.filename)).toBe(true)
+})
+
+it('retries an unacknowledged refusal save without replaying proof', async () => {
+  const test = await profile((legacy) =>
+    legacy.exec("INSERT INTO journal_rows VALUES('orphan','old',1,1,'owed text')")
+  )
+  const maintain = test.writer.maintain.bind(test.writer)
+  let saves = 0
+  vi.spyOn(test.writer, 'maintain').mockImplementation(async (operation) => {
+    if (operation.kind === 'legacy-retirement-save' && ++saves === 1) {
+      throw new Error('write temporarily unavailable')
+    }
+    return maintain(operation)
+  })
+  const proof = vi.spyOn(test.readers, 'legacyProof')
+  let now = 0
+  vi.spyOn(performance, 'now').mockImplementation(() => now)
+  await test.retirement.run()
+  const calls = proof.mock.calls.length
+  now = 60001
+  await test.retirement.run()
+  expect(saves).toBe(2)
+  expect(proof).toHaveBeenCalledTimes(calls)
+  expect(test.receiptState().phase).toBe('refused')
+  expect(existsSync(test.filename)).toBe(true)
+})
+
+it('resumes late failed proof pages without rereading completed prefixes at N and 10N', async () => {
+  for (const pages of [2, 20]) {
+    for (const retries of [1, 10]) {
+      const test = await profile((legacy) => {
+        legacy.exec('BEGIN')
+        const session = legacy.prepare("INSERT INTO journal_sessions VALUES(?,'folder','epoch')")
+        const row = legacy.prepare(
+          "INSERT INTO journal_rows VALUES(?,'epoch',1,1,'preserved text')"
+        )
+        for (let n = 0; n < (pages + 1) * 128; n++) {
+          const id = `chat-${String(n).padStart(5, '0')}`
+          session.run(id)
+          row.run(id)
+        }
+        legacy.exec('COMMIT')
+      })
+      const ready = test.db.prepare(
+        "UPDATE journal_chat_state SET epoch='epoch',last_sequence=1,highest_fence=0 WHERE session_id=?"
+      )
+      for (let n = 0; n < (pages + 1) * 128; n++) {
+        ready.run(`chat-${String(n).padStart(5, '0')}`)
+      }
+      const actual = test.readers.legacyProof
+      const completed = new Map<string, number>()
+      let failing = true
+      let failed = 0
+      let contentPages = 0
+      let failedCursor: string | undefined
+      const proof = vi.spyOn(test.readers, 'legacyProof').mockImplementation(async (cursor) => {
+        const key = JSON.stringify(cursor)
+        if (cursor.table === 1 && (failedCursor === key || contentPages === pages) && failing) {
+          failedCursor = key
+          failed++
+          throw new Error('reader exited')
+        }
+        const result = await actual(cursor)
+        completed.set(key, (completed.get(key) ?? 0) + 1)
+        if (cursor.table === 1) {
+          contentPages++
+        }
+        return result
+      })
+      const ensure = vi.spyOn(test.conversions, 'ensure')
+      let now = 0
+      vi.spyOn(performance, 'now').mockImplementation(() => now)
+      await test.retirement.run()
+      const prefix = ensure.mock.calls.length
+      for (let retry = 0; retry < retries; retry++) {
+        now += 3600001
+        await test.retirement.run()
+      }
+      expect(failed).toBe(retries + 1)
+      expect(contentPages).toBe(pages)
+      expect(proof.mock.calls.length).toBe(completed.size + failed)
+      expect(ensure).toHaveBeenCalledTimes(prefix)
+      expect([...completed.values()].every((count) => count === 1)).toBe(true)
+      expect(
+        test.db
+          .prepare(
+            "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-retirement.v1'"
+          )
+          .get()
+      ).toBeUndefined()
+      failing = false
+      now += 3600001
+      await test.retirement.run()
+      expect(test.receiptState().phase).toBe('grace')
+      expect([...completed.values()].every((count) => count === 1)).toBe(true)
+      expect(ensure).toHaveBeenCalledTimes(prefix)
+      vi.restoreAllMocks()
+    }
+  }
+})
+
+it('yields a failed conversion to its normal owner and probes only the pending session on retries', async () => {
+  for (const retries of [1, 10]) {
+    const test = await profile((legacy) =>
+      legacy.exec("INSERT INTO journal_sessions VALUES('a-completed','folder','empty')")
+    )
+    const inode = statSync(journalDomainDatabasePath(test.root)).ino
+    const normal = test.conversions.ensure
+    const ensure = vi.spyOn(test.conversions, 'ensure').mockImplementation(async (id) => {
+      if (id === 'unopened') {
+        prepareJournalDomainImport(test.source(), id)
+        throw new Error('recoverable conversion publication failure')
+      }
+      return normal(id)
+    })
+    const state = vi.spyOn(test.readers, 'importState')
+    let now = 0
+    vi.spyOn(performance, 'now').mockImplementation(() => now)
+    await test.retirement.run()
+    const attempts = ensure.mock.calls.length
+    state.mockClear()
+    for (let retry = 0; retry < retries; retry++) {
+      now += 3600001
+      await test.retirement.run()
+    }
+    expect(ensure).toHaveBeenCalledTimes(attempts)
+    expect(state.mock.calls).toEqual(Array.from({ length: retries }, () => ['unopened']))
+    expect(
+      test.db
+        .prepare(
+          "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-retirement.v1'"
+        )
+        .get()
+    ).toBeUndefined()
+    await normal('unopened')
+    now += 3600001
+    await test.retirement.run()
+    expect(test.receiptState().phase).toBe('grace')
+    expect(ensure).toHaveBeenCalledTimes(attempts)
+    expect(statSync(journalDomainDatabasePath(test.root)).ino).toBe(inode)
+    vi.restoreAllMocks()
+  }
+})
+
+it('uses bounded covering key pages independently of duplicate counts and payload bytes', async () => {
+  for (const [rows, keys, payload] of [
+    [256, 3, 8],
+    [2560, 3, 8],
+    [256, 3, 8192],
+    [256, 128, 8],
+    [2560, 1280, 8]
+  ]) {
+    const test = await profile((legacy) => {
+      legacy.exec('BEGIN')
+      const session = legacy.prepare("INSERT INTO journal_sessions VALUES(?,'folder','epoch')")
+      const row = legacy.prepare('INSERT INTO journal_rows VALUES(?,?,?,1,?)')
+      const generation = legacy.prepare('INSERT INTO journal_outcome_generations VALUES(?,?,?)')
+      const outcome = legacy.prepare(
+        "INSERT INTO journal_outcomes VALUES(?,?,'field','key',1,'0',?,0)"
+      )
+      for (let key = 0; key < keys!; key++) {
+        const id = `key-${String(key).padStart(5, '0')}`
+        session.run(id)
+        generation.run(key + 1, id, 'epoch')
+      }
+      for (let n = 0; n < rows!; n++) {
+        const key = n % keys!
+        row.run(`key-${String(key).padStart(5, '0')}`, 'epoch', n + 1, 'x'.repeat(payload!))
+        outcome.run(key + 1, n + 1, 'y'.repeat(payload!))
+      }
+      legacy.exec('COMMIT')
+    })
+    const source = test.source()
+    const prepare = vi.spyOn(source, 'prepare')
+    for (const table of [1, 4]) {
+      prepare.mockClear()
+      let cursor: LegacyProofCursor | null = { table, after: null }
+      let pages = 0
+      while (cursor?.table === table) {
+        cursor = readJournalLegacyProofPage(source, cursor, test.db).next
+        pages++
+      }
+      const sql = prepare.mock.calls.map(([sql]) => sql)
+      const keyQueries = sql.filter((sql) => sql.startsWith('SELECT DISTINCT'))
+      const bound = table === 1 ? keys! + 1 : keys!
+      expect(pages).toBeLessThanOrEqual(Math.ceil(bound / 128) + 1)
+      const lookups = sql.filter((sql) =>
+        table === 1
+          ? sql.startsWith('SELECT 1 FROM journal_sessions WHERE')
+          : sql.startsWith('SELECT session_id,epoch FROM journal_outcome_generations')
+      )
+      expect(lookups).toHaveLength(bound)
+      for (const query of new Set(keyQueries)) {
+        expect(query).not.toMatch(/row_json|value_json|OFFSET|rowid/i)
+        const args = query.includes('WHERE') ? (table === 1 ? ['key-00000', 'epoch'] : [1]) : []
+        const plans = source
+          .prepare(`EXPLAIN QUERY PLAN ${query}`)
+          .all(...args)
+          .map((row) => String(row.detail))
+          .join('\n')
+        expect(plans).toContain('USING COVERING INDEX')
+        expect(plans).not.toMatch(/TEMP B-TREE|SORT/)
+        if (query.includes('WHERE')) {
+          expect(plans).toContain('SEARCH')
+        }
+      }
+    }
+    prepare.mockRestore()
+    const old = new Database(test.filename)
+    old
+      .prepare("INSERT INTO journal_rows VALUES('zz-orphan','missing-epoch',1,1,'saved orphan')")
+      .run()
+    old.close()
+    let cursor: LegacyProofCursor | null = { table: 1, after: null }
+    expect(() => {
+      while (cursor?.table === 1) {
+        cursor = readJournalLegacyProofPage(source, cursor, test.db).next
+      }
+    }).toThrow('legacy source has orphan content')
+    const append = new Database(test.filename)
+    append
+      .prepare(
+        "INSERT INTO journal_outcomes VALUES(999999,1,'field','key',1,'0','saved orphan outcome',0)"
+      )
+      .run()
+    append.close()
+    cursor = { table: 4, after: null }
+    expect(() => {
+      while (cursor?.table === 4) {
+        cursor = readJournalLegacyProofPage(source, cursor, test.db).next
+      }
+    }).toThrow('legacy source has an orphan outcome')
+    cursor = { table: 1, after: 'key-00000', epoch: 'epoch' }
+    const epochSource = new Database(test.filename)
+    epochSource
+      .prepare("INSERT INTO journal_rows VALUES('key-00000','wrong-epoch',1,1,'saved other epoch')")
+      .run()
+    epochSource.close()
+    expect(() => readJournalLegacyProofPage(source, cursor!, test.db)).toThrow(
+      'legacy source has orphan content'
+    )
+  }
+})
+
+it('classifies only source corruption through the real reader transport and recovers busy/exit/destination failures', async () => {
+  for (const fault of ['source-corrupt', 'source-busy', 'source-exit', 'destination-corrupt']) {
+    const test = await profile()
+    const marker = join(test.root, 'injected-reader-fault')
+    writeFileSync(marker, fault)
+    const worker = join(test.root, 'fault-reader.cjs')
+    writeFileSync(
+      worker,
+      `
+const sqlite = require('node:sqlite')
+const { workerData } = require('node:worker_threads')
+const { existsSync } = require('node:fs')
+const Original = sqlite.DatabaseSync
+sqlite.DatabaseSync = class extends Original {
+  constructor(filename, options) { super(filename, options); this.source = filename === workerData.originalFilename }
+  prepare(sql) {
+    const sourceBoundary = this.source && sql.startsWith('SELECT rowid AS rowid,session_id,epoch,generation FROM journal_outcome_generations')
+    const destinationBoundary = !this.source && sql.startsWith('PRAGMA table_info(')
+    if (existsSync(${JSON.stringify(marker)}) && (${JSON.stringify(fault)} === 'destination-corrupt' ? destinationBoundary : sourceBoundary)) {
+      if (${JSON.stringify(fault)} === 'source-exit') process.exit(7)
+      throw Object.assign(new Error('injected sqlite failure'), { code: 'ERR_SQLITE_ERROR', errcode: ${fault === 'source-busy' ? 5 : 11} })
+    }
+    return super.prepare(sql)
+  }
+}
+require(${JSON.stringify(journalSqlReaderPath())})
+`
+    )
+    const readers = new JournalSqlReaderPool(
+      journalDomainDatabasePath(test.root),
+      test.filename,
+      worker
+    )
+    const proof = vi.spyOn(readers, 'legacyProof')
+    const ensure = vi.spyOn(test.conversions, 'ensure')
+    const check = vi.fn(async () => {})
+    const owner = () =>
+      new JournalLegacyRetirement(
+        test.root,
+        test.writer,
+        readers,
+        test.conversions,
+        async () => true,
+        (error) => test.errors.push(error),
+        () => true,
+        async () => ({ held: () => true, close: async () => {} }),
+        undefined,
+        { present: () => {}, check }
+      )
+    let now = 0
+    vi.spyOn(performance, 'now').mockImplementation(() => now)
+    try {
+      const retirement = owner()
+      await retirement.run()
+      if (fault === 'source-corrupt') {
+        expect(test.receiptState()).toMatchObject({ phase: 'refused', reason: 'source-corrupt' })
+        proof.mockClear()
+        ensure.mockClear()
+        check.mockClear()
+        for (let n = 0; n < 10; n++) {
+          now += 3600001
+          await retirement.run()
+        }
+        retirement.stop()
+        await retirement.flushCredit()
+        await owner().run()
+        expect(proof).not.toHaveBeenCalled()
+        expect(ensure).not.toHaveBeenCalled()
+        expect(check).not.toHaveBeenCalled()
+      } else {
+        expect(
+          test.db
+            .prepare(
+              "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-retirement.v1'"
+            )
+            .get()
+        ).toBeUndefined()
+        if (fault !== 'source-exit') {
+          expect(test.errors[0]).toMatchObject({
+            code: 'ERR_SQLITE_ERROR',
+            errcode: fault === 'source-busy' ? 5 : 11
+          })
+        }
+        unlinkSync(marker)
+        now += 3600001
+        await retirement.run()
+        expect(test.receiptState().phase).toBe('grace')
+      }
+      expect(existsSync(test.filename)).toBe(true)
+    } finally {
+      await readers.close()
+      vi.restoreAllMocks()
+    }
+  }
+})
+
+it('settles terminal retirement before writer admission on the real minute timer', async () => {
+  for (const disposition of [
+    'absent',
+    'no-source',
+    'uncertifiable',
+    'nonempty-wal',
+    'complete',
+    'refused',
+    'ineligible'
+  ]) {
+    const test = await profile(
+      disposition === 'refused'
+        ? (legacy) =>
+            legacy.exec("INSERT INTO journal_rows VALUES('orphan','old',1,1,'saved orphan')")
+        : undefined
+    )
+    if (disposition === 'absent') {
+      test.db
+        .prepare("DELETE FROM agent_session_store_meta WHERE key='journal.legacy-source-copy.v1'")
+        .run()
+    } else if (['no-source', 'uncertifiable', 'nonempty-wal'].includes(disposition)) {
+      const receipt = JSON.parse(
+        String(
+          test.db
+            .prepare(
+              "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-source-copy.v1'"
+            )
+            .get()!.value
+        )
+      )
+      if (disposition === 'nonempty-wal') {
+        receipt.wal = { kind: 'nonempty' }
+      } else {
+        receipt.disposition =
+          disposition === 'no-source' ? 'no-source-at-initialization' : 'uncertifiable'
+      }
+      test.db
+        .prepare(
+          "UPDATE agent_session_store_meta SET value=? WHERE key='journal.legacy-source-copy.v1'"
+        )
+        .run(JSON.stringify(receipt))
+    }
+    if (disposition === 'complete') {
+      await test.retirement.run()
+      test.arm()
+      await test.retirement.run()
+      expect(existsSync(test.filename)).toBe(false)
+    }
+    if (disposition === 'refused') {
+      await test.retirement.run()
+      expect(test.receiptState().phase).toBe('refused')
+    }
+    const factory = fileLock.createJournalLegacyFileEligibility
+    vi.spyOn(fileLock, 'createJournalLegacyFileEligibility').mockImplementation(() => ({
+      ...factory(),
+      present: () => {}
+    }))
+    vi.spyOn(profileLock, 'ownsOrcadProfileDirectory').mockReturnValue(true)
+    const spawn = vi.mocked(spawnProcess).mockClear()
+    spawn.mockImplementation(() => {
+      const helper = Object.assign(new ChildProcess(), {
+        stdin: new PassThrough(),
+        stdout: new PassThrough(),
+        stderr: new PassThrough()
+      })
+      queueMicrotask(() => helper.emit('exit', 4))
+      cleanup.push(() => {
+        helper.stdin.destroy()
+        helper.stdout.destroy()
+        helper.stderr.destroy()
+      })
+      return helper
+    })
+    const maintain = vi.spyOn(test.writer, 'maintain')
+    const proof = vi.spyOn(test.readers, 'legacyProof')
+    const presence = vi.spyOn(retirementFiles, 'legacyFilePresent')
+    const observations = vi.spyOn(sourceReceipt, 'readJournalLegacySourceReceipt')
+    const identities = vi.spyOn(retirementFiles, 'legacyDestinationMain')
+    const maintenance = new JournalSqlBackgroundMaintenance(
+      test.writer,
+      { database: async (operation) => readJournalSqlDatabase(test.db, operation) },
+      test.conversions,
+      new JournalSqlActivity(),
+      (error) => test.errors.push(error)
+    )
+    maintenance.ownLegacySource(test.root, test.readers)
+    vi.useFakeTimers({ toFake: ['Date', 'setTimeout', 'clearTimeout'] })
+    vi.spyOn(performance, 'now').mockImplementation(() => Date.now())
+    try {
+      await maintenance.start()
+      const initial = maintain.mock.calls.filter(([operation]) =>
+        operation.kind.startsWith('legacy-retirement')
+      ).length
+      presence.mockClear()
+      observations.mockClear()
+      identities.mockClear()
+      for (let wake = 1; wake <= (disposition === 'ineligible' ? 1440 : 100); wake++) {
+        await vi.advanceTimersByTimeAsync(60000)
+        await maintenance.settled()
+        if (wake === 10 || wake === 100 || wake === 1440) {
+          expect(
+            maintain.mock.calls.filter(([operation]) =>
+              operation.kind.startsWith('legacy-retirement')
+            )
+          ).toHaveLength(initial)
+          expect(proof).not.toHaveBeenCalled()
+          expect(spawn).toHaveBeenCalledTimes(disposition === 'ineligible' ? 1 : 0)
+          if (disposition === 'refused' || disposition === 'ineligible') {
+            expect(observations.mock.calls.length).toBeLessThanOrEqual(wake)
+            expect(identities.mock.calls.length).toBeLessThanOrEqual(wake)
+          }
+          if (disposition === 'complete') {
+            expect(
+              presence.mock.calls.filter(([path]) => path === test.filename).length
+            ).toBeLessThanOrEqual(wake)
+          }
+        }
+      }
+      if (disposition === 'ineligible') {
+        identities.mockResolvedValue('changed-synthetic-main')
+        await vi.advanceTimersByTimeAsync(60000)
+        await maintenance.settled()
+        expect(spawn).toHaveBeenCalledTimes(2)
+        expect(proof).not.toHaveBeenCalled()
+      }
+      const beforeClose = maintain.mock.calls.filter(([operation]) =>
+        operation.kind.startsWith('legacy-retirement')
+      ).length
+      maintenance.stop()
+      await maintenance.settled()
+      expect(
+        maintain.mock.calls.filter(([operation]) => operation.kind.startsWith('legacy-retirement'))
+      ).toHaveLength(beforeClose)
+      if (disposition === 'ineligible') {
+        const fresh = new JournalLegacyRetirement(
+          test.root,
+          test.writer,
+          test.readers,
+          test.conversions,
+          async () => true,
+          (error) => test.errors.push(error),
+          () => true
+        )
+        await fresh.run()
+        expect(spawn).toHaveBeenCalledTimes(3)
+        fresh.stop()
+        await fresh.flushCredit()
+      }
+      if (disposition === 'complete') {
+        writeFileSync(test.filename, 'recreated old-build history')
+        await test.retirement.run()
+        expect(
+          maintain.mock.calls.filter(([operation]) => operation.kind === 'legacy-retirement-state')
+            .length
+        ).toBeGreaterThan(1)
+        expect(readFileSync(test.filename, 'utf8')).toBe('recreated old-build history')
+      }
+    } finally {
+      maintenance.stop()
+      await maintenance.settled()
+      vi.useRealTimers()
+      vi.restoreAllMocks()
+    }
+  }
+})
+
+it('cancels before admitting another proof unit and still flushes grace credit on close', async () => {
+  const cancelled = await profile()
+  const actual = cancelled.readers.legacyProof
+  const proof = vi.spyOn(cancelled.readers, 'legacyProof')
+  const ensure = vi.spyOn(cancelled.conversions, 'ensure')
+  proof.mockImplementation(async (cursor) => {
+    const result = await actual(cursor)
+    cancelled.retirement.stop()
+    return result
+  })
+  await cancelled.retirement.run()
+  await cancelled.retirement.run()
+  expect(proof).toHaveBeenCalledOnce()
+  expect(ensure).not.toHaveBeenCalled()
+  expect(existsSync(cancelled.filename)).toBe(true)
+
+  const grace = await profile()
+  let mono = 0n
+  vi.spyOn(process.hrtime, 'bigint').mockImplementation(() => mono)
+  const wall = Date.now()
+  vi.spyOn(Date, 'now').mockReturnValue(wall)
+  await grace.retirement.run()
+  const save = vi.spyOn(grace.writer, 'maintain')
+  mono = 21600000000000n
+  vi.mocked(Date.now).mockReturnValue(wall + 21600000)
+  await grace.retirement.run()
+  expect(grace.receiptState().credit).toBe(21600000)
+  mono += 3600000000000n
+  vi.mocked(Date.now).mockReturnValue(wall + 25200000)
+  grace.retirement.stop()
+  await grace.retirement.flushCredit()
+  expect(grace.receiptState().credit).toBe(25200000)
+  expect(
+    save.mock.calls.filter(([operation]) => operation.kind === 'legacy-retirement-save')
+  ).toHaveLength(2)
+})
+
+it('invalidates partial session progress when the destination descriptor changes', async () => {
+  const test = await profile((legacy) =>
+    legacy.exec("INSERT INTO journal_sessions VALUES('a-completed','folder','empty')")
+  )
+  const normal = test.conversions.ensure
+  let failed = false
+  const ensure = vi.spyOn(test.conversions, 'ensure').mockImplementation(async (id) => {
+    if (id === 'unopened' && !failed) {
+      failed = true
+      throw new Error('reader exited during conversion')
+    }
+    return normal(id)
+  })
+  let now = 0
+  vi.spyOn(performance, 'now').mockImplementation(() => now)
+  await test.retirement.run()
+  vi.spyOn(retirementFiles, 'legacyDestinationMain').mockResolvedValue('different-synthetic-main')
+  now = 60001
+  await test.retirement.run()
+  expect(ensure.mock.calls.map(([id]) => id)).toEqual([
+    'a-completed',
+    'unopened',
+    'a-completed',
+    'unopened'
+  ])
+  expect(test.receiptState().phase).toBe('grace')
+  expect(existsSync(test.filename)).toBe(true)
+})

src/main/native-chat/agent-session-journal/journal-legacy-retirement-state.ts

@@ -1,3 +1,7 @@
+import {
+  LEGACY_SOURCE_REFUSALS,
+  type LegacySourceRefusalReason
+} from './journal-legacy-retention-error'
 import type { LegacyCompanionManifest } from './journal-legacy-retirement-files'
 
 export const LEGACY_RETIREMENT_WEEK = 7 * 24 * 60 * 60 * 1000
@@ -14,11 +18,59 @@ export type Retirement = {
   manifest?: LegacyCompanionManifest
 }
 
+export type LegacySourceRefusal = {
+  version: 1
+  phase: 'refused'
+  policy: 1
+  proof: string
+  observed: string
+  destination: string
+  reason: LegacySourceRefusalReason
+}
+
+export function decodeLegacySourceRefusal(value: string | null): LegacySourceRefusal | null {
+  const data: unknown = value ? JSON.parse(value) : null
+  if (!data || typeof data !== 'object' || !('phase' in data) || data.phase !== 'refused') {
+    return null
+  }
+  if (
+    !('version' in data) ||
+    data.version !== 1 ||
+    !('policy' in data) ||
+    data.policy !== 1 ||
+    !('proof' in data) ||
+    typeof data.proof !== 'string' ||
+    !('observed' in data) ||
+    typeof data.observed !== 'string' ||
+    !('destination' in data) ||
+    typeof data.destination !== 'string' ||
+    !('reason' in data) ||
+    typeof data.reason !== 'string'
+  ) {
+    return null
+  }
+  const reason = LEGACY_SOURCE_REFUSALS.find((reason) => reason === data.reason)
+  return reason
+    ? {
+        version: 1,
+        phase: 'refused',
+        policy: 1,
+        proof: data.proof,
+        observed: data.observed,
+        destination: data.destination,
+        reason
+      }
+    : null
+}
+
 export function decodeRetirement(value: string | null): Retirement | null {
   if (!value) {
     return null
   }
-  const data: Retirement = JSON.parse(value)
+  const data: Retirement | { phase: 'refused' } = JSON.parse(value)
+  if (data.phase === 'refused') {
+    return null
+  }
   if (
     data.version !== 1 ||
     typeof data.proof !== 'string' ||

src/main/native-chat/agent-session-journal/journal-legacy-retirement.test-fixture.ts

@@ -0,0 +1,176 @@
+import { mkdtempSync, rmSync } from 'node:fs'
+import { join } from 'node:path'
+import { tmpdir } from 'node:os'
+import { afterEach, vi } from 'vitest'
+import Database from '../../sqlite/sync-database'
+import { openJournalDatabase, runJournalTransaction } from './database/journal-database'
+import {
+  initializeJournalDomainDatabase,
+  openJournalDomainDatabase
+} from './database/journal-domain-database'
+import { journalDatabasePath } from './database/journal-host-database'
+import { JournalSqlBatch } from './journal-sql-batch'
+import { applyJournalSqlMaintenance, type JournalSqlMaintenance } from './journal-sql-maintenance'
+import { readJournalLegacyProofPage } from './journal-legacy-source-proof'
+import type { JournalSqlReaderPool } from './journal-sql-reader-pool'
+import { journalDomainImportState } from './journal-domain-import-rows'
+import { prepareJournalDomainImport } from './journal-domain-import'
+import { journalSqlRetryRows } from './journal-sql-retry-rows.test-fixture'
+import { insertJournalRow, publishJournalSessionEpoch } from './durable-rows/journal-row-table'
+import { acquireJournalLegacyFileLock } from './journal-legacy-file-lock'
+import { JournalLegacyRetirement } from './journal-legacy-retirement'
+
+export const cleanup: (() => void)[] = []
+afterEach(() => {
+  vi.useRealTimers()
+  vi.restoreAllMocks()
+  for (const close of cleanup.splice(0).toReversed()) {
+    close()
+  }
+})
+
+export async function profile(
+  seed?: (db: Database.Database) => void,
+  eligibility?: ConstructorParameters<typeof JournalLegacyRetirement>[9]
+) {
+  const root = mkdtempSync(join(tmpdir(), 'orca-legacy-retirement-'))
+  cleanup.push(() => rmSync(root, { recursive: true, force: true }))
+  const filename = journalDatabasePath(root)
+  const { db: legacy } = openJournalDatabase(filename)
+  runJournalTransaction(legacy, () => {
+    publishJournalSessionEpoch(
+      legacy,
+      { sessionId: 'unopened', workspaceId: 'folder-without-git' },
+      'epoch'
+    )
+    for (const row of journalSqlRetryRows(true)) {
+      insertJournalRow(legacy, 'unopened', row)
+    }
+  })
+  seed?.(legacy)
+  legacy.close()
+  const { db } = openJournalDomainDatabase(await initializeJournalDomainDatabase(root))
+  let closed = false
+  const closeDomain = () => {
+    if (!closed) {
+      db.close()
+      closed = true
+    }
+  }
+  cleanup.push(closeDomain)
+  let original: Database.Database | undefined
+  const source = () =>
+    (original ??= new Database(filename, { readonly: true, fileMustExist: true }))
+  cleanup.push(() => original?.close())
+  const batch = new JournalSqlBatch(db)
+  const writer = {
+    maintain: async (operation: JournalSqlMaintenance) => {
+      const result = await batch.run(
+        [operation],
+        (unit) => applyJournalSqlMaintenance(db, unit),
+        operation.kind === 'legacy-retirement-save' && operation.durable ? 'retirement' : 'normal'
+      )
+      const outcome = result.outcomes[0]!
+      if (!outcome.committed) {
+        throw outcome.error
+      }
+      return outcome.value
+    }
+  }
+  const readers = {
+    legacyProof: async (cursor) => readJournalLegacyProofPage(source(), cursor, db),
+    importState: async (id) => journalDomainImportState(db, id),
+    recycleSourceReaders: async () => {
+      original?.close()
+      original = undefined
+      return new AbortController()
+    },
+    releaseSourceReaders: () => {}
+  } satisfies Pick<
+    JournalSqlReaderPool,
+    'legacyProof' | 'importState' | 'recycleSourceReaders' | 'releaseSourceReaders'
+  >
+  const conversions = {
+    ensure: async (id: string) => {
+      if (journalDomainImportState(db, id).state === 'ready') {
+        return 'epoch'
+      }
+      const imported = prepareJournalDomainImport(source(), id)
+      if (!imported) {
+        return null
+      }
+      for (const page of imported.pages) {
+        await writer.maintain({ kind: 'import-page', target: imported, page })
+      }
+      return String(await writer.maintain({ kind: 'import-finish', header: imported }))
+    }
+  }
+  const errors: unknown[] = []
+  const lock =
+    process.env.ORCA_JOURNAL_NATIVE_LOCK_TEST === '1'
+      ? acquireJournalLegacyFileLock
+      : async () => ({ held: () => true, close: async () => {} })
+  const reopen = () =>
+    new JournalLegacyRetirement(
+      root,
+      writer,
+      readers,
+      conversions,
+      async () => true,
+      (error) => errors.push(error),
+      () => true,
+      lock,
+      undefined,
+      eligibility ??
+        (process.env.ORCA_JOURNAL_NATIVE_LOCK_TEST === '1'
+          ? undefined
+          : { check: async () => {}, present: () => {} })
+    )
+  const retirement = reopen()
+  const receiptState = () =>
+    JSON.parse(
+      String(
+        db
+          .prepare(
+            "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-retirement.v1'"
+          )
+          .get()!.value
+      )
+    )
+  const saveState = (state: unknown) =>
+    db
+      .prepare(
+        "UPDATE agent_session_store_meta SET value=? WHERE key='journal.legacy-retirement.v1'"
+      )
+      .run(JSON.stringify(state))
+  const arm = () => {
+    const row = db
+      .prepare(
+        "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-retirement.v1'"
+      )
+      .get()
+    const state = JSON.parse(String(row!.value))
+    state.credit = 604800000
+    state.process = 'prior-process'
+    db.prepare(
+      "UPDATE agent_session_store_meta SET value=? WHERE key='journal.legacy-retirement.v1'"
+    ).run(JSON.stringify(state))
+  }
+  return {
+    root,
+    filename,
+    db,
+    retirement,
+    reopen,
+    conversions,
+    errors,
+    arm,
+    source,
+    receiptState,
+    saveState,
+    writer,
+    closeDomain,
+    readers,
+    recycle: readers.recycleSourceReaders
+  }
+}

src/main/native-chat/agent-session-journal/journal-legacy-retirement.test.ts

@@ -1,7 +1,5 @@
 import { JournalLegacyRetentionError } from './journal-legacy-retention-error'
 import {
-  mkdtempSync,
-  rmSync,
   existsSync,
   writeFileSync,
   readFileSync,
@@ -11,17 +9,11 @@ import {
   statfsSync
 } from 'node:fs'
 import { join, basename } from 'node:path'
-import { tmpdir } from 'node:os'
 import { ChildProcess } from 'node:child_process'
 import { PassThrough } from 'node:stream'
 import { spawnProcess } from '@orca/process-host'
-import { afterEach, expect, it, vi } from 'vitest'
+import { expect, it, vi } from 'vitest'
 import Database from '../../sqlite/sync-database'
-import { openJournalDatabase, runJournalTransaction } from './database/journal-database'
-import {
-  initializeJournalDomainDatabase,
-  openJournalDomainDatabase
-} from './database/journal-domain-database'
 import { journalLegacyFileIdentity } from './database/journal-legacy-source-receipt'
 import { JournalSqlDatabase } from './database/journal-sql-database'
 import {
@@ -29,178 +21,17 @@ import {
   journalSqlReaderPath
 } from './journal-sql-worker-bundle.test-fixture'
 import { message } from './journal-sql-message.test-fixture'
-import { journalDatabasePath } from './database/journal-host-database'
-import { JournalSqlBatch } from './journal-sql-batch'
-import {
-  acquireJournalLegacyFileLock,
-  createJournalLegacyFileEligibility
-} from './journal-legacy-file-lock'
-import { JournalLegacyRetirement } from './journal-legacy-retirement'
+import { createJournalLegacyFileEligibility } from './journal-legacy-file-lock'
 import {
   advanceLegacyRetirementClock,
   JournalLegacyGraceClock
 } from './journal-legacy-retirement-state'
-import { readJournalLegacyProofPage } from './journal-legacy-source-proof'
-import { applyJournalSqlMaintenance, type JournalSqlMaintenance } from './journal-sql-maintenance'
-import type { JournalSqlReaderPool } from './journal-sql-reader-pool'
-import { journalDomainImportState } from './journal-domain-import-rows'
-import { prepareJournalDomainImport } from './journal-domain-import'
 import { journalSqlRetryRows } from './journal-sql-retry-rows.test-fixture'
-import { insertJournalRow, publishJournalSessionEpoch } from './durable-rows/journal-row-table'
+import { insertJournalRow } from './durable-rows/journal-row-table'
+import { profile, cleanup } from './journal-legacy-retirement.test-fixture'
 
 vi.mock('@orca/process-host', { spy: true })
 
-const cleanup: (() => void)[] = []
-afterEach(() => {
-  vi.useRealTimers()
-  vi.restoreAllMocks()
-  for (const close of cleanup.splice(0).toReversed()) {
-    close()
-  }
-})
-
-async function profile(
-  seed?: (db: Database.Database) => void,
-  eligibility?: ConstructorParameters<typeof JournalLegacyRetirement>[9]
-) {
-  const root = mkdtempSync(join(tmpdir(), 'orca-legacy-retirement-'))
-  cleanup.push(() => rmSync(root, { recursive: true, force: true }))
-  const filename = journalDatabasePath(root)
-  const { db: legacy } = openJournalDatabase(filename)
-  runJournalTransaction(legacy, () => {
-    publishJournalSessionEpoch(
-      legacy,
-      { sessionId: 'unopened', workspaceId: 'folder-without-git' },
-      'epoch'
-    )
-    for (const row of journalSqlRetryRows(true)) {
-      insertJournalRow(legacy, 'unopened', row)
-    }
-  })
-  seed?.(legacy)
-  legacy.close()
-  const { db } = openJournalDomainDatabase(await initializeJournalDomainDatabase(root))
-  let closed = false
-  const closeDomain = () => {
-    if (!closed) {
-      db.close()
-      closed = true
-    }
-  }
-  cleanup.push(closeDomain)
-  let original: Database.Database | undefined
-  const source = () =>
-    (original ??= new Database(filename, { readonly: true, fileMustExist: true }))
-  cleanup.push(() => original?.close())
-  const batch = new JournalSqlBatch(db)
-  const writer = {
-    maintain: async (operation: JournalSqlMaintenance) => {
-      const result = await batch.run(
-        [operation],
-        (unit) => applyJournalSqlMaintenance(db, unit),
-        operation.kind === 'legacy-retirement-save' && operation.durable ? 'retirement' : 'normal'
-      )
-      const outcome = result.outcomes[0]!
-      if (!outcome.committed) {
-        throw outcome.error
-      }
-      return outcome.value
-    }
-  }
-  const readers = {
-    legacyProof: async (cursor) => readJournalLegacyProofPage(source(), cursor, db),
-    importState: async (id) => journalDomainImportState(db, id),
-    recycleSourceReaders: async () => {
-      original?.close()
-      original = undefined
-      return new AbortController()
-    },
-    releaseSourceReaders: () => {}
-  } satisfies Pick<
-    JournalSqlReaderPool,
-    'legacyProof' | 'importState' | 'recycleSourceReaders' | 'releaseSourceReaders'
-  >
-  const conversions = {
-    ensure: async (id: string) => {
-      if (journalDomainImportState(db, id).state === 'ready') {
-        return 'epoch'
-      }
-      const imported = prepareJournalDomainImport(source(), id)
-      if (!imported) {
-        return null
-      }
-      for (const page of imported.pages) {
-        await writer.maintain({ kind: 'import-page', target: imported, page })
-      }
-      return String(await writer.maintain({ kind: 'import-finish', header: imported }))
-    }
-  }
-  const errors: unknown[] = []
-  const lock =
-    process.env.ORCA_JOURNAL_NATIVE_LOCK_TEST === '1'
-      ? acquireJournalLegacyFileLock
-      : async () => ({ held: () => true, close: async () => {} })
-  const retirement = new JournalLegacyRetirement(
-    root,
-    writer,
-    readers,
-    conversions,
-    async () => true,
-    (error) => errors.push(error),
-    () => true,
-    lock,
-    undefined,
-    eligibility ??
-      (process.env.ORCA_JOURNAL_NATIVE_LOCK_TEST === '1'
-        ? undefined
-        : { check: async () => {}, present: () => {} })
-  )
-  const receiptState = () =>
-    JSON.parse(
-      String(
-        db
-          .prepare(
-            "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-retirement.v1'"
-          )
-          .get()!.value
-      )
-    )
-  const saveState = (state: unknown) =>
-    db
-      .prepare(
-        "UPDATE agent_session_store_meta SET value=? WHERE key='journal.legacy-retirement.v1'"
-      )
-      .run(JSON.stringify(state))
-  const arm = () => {
-    const row = db
-      .prepare(
-        "SELECT value FROM agent_session_store_meta WHERE key='journal.legacy-retirement.v1'"
-      )
-      .get()
-    const state = JSON.parse(String(row!.value))
-    state.credit = 604800000
-    state.process = 'prior-process'
-    db.prepare(
-      "UPDATE agent_session_store_meta SET value=? WHERE key='journal.legacy-retirement.v1'"
-    ).run(JSON.stringify(state))
-  }
-  return {
-    root,
-    filename,
-    db,
-    retirement,
-    errors,
-    arm,
-    source,
-    receiptState,
-    saveState,
-    writer,
-    closeDomain,
-    readers,
-    recycle: readers.recycleSourceReaders
-  }
-}
-
 it('accounts for an unopened folder chat, removes only the certified file and keeps current history', async () => {
   const test = await profile((legacy) => {
     legacy.exec(`INSERT INTO journal_outcome_generations VALUES(900,'unopened','retired-epoch');
@@ -374,6 +205,14 @@ it('retains downgrade writes, a rollback journal and a nonempty WAL without chan
       expect(readFileSync(test.filename + (kind === 'wal' ? '-wal' : '-journal'), 'utf8')).toBe(
         'owed recovery bytes'
       )
+      if (kind === 'journal') {
+        unlinkSync(`${test.filename}-journal`)
+        const now = performance.now()
+        const clock = vi.spyOn(performance, 'now').mockReturnValue(now + 3600001)
+        await test.retirement.run()
+        expect(existsSync(test.filename)).toBe(false)
+        clock.mockRestore()
+      }
     }
   }
 })
@@ -419,13 +258,18 @@ it('retains queued-only text, recovery obligations and orphan history', async ()
     'repair',
     'orphan',
     'orphan-outcome',
-    'extra-profile-column'
+    'extra-profile-column',
+    'malformed-queue'
   ] as const) {
     const test = await profile((legacy) => {
       if (kind === 'queue') {
         legacy.exec(
           'DELETE FROM journal_rows; INSERT INTO queued_messages(session_id,message_id,position,body_json,fingerprint,created_at,host_instance,state,hold_reason) VALUES(\'unopened\',\'held\',1,\'{"kind":"message","role":"user","blocks":[{"type":"text","text":"held user text"}]}\',\'f\',1,\'h\',\'waiting\',\'manual\')'
         )
+      } else if (kind === 'malformed-queue') {
+        legacy.exec(
+          "INSERT INTO queued_messages(session_id,message_id,position,body_json,fingerprint,created_at,host_instance,state,hold_reason) VALUES('unopened','held',1,'unreadable saved text','f',1,'h','waiting','manual')"
+        )
       } else if (kind === 'orphan-outcome') {
         legacy.exec(
           `INSERT INTO journal_outcomes VALUES(901,1,'orphan','key',1,'0','"orphan user text"',0)`
@@ -440,6 +284,13 @@ it('retains queued-only text, recovery obligations and orphan history', async ()
         legacy.exec("INSERT INTO journal_rows VALUES('orphan','old',1,1,'orphan user text')")
       }
     })
+    if (kind === 'malformed-queue') {
+      test.db
+        .prepare(
+          "UPDATE journal_chat_state SET epoch='epoch',last_sequence=1,highest_fence=0 WHERE session_id='unopened'"
+        )
+        .run()
+    }
     if (kind === 'queue') {
       const current = await JournalSqlDatabase.open(test.root, {
         writerPath: journalSqlWorkerPath(),
@@ -461,6 +312,9 @@ it('retains queued-only text, recovery obligations and orphan history', async ()
     const before = readFileSync(test.filename)
     await test.retirement.run()
     expect(test.errors.length).toBeGreaterThan(0)
+    if (kind === 'malformed-queue') {
+      expect(test.receiptState()).toMatchObject({ phase: 'refused', reason: 'source-queue' })
+    }
     expect(readFileSync(test.filename)).toEqual(before)
     if (kind === 'queue') {
       expect(test.source().prepare('SELECT body_json FROM queued_messages').get()).toEqual({
@@ -506,50 +360,6 @@ it('resumes only the prepared empty companions and preserves a recreated main',
   }
 })
 
-it('settles an unchanged permanent proof refusal and retries a recoverable reader failure', async () => {
-  const retained = await profile((legacy) => {
-    legacy.exec("INSERT INTO journal_rows VALUES('orphan','old',1,1,'orphan user text')")
-  })
-  const pages = vi.spyOn(retained.readers, 'legacyProof')
-  await retained.retirement.run()
-  const firstPass = pages.mock.calls.length
-  expect(firstPass).toBeGreaterThan(0)
-  await retained.retirement.run()
-  await retained.retirement.run()
-  expect(pages.mock.calls).toHaveLength(firstPass)
-  expect(readFileSync(retained.filename).includes(Buffer.from('orphan user text'))).toBe(true)
-
-  const unsupported = await profile(undefined, {
-    check: async () => {
-      throw new JournalLegacyRetentionError('unsupported volume')
-    },
-    present: () => {}
-  })
-  const unsupportedProof = vi.spyOn(unsupported.readers, 'legacyProof')
-  const recycle = vi.spyOn(unsupported.readers, 'recycleSourceReaders')
-  await unsupported.retirement.run()
-  await unsupported.retirement.run()
-  expect(unsupportedProof).not.toHaveBeenCalled()
-  expect(recycle).not.toHaveBeenCalled()
-  expect(existsSync(unsupported.filename)).toBe(true)
-
-  const recovered = await profile()
-  let now = 0
-  vi.spyOn(performance, 'now').mockImplementation(() => now)
-  const retry = vi
-    .spyOn(recovered.readers, 'legacyProof')
-    .mockRejectedValueOnce(new Error('reader exited'))
-  await recovered.retirement.run()
-  await recovered.retirement.run()
-  expect(retry.mock.calls).toHaveLength(1)
-  now = 60_001
-  await recovered.retirement.run()
-  expect(recovered.receiptState().phase).toBe('grace')
-  recovered.arm()
-  await recovered.retirement.run()
-  expect(existsSync(recovered.filename)).toBe(false)
-})
-
 it('retires in the same owner after a missing helper is restored', async () => {
   let available = false
   let now = 0

src/main/native-chat/agent-session-journal/journal-legacy-retirement.ts

@@ -2,11 +2,9 @@ import { JournalLegacyRetentionError } from './journal-legacy-retention-error'
 import { randomUUID } from 'node:crypto'
 import { ownsOrcadProfileDirectory } from '../../orcad/instance-lock/orcad-instance-lock-hold'
 import { journalDatabasePath } from './database/journal-host-database'
-import { journalDomainDatabasePath } from './database/journal-domain-database'
 import {
   legacyFilePresent as filePresent,
   removeLegacyCompanions,
-  legacyDestinationMain,
   retireJournalLegacyMain,
   type LegacyDestinationIdentity
 } from './journal-legacy-retirement-files'
@@ -24,7 +22,7 @@ import {
 import type { JournalSqlOwner } from './journal-sql-owner'
 import type { JournalSqlReaderPool } from './journal-sql-reader-pool'
 import type { JournalSqlConversions } from './journal-sql-conversions'
-import { completeJournalLegacyProof } from './journal-legacy-source-proof'
+import { JournalLegacyProof } from './journal-legacy-proof'
 
 const WEEK = LEGACY_RETIREMENT_WEEK
 const SAVE_INTERVAL = 6 * 60 * 60 * 1000
@@ -37,10 +35,9 @@ export class JournalLegacyRetirement {
   private lastSaved = 0
   private diagnosed?: string
   private stopped = new AbortController()
-  private descriptor?: string
-  private retained = false
-  private eligible = false
-  private proven = false
+  private excluded = false
+  private completeAbsent = false
+  private readonly proof: JournalLegacyProof
   private retryAt = 0
   private retryDelay = 60_000
 
@@ -51,15 +48,25 @@ export class JournalLegacyRetirement {
       JournalSqlReaderPool,
       'legacyProof' | 'importState' | 'recycleSourceReaders' | 'releaseSourceReaders'
     >,
-    private readonly conversions: Pick<JournalSqlConversions, 'ensure'>,
+    conversions: Pick<JournalSqlConversions, 'ensure'>,
     private readonly boundary: () => Promise<boolean>,
     private readonly report: (error: unknown) => void,
     private readonly ownsProfile = () => ownsOrcadProfileDirectory(directory),
     private readonly lock = acquireJournalLegacyFileLock,
-    private readonly rest: (started: number) => Promise<void> = async () => {},
+    rest: (started: number) => Promise<void> = async () => {},
     private readonly eligibility = createJournalLegacyFileEligibility()
   ) {
     this.filename = journalDatabasePath(directory)
+    this.proof = new JournalLegacyProof(
+      directory,
+      writer,
+      readers,
+      conversions,
+      boundary,
+      this.stopped.signal,
+      rest,
+      eligibility
+    )
   }
 
   stop(): void {
@@ -84,7 +91,12 @@ export class JournalLegacyRetirement {
   }
 
   async flushCredit(): Promise<void> {
-    if (!this.stopped.signal.aborted) {
+    if (
+      !this.stopped.signal.aborted ||
+      this.excluded ||
+      this.completeAbsent ||
+      this.proof.terminal
+    ) {
       return
     }
     try {
@@ -144,31 +156,25 @@ export class JournalLegacyRetirement {
     return JSON.stringify(await readJournalLegacySourceReceipt(this.filename)) === proof
   }
 
-  private async prove(): Promise<void> {
-    if (this.proven) {
-      return
-    }
-    await completeJournalLegacyProof(
-      this.readers,
-      this.conversions,
-      this.writer,
-      this.boundary,
-      this.stopped.signal,
-      this.rest
-    ).catch((error) => {
-      this.retained = error instanceof JournalLegacyRetentionError
-      throw error
-    })
-    this.proven = true
-    this.retryDelay = 60_000
-  }
-
   private async attempt(): Promise<void> {
     if (this.stopped.signal.aborted || performance.now() < this.retryAt) {
       return
     }
+    if (this.excluded) {
+      return
+    }
+    if (this.completeAbsent) {
+      if (!(await filePresent(this.filename))) {
+        return
+      }
+      this.completeAbsent = false
+    }
+    if (await this.proof.retained()) {
+      return
+    }
     const saved = await this.savedState()
     if (!saved.receipt) {
+      this.excluded = true
       return
     }
     const receipt = JSON.parse(saved.receipt)
@@ -178,26 +184,17 @@ export class JournalLegacyRetirement {
       receipt.disposition !== 'copied-source' ||
       receipt.wal?.kind !== 'empty'
     ) {
+      this.excluded = true
       return
     }
     const proof = saved.receipt
-    const observed = JSON.stringify(await readJournalLegacySourceReceipt(this.filename))
-    const descriptor = `${proof}\n${observed}\n${await legacyDestinationMain(journalDomainDatabasePath(this.directory))}`
-    if (descriptor !== this.descriptor) {
-      this.descriptor = descriptor
-      this.retained = this.eligible = this.proven = false
-      this.retryAt = 0
-      this.retryDelay = 60_000
-    }
-    if (this.retained) {
-      return
-    }
     let state = decodeRetirement(saved.retirement)
-    if (state && state.proof !== saved.receipt) {
+    if (state && state.proof !== proof) {
       throw new JournalLegacyRetentionError('legacy retirement provenance changed')
     }
     if (state?.phase === 'complete') {
       if (!(await filePresent(this.filename))) {
+        this.completeAbsent = true
         return
       }
       state = { ...state, phase: 'prepared' }
@@ -206,6 +203,11 @@ export class JournalLegacyRetirement {
       await this.removeCompanions(state)
       return
     }
+    const observation = await this.proof.observe(proof)
+    const { observed } = observation
+    if (this.proof.restore(saved.retirement, observation)) {
+      return
+    }
     if (observed !== proof) {
       throw new JournalLegacyRetentionError('legacy source no longer matches its copy receipt')
     }
@@ -216,13 +218,11 @@ export class JournalLegacyRetirement {
     if (await filePresent(`${this.filename}-journal`)) {
       throw new JournalLegacyRetentionError('legacy rollback journal retained')
     }
-    if (!this.eligible) {
-      await this.eligibility.check(this.filename, this.stopped.signal)
-      this.eligible = true
-    }
+    await this.proof.checkEligibility()
     const now = Date.now()
     if (!state) {
-      await this.prove()
+      await this.proof.run(observation, state)
+      this.retryDelay = 60_000
       if (!(await this.matches(proof))) {
         throw new Error('legacy source changed during proof')
       }
@@ -242,12 +242,13 @@ export class JournalLegacyRetirement {
     }
     const advanced = this.clock.observe(state, now, PROCESS_TOKEN)
     if (advanced.process !== state.process || advanced.completedAt !== state.completedAt) {
-      this.proven = false
-      await this.prove()
+      await this.proof.run(observation, state, true)
+      this.retryDelay = 60_000
       if (!(await this.matches(proof))) {
         return
       }
       await this.save(advanced, true)
+      this.proof.restamped()
       return
     }
     state = advanced
@@ -257,7 +258,8 @@ export class JournalLegacyRetirement {
       }
       return
     }
-    await this.prove()
+    await this.proof.run(observation, state)
+    this.retryDelay = 60_000
     if (!(await this.matches(proof)) || !this.ownsProfile()) {
       return
     }
@@ -302,6 +304,7 @@ export class JournalLegacyRetirement {
       )
     ) {
       await this.save({ ...state, phase: 'complete' }, true)
+      this.completeAbsent = true
     }
   }
 }

src/main/native-chat/agent-session-journal/journal-legacy-source-proof.ts

@@ -1,9 +1,9 @@
-import { JournalLegacyRetentionError } from './journal-legacy-retention-error'
-import { journalOpenRefusal } from './replay/journal-open-failure'
+import {
+  JournalLegacyRetentionError,
+  readJournalLegacySource,
+  type LegacySourceRefusalReason
+} from './journal-legacy-retention-error'
 import { readStoredQueuedMessageRow } from './queued-messages/queued-message-stored-row'
-import type { JournalSqlReaderPool } from './journal-sql-reader-pool'
-import type { JournalSqlConversions } from './journal-sql-conversions'
-import type { JournalSqlOwner } from './journal-sql-owner'
 import type Database from '../../sqlite/sync-database'
 
 const PROFILE_TABLES = new Set([
@@ -40,7 +40,7 @@ const SOURCE_COLUMNS: Readonly<Record<string, string>> = {
     'session_id message_id position body_json fingerprint created_at host_instance state hold_reason returned_reason returned_rejection settled_at settled_by_op consumed_as carried_from queued_epoch queued_sequence options_json'
 }
 const RECOVERY_TABLES = ['journal_repairs', 'journal_imports', 'journal_set_aside']
-export type LegacyProofCursor = { table: number; after: string | null }
+export type LegacyProofCursor = { table: number; after: string | null; epoch?: string }
 export type LegacySourceSession = {
   sessionId: string
   workspaceId: string
@@ -51,6 +51,7 @@ export type LegacyProofPage = {
   next: LegacyProofCursor | null
   sessions: LegacySourceSession[]
   retained?: string
+  reason?: LegacySourceRefusalReason
 }
 
 /** Every saved content row must belong to the current catalog before any source file can retire. */
@@ -58,6 +59,30 @@ export function readJournalLegacyProofPage(
   db: Database.Database,
   cursor: LegacyProofCursor,
   destination: Database.Database
+): LegacyProofPage {
+  const copiedColumns = new Map<string, Set<unknown>>()
+  if (cursor.table === 0 && cursor.after === null) {
+    for (const table of PROFILE_TABLES) {
+      if (table !== 'sqlite_sequence') {
+        copiedColumns.set(
+          table,
+          new Set(
+            destination
+              .prepare(`PRAGMA table_info(${table})`)
+              .all()
+              .map((column) => column.name)
+          )
+        )
+      }
+    }
+  }
+  return readJournalLegacySource(() => readSourcePage(db, cursor, copiedColumns))
+}
+
+function readSourcePage(
+  db: Database.Database,
+  cursor: LegacyProofCursor,
+  copiedColumns: Map<string, Set<unknown>>
 ): LegacyProofPage {
   const tables = new Set(
     db
@@ -68,25 +93,26 @@ export function readJournalLegacyProofPage(
   if (cursor.table === 0 && cursor.after === null) {
     const version = Number(db.pragma('user_version', { simple: true }))
     if (version !== 3 && version !== 4) {
-      throw new JournalLegacyRetentionError('unsupported legacy source inventory')
+      throw new JournalLegacyRetentionError(
+        'unsupported legacy source inventory',
+        'source-inventory'
+      )
     }
     for (const table of tables) {
-      const names =
-        PROFILE_TABLES.has(table) && table !== 'sqlite_sequence'
-          ? destination
-              .prepare(`PRAGMA table_info(${table})`)
-              .all()
-              .map((column) => column.name)
-          : SOURCE_COLUMNS[table]?.split(' ')
-      if (names) {
-        const columns = new Set(names)
+      const columns =
+        copiedColumns.get(table) ??
+        (SOURCE_COLUMNS[table] ? new Set(SOURCE_COLUMNS[table]!.split(' ')) : undefined)
+      if (columns) {
         if (
           db
             .prepare(`PRAGMA table_info(${table})`)
             .all()
             .some((column) => !columns.has(column.name))
         ) {
-          throw new JournalLegacyRetentionError('legacy source has unaccounted columns')
+          throw new JournalLegacyRetentionError(
+            'legacy source has unaccounted columns',
+            'source-inventory'
+          )
         }
       }
 
@@ -99,10 +125,16 @@ export function readJournalLegacyProofPage(
       }
       const quoted = `"${table.replaceAll('"', '""')}"`
       if (db.prepare(`SELECT 1 FROM ${quoted} LIMIT 1`).get()) {
-        throw new JournalLegacyRetentionError('legacy source has unaccounted recovery data')
+        throw new JournalLegacyRetentionError(
+          'legacy source has unaccounted recovery data',
+          'source-recovery'
+        )
       }
       if (!RECOVERY_TABLES.includes(table)) {
-        throw new JournalLegacyRetentionError('legacy source has an unknown table')
+        throw new JournalLegacyRetentionError(
+          'legacy source has an unknown table',
+          'source-inventory'
+        )
       }
     }
   }
@@ -130,11 +162,22 @@ export function readJournalLegacyProofPage(
           : table === 'journal_rows'
             ? 'session_id,epoch'
             : 'session_id,epoch,generation'
+    const indexed = table === 'journal_rows' || table === 'journal_outcomes'
     const statement = db.prepare(
-      `SELECT rowid AS rowid,${columns} FROM ${table} ${cursor.after === null ? '' : 'WHERE rowid>?'} ORDER BY rowid LIMIT 128`
+      indexed
+        ? table === 'journal_rows'
+          ? `SELECT DISTINCT session_id,epoch FROM journal_rows ${cursor.after === null ? '' : 'WHERE (session_id,epoch)>(?,?)'} ORDER BY session_id,epoch LIMIT 128`
+          : `SELECT DISTINCT generation FROM journal_outcomes ${cursor.after === null ? '' : 'WHERE generation>?'} ORDER BY generation LIMIT 128`
+        : `SELECT rowid AS rowid,${columns} FROM ${table} ${cursor.after === null ? '' : 'WHERE rowid>?'} ORDER BY rowid LIMIT 128`
     )
     statement.setReadBigInts(true)
-    const rows = statement.all(...(cursor.after === null ? [] : [BigInt(cursor.after)]))
+    const rows = statement.all(
+      ...(cursor.after === null
+        ? []
+        : table === 'journal_rows'
+          ? [cursor.after, cursor.epoch!]
+          : [BigInt(cursor.after)])
+    )
     for (const row of rows) {
       if (
         table === 'queued_messages' &&
@@ -142,7 +185,10 @@ export function readJournalLegacyProofPage(
           db.prepare('SELECT * FROM queued_messages WHERE rowid=?').get(row.rowid!)
         )
       ) {
-        throw new JournalLegacyRetentionError('legacy source has unreadable queued text')
+        throw new JournalLegacyRetentionError(
+          'legacy source has unreadable queued text',
+          'source-queue'
+        )
       }
 
       const session =
@@ -154,7 +200,10 @@ export function readJournalLegacyProofPage(
               .get(row.generation!)
           : row
       if (table === 'journal_outcomes' && !session) {
-        throw new JournalLegacyRetentionError('legacy source has an orphan outcome')
+        throw new JournalLegacyRetentionError(
+          'legacy source has an orphan outcome',
+          'source-orphan'
+        )
       }
       if (table === 'journal_outcomes' || table === 'journal_outcome_generations') {
         if (
@@ -176,12 +225,22 @@ export function readJournalLegacyProofPage(
           )
           .get(session.session_id!, ...(matchEpoch ? [session.epoch!] : []))
       ) {
-        throw new JournalLegacyRetentionError('legacy source has orphan content')
+        throw new JournalLegacyRetentionError('legacy source has orphan content', 'source-orphan')
       }
     }
     return {
       next: rows.length
-        ? { table: cursor.table, after: String(rows.at(-1)!.rowid) }
+        ? {
+            table: cursor.table,
+            after: String(
+              table === 'journal_rows'
+                ? rows.at(-1)!.session_id
+                : table === 'journal_outcomes'
+                  ? rows.at(-1)!.generation
+                  : rows.at(-1)!.rowid
+            ),
+            ...(table === 'journal_rows' ? { epoch: String(rows.at(-1)!.epoch) } : {})
+          }
         : cursor.table < CONTENT_TABLES.length
           ? { table: cursor.table + 1, after: null }
           : null,
@@ -200,7 +259,10 @@ export function readJournalLegacyProofPage(
       tables.has('queued_messages') &&
       !!db.prepare('SELECT 1 FROM queued_messages WHERE session_id=? LIMIT 1').get(id)
     if (!hasRows && hasQueue) {
-      throw new JournalLegacyRetentionError('legacy queued-only history still requires its source')
+      throw new JournalLegacyRetentionError(
+        'legacy queued-only history still requires its source',
+        'source-queued-only-v1'
+      )
     }
     const empty =
       !hasRows &&
@@ -216,68 +278,3 @@ export function readJournalLegacyProofPage(
     sessions
   }
 }
-
-export async function completeJournalLegacyProof(
-  readers: Pick<JournalSqlReaderPool, 'legacyProof' | 'importState'>,
-  conversions: Pick<JournalSqlConversions, 'ensure'>,
-  writer: Pick<JournalSqlOwner, 'maintain'>,
-  boundary: () => Promise<boolean>,
-  signal: AbortSignal,
-  rest: (started: number) => Promise<void>
-): Promise<void> {
-  let cursor: LegacyProofCursor | null = { table: 0, after: null }
-  while (cursor && !signal.aborted) {
-    if (!(await boundary())) {
-      throw new Error('legacy proof was interrupted')
-    }
-    const started = performance.now()
-    const page = await readers.legacyProof(cursor)
-    await rest(started)
-    if (page.retained) {
-      throw new JournalLegacyRetentionError(page.retained)
-    }
-    for (const source of page.sessions) {
-      try {
-        await conversions.ensure(source.sessionId, false)
-      } catch (error) {
-        const refusal = journalOpenRefusal(error)
-        if (
-          refusal.code === 'agent_session_journal_unreadable' &&
-          refusal.details?.reason !== 'journalUnavailable'
-        ) {
-          throw new JournalLegacyRetentionError(refusal.message)
-        }
-        throw error
-      }
-      if ((await readers.importState(source.sessionId)).state !== 'ready' && source.empty) {
-        await writer.maintain({
-          kind: 'import-finish',
-          header: {
-            sessionId: source.sessionId,
-            workspaceId: source.workspaceId,
-            epoch: source.epoch,
-            head: {
-              lastSequence: 0,
-              highestFence: 0,
-              contextBoundary: 0,
-              lastActivityAt: 0,
-              derivedTurnItemId: null,
-              historyOrigin: null,
-              queueFailureBoundary: 0,
-              queuePauseMarks: { latestStop: null, resumedSequence: 0, reopenedSequence: 0 }
-            }
-          }
-        })
-      }
-      if ((await readers.importState(source.sessionId)).state !== 'ready') {
-        throw new JournalLegacyRetentionError(
-          'legacy source still has unconverted or queued history'
-        )
-      }
-    }
-    cursor = page.next
-  }
-  if (signal.aborted) {
-    throw new Error('legacy proof was interrupted')
-  }
-}

src/main/native-chat/agent-session-journal/journal-sql-background-maintenance.ts

@@ -33,7 +33,10 @@ export class JournalSqlBackgroundMaintenance {
 
   private legacyRetirement?: JournalLegacyRetirement
 
-  ownLegacySource(directory: string, readers: JournalSqlReaderPool): void {
+  ownLegacySource(
+    directory: string,
+    readers: ConstructorParameters<typeof JournalLegacyRetirement>[2]
+  ): void {
     this.legacyRetirement = new JournalLegacyRetirement(
       directory,
       this.writer,

src/main/native-chat/agent-session-journal/journal-sql-reader-entry.ts

@@ -1,4 +1,7 @@
-import { JournalLegacyRetentionError } from './journal-legacy-retention-error'
+import {
+  JournalLegacyRetentionError,
+  readJournalLegacySource
+} from './journal-legacy-retention-error'
 import { readJournalLegacyProofPage } from './journal-legacy-source-proof'
 import { JournalForkPreparation } from './journal-fork-preparation'
 import { parentPort, workerData } from 'node:worker_threads'
@@ -89,11 +92,14 @@ port.on('message', ({ id, operation }: JournalSqlReaderRequest) => {
       case 'conversion-prepare': {
         // Each frozen-source read is its own short statement; no snapshot spans the restore.
         if (!original && existsSync(workerData.originalFilename)) {
-          original = new Database(workerData.originalFilename, {
-            readonly: true,
-            fileMustExist: true,
-            timeout: 5000
-          })
+          const openSource = () =>
+            new Database(workerData.originalFilename, {
+              readonly: true,
+              fileMustExist: true,
+              timeout: 5000
+            })
+          original =
+            operation.kind === 'legacy-proof' ? readJournalLegacySource(openSource) : openSource()
         }
         if (operation.kind === 'legacy-proof') {
           if (!original) {
@@ -138,7 +144,7 @@ port.on('message', ({ id, operation }: JournalSqlReaderRequest) => {
       reply({
         type: 'legacy-proof',
         id,
-        page: { next: null, sessions: [], retained: error.message }
+        page: { next: null, sessions: [], retained: error.message, reason: error.reason }
       })
       return
     }