diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-file-lock.ts b/src/main/native-chat/agent-session-journal/journal-legacy-file-lock.ts index 301b9466ee3..ee51b3ac3ba 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-file-lock.ts +++ b/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) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-proof-frontier.ts b/src/main/native-chat/agent-session-journal/journal-legacy-proof-frontier.ts new file mode 100644 index 00000000000..c7d1ffb9090 --- /dev/null +++ b/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, + conversions: Pick, + writer: Pick, + boundary: () => Promise, + signal: AbortSignal, + rest: (started: number) => Promise, + frontier = createLegacyProofFrontier() +): Promise { + 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() +} diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-proof.ts b/src/main/native-chat/agent-session-journal/journal-legacy-proof.ts new file mode 100644 index 00000000000..2f91db00cfc --- /dev/null +++ b/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, + private readonly readers: Pick, + private readonly conversions: Pick, + private readonly boundary: () => Promise, + private readonly signal: AbortSignal, + private readonly rest: (started: number) => Promise, + private readonly eligibility: ReturnType + ) {} + + get terminal(): boolean { + return !!this.refusal || this.ineligible + } + + async retained(): Promise { + 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 { + 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 { + 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 { + 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 { + 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 + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-retention-error.ts b/src/main/native-chat/agent-session-journal/journal-legacy-retention-error.ts index 93590dae3ab..1633fbf2989 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-retention-error.ts +++ b/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(read: () => T): T { + try { + return read() + } catch (error) { + if (isSqliteCorruption(error)) { + throw new JournalLegacyRetentionError('legacy source is corrupt', 'source-corrupt') + } + throw error + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-retirement-cost.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-retirement-cost.test.ts new file mode 100644 index 00000000000..9b0dd0d022a --- /dev/null +++ b/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() + 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) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-retirement-state.ts b/src/main/native-chat/agent-session-journal/journal-legacy-retirement-state.ts index bf30e582c95..f0508bbcd85 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-retirement-state.ts +++ b/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' || diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-retirement.test-fixture.ts b/src/main/native-chat/agent-session-journal/journal-legacy-retirement.test-fixture.ts new file mode 100644 index 00000000000..f10560b947d --- /dev/null +++ b/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[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 + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-retirement.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-retirement.test.ts index 4c4f06f4ae2..2d127da6ab1 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-retirement.test.ts +++ b/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[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 diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-retirement.ts b/src/main/native-chat/agent-session-journal/journal-legacy-retirement.ts index 39bbe30ab8c..815730b16e3 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-retirement.ts +++ b/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, + conversions: Pick, private readonly boundary: () => Promise, private readonly report: (error: unknown) => void, private readonly ownsProfile = () => ownsOrcadProfileDirectory(directory), private readonly lock = acquireJournalLegacyFileLock, - private readonly rest: (started: number) => Promise = async () => {}, + rest: (started: number) => Promise = 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 { - 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 { - 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 { 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 } } } diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-source-proof.ts b/src/main/native-chat/agent-session-journal/journal-legacy-source-proof.ts index 0bbf38672ee..f6cf45be580 100644 --- a/src/main/native-chat/agent-session-journal/journal-legacy-source-proof.ts +++ b/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> = { '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>() + 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> ): 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, - conversions: Pick, - writer: Pick, - boundary: () => Promise, - signal: AbortSignal, - rest: (started: number) => Promise -): Promise { - 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') - } -} diff --git a/src/main/native-chat/agent-session-journal/journal-sql-background-maintenance.ts b/src/main/native-chat/agent-session-journal/journal-sql-background-maintenance.ts index 5cf0ba175cd..a75c7aae2e3 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-background-maintenance.ts +++ b/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[2] + ): void { this.legacyRetirement = new JournalLegacyRetirement( directory, this.writer, diff --git a/src/main/native-chat/agent-session-journal/journal-sql-reader-entry.ts b/src/main/native-chat/agent-session-journal/journal-sql-reader-entry.ts index 48ecb3d00e6..d69e16ad2e2 100644 --- a/src/main/native-chat/agent-session-journal/journal-sql-reader-entry.ts +++ b/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 }