From 254c8fffc98fc187cdb8047538f302b12ee26206 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:08:32 -0700 Subject: [PATCH 1/5] fix(native-chat): an older Orca skips and keeps a journal row of a kind it does not know --- .../agent-session-journal/journal-cursor.ts | 4 +- .../journal-epoch-controller.ts | 3 + .../journal-epoch-replacement.test.ts | 3 +- .../journal-epoch-replacement.ts | 70 +++++++++++++++--- .../agent-session-journal/journal-open.ts | 32 +++++---- .../agent-session-journal/journal-reducer.ts | 15 +++- .../journal-row-schema.ts | 37 ++++++---- .../journal-row-table.ts | 23 ++++++ .../journal-skipped-row.ts | 71 +++++++++++++++++++ .../journal-store-collaborators.ts | 1 + .../journal-store-contracts.ts | 4 +- .../queued-message-pause.ts | 13 ++-- .../agent-session-journal-batch.ts | 12 ++-- ...-session-stop-event-downgrade.unit.test.ts | 2 +- 14 files changed, 233 insertions(+), 57 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-skipped-row.ts diff --git a/src/main/native-chat/agent-session-journal/journal-cursor.ts b/src/main/native-chat/agent-session-journal/journal-cursor.ts index bd43623ae497..9f7b8fd5e766 100644 --- a/src/main/native-chat/agent-session-journal/journal-cursor.ts +++ b/src/main/native-chat/agent-session-journal/journal-cursor.ts @@ -9,7 +9,7 @@ import type { AgentJournalResetReason } from '../../../shared/agent-session-journal-types' import type { JournalReadSince } from './journal-store-contracts' -import type { JournalRow } from './journal-row-schema' +import type { JournalReadRow } from './journal-skipped-row' export type JournalCursorRange = { epoch: string @@ -77,7 +77,7 @@ export function findSequenceGap( export function readJournalSince( source: { state: { epoch: string; lastSequence: number; oldestSequence: number } - rowsAfter: (afterSequence: number) => JournalRow[] + rowsAfter: (afterSequence: number) => JournalReadRow[] readOnly: boolean }, cursor: AgentJournalCursor, diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts b/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts index 51d3a4b2fd53..2537c93ba70f 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts @@ -23,6 +23,8 @@ export class JournalEpochController { highestFence: () => number /** What of the live epoch's Stop and Resume a replacement restates. */ queuePauseRestatement: () => JournalQueuePauseRestatement + /** The live epoch's rows of a newer build's kind, which a replacement carries. */ + skippedSequences: () => readonly number[] cursor: () => AgentJournalCursor adopt: (loaded: JournalLoad) => void } @@ -71,6 +73,7 @@ export class JournalEpochController { fence, items, queuePause: this.deps.queuePauseRestatement(), + skippedSequences: this.deps.skippedSequences(), now: this.deps.now, mintEpoch: this.deps.mintEpoch, onPublished: this.deps.adopt diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts index 2992459a787a..5818f87f6b94 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts @@ -67,7 +67,8 @@ function replace(input: { reason: 'legacy_import', fence: 1, items: input.items, - queuePause: { lifted: false, liveStop: null }, + queuePause: { liftedAt: 0, liveStop: null }, + skippedSequences: [], now, mintEpoch: () => `epoch-${clock}`, onPublished: input.onPublished ?? (() => undefined) diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts index 70efc8912837..dd6d470db315 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts @@ -3,6 +3,8 @@ // One transaction: discard the old epoch's rows, insert the epoch row plus the // replacement items, move the session projection, and retire any repair marker // — this republished history is exactly what the marker was holding out for. +// A newer build's rows this build skips are carried, not discarded: that build +// still reads them after a downgrade and upgrade. import type { AgentJournalItemBody, @@ -24,10 +26,18 @@ import type { JournalQueuePauseRestatement } from './queued-message-pause' import { deleteJournalEpochRows, insertJournalRow, + insertJournalStoredRow, publishJournalSessionEpoch, - readJournalSessionEpoch + readJournalRow, + readJournalSessionEpoch, + type JournalStoredRow } from './journal-row-table' -import type { AgentJournalEpochReason, JournalRow } from './journal-row-schema' +import { + parseJournalRow, + type AgentJournalEpochReason, + type JournalRow +} from './journal-row-schema' +import { restampSkippedJournalRow } from './journal-skipped-row' import { assertJournalFence } from './journal-write-guards' export type JournalReplacementItem = AgentJournalProducerLinkage & { @@ -48,6 +58,8 @@ export function replaceJournalEpoch(input: { /** Restated in the new epoch, or the rewind would release cards the person stopped, or bring * back a /clear pause they already lifted. */ queuePause: JournalQueuePauseRestatement + /** The live epoch's rows of a newer build's kind. */ + skippedSequences: readonly number[] now: () => number mintEpoch: () => string /** Called the instant the transaction commits, before any fallible follow-up. */ @@ -78,20 +90,45 @@ export function replaceJournalEpoch(input: { applyJournalRow(state, row) rows.push(row) } - const { lifted, liveStop } = input.queuePause + const { sessionId } = input.identity + const { liftedAt, liveStop } = input.queuePause const place = () => ({ state, seq: state.lastSequence + 1, fence: input.fence, ts: input.now() }) - if (lifted) { - const row = buildJournalQueueResumeRow(place()) + const append = (row: JournalRow) => { applyJournalRow(state, row) rows.push(row) } + const carried: JournalStoredRow[] = [] + // After the rebuilt history, in the order their source rows held: a newer build may read a + // carried row against a restated Stop or Resume. + const tail: { from: number; write: () => void }[] = [] + if (liftedAt > 0) { + tail.push({ from: liftedAt, write: () => append(buildJournalQueueResumeRow(place())) }) + } if (liveStop) { - const row = buildJournalStopEventRow({ ...place(), event: liveStop }) - applyJournalRow(state, row) - rows.push(row) + const { sequence, event } = liveStop + const write = () => append(buildJournalStopEventRow({ ...place(), event })) + tail.push({ from: sequence, write }) + } + for (const stored of readLiveJournalRows(input.database, sessionId, input.skippedSequences)) { + const parsed = parseJournalRow(stored.rowJson) + const skipped = !parsed.ok && parsed.skipped + if (!skipped) { + continue + } + const write = () => { + const at = { epoch, seq: state.lastSequence + 1, fence: input.fence } + const moved = restampSkippedJournalRow(stored.rowJson, skipped, at) + if (moved) { + applyJournalRow(state, moved.row) + carried.push({ epoch, seq: moved.row.seq, ts: moved.row.ts, rowJson: moved.rowJson }) + } + } + tail.push({ from: stored.seq, write }) + } + for (const entry of tail.sort((left, right) => left.from - right.from)) { + entry.write() } - const { sessionId } = input.identity input.database.transaction((db) => { const retired = readJournalSessionEpoch(db, sessionId) if (retired !== null) { @@ -101,6 +138,9 @@ export function replaceJournalEpoch(input: { for (const row of rows) { insertJournalRow(db, sessionId, row) } + for (const row of carried) { + insertJournalStoredRow(db, sessionId, row) + } publishJournalSessionEpoch(db, input.identity, epoch) }) @@ -110,3 +150,15 @@ export function replaceJournalEpoch(input: { state.oldestSequence = 1 input.onPublished({ state, readOnly: false, corrupt: false, malformedRows: 0 }) } + +/** The live epoch's rows at these sequences, as stored. */ +function readLiveJournalRows( + database: JournalHostDatabase, + sessionId: string, + sequences: readonly number[] +): JournalStoredRow[] { + const live = readJournalSessionEpoch(database.db, sessionId) + return live === null + ? [] + : sequences.flatMap((seq) => readJournalRow(database.db, sessionId, live, seq) ?? []) +} diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index 8f947272dbc7..c18d5cf99eb4 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -20,6 +20,7 @@ import { import { JOURNAL_REPAIR_DISCLOSURE_ITEM_ID } from './journal-repair-disclosure' import { pendingJournalRepairSequence } from './journal-repair-marker' import { isJournalStopOrResumeRow, parseJournalRow, type JournalRow } from './journal-row-schema' +import { journalReadRow, type JournalReadRow } from './journal-skipped-row' /** Every epoch row is sequence 1, and no compaction moves that floor. */ const FIRST_JOURNAL_SEQUENCE = 1 @@ -30,8 +31,9 @@ export type JournalLoad = { readOnly: boolean /** Set when the surviving prefix is unusable and the caller must roll the epoch. */ corrupt: boolean - /** Rows skipped because their body failed to parse (future-version rows are - * `readOnly`, never counted here). The store discloses these in the timeline. */ + /** Rows dropped because their body failed to parse (a future version latches `readOnly`, and + * a newer build's unknown kind is skipped and kept; neither counts here). The store discloses + * these in the timeline. */ malformedRows: number /** Directory-internal: the first sequence of an unusable suffix. The store * deletes from here before it accepts a write; a probe leaves it alone. */ @@ -92,13 +94,14 @@ export function startJournalRowFold(input: JournalRowFoldInput): { const add = (entry: { seq: number; rowJson: string }): boolean => { const parsed = parseJournalRow(entry.rowJson) - if (!parsed.ok) { + // A newer build's kind still holds its place, so it is sequence-checked like any row. + const row = journalReadRow(parsed) + if (!row) { truncateFrom = entry.seq - latched = parsed.unreadable - malformedRows = parsed.unreadable ? 0 : 1 + latched = !parsed.ok && parsed.unreadable + malformedRows = latched ? 0 : 1 return false } - const row = parsed.row // Parse past a gap so an unreadable future row still latches read-only. if (gapSequence !== undefined) { return true @@ -119,6 +122,9 @@ export function startJournalRowFold(input: JournalRowFoldInput): { return true } applyJournalRow(state, row) + if (row.kind === 'skipped') { + return true + } const disclosure = row.kind === 'item' && row.itemId === JOURNAL_REPAIR_DISCLOSURE_ITEM_ID // A Stop or Resume is no history, so it never reads as a rebuilt or provider-backed epoch. if (!disclosure && !isJournalStopOrResumeRow(row)) { @@ -147,22 +153,22 @@ export function startJournalRowFold(input: JournalRowFoldInput): { return { add, finish } } -/** Rows after a cursor, in sequence order. Stops at the first row this build - * cannot parse, exactly as replay does. */ +/** Rows after a cursor, in sequence order, a newer build's kinds as placeholders that keep the + * sequence whole. Stops at the first row this build cannot parse, exactly as replay does. */ export function readJournalRowsAfterCursor( db: Database.Database, sessionId: string, epoch: string, afterSequence: number, limit?: number -): JournalRow[] { - const rows: JournalRow[] = [] +): JournalReadRow[] { + const rows: JournalReadRow[] = [] for (const stored of readJournalRowsAfter(db, sessionId, epoch, afterSequence, limit)) { - const parsed = parseJournalRow(stored.rowJson) - if (!parsed.ok) { + const row = journalReadRow(parseJournalRow(stored.rowJson)) + if (!row) { break } - rows.push(parsed.row) + rows.push(row) } return rows } diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index a7d6f150e70c..693a139cc970 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -26,7 +26,8 @@ import { structuredAgentSessionPayloadFingerprint } from '../../../shared/struct import { JournalDerivedTurnScope } from './journal-derived-turn-scope' import { removeJournalItem, statedOrDerivedTurnScope, upsertJournalItem } from './journal-item-fold' import { journalItemRevisionIsStale } from './journal-item-revision' -import { isJournalStopOrResumeRow, type JournalRow } from './journal-row-schema' +import { isJournalStopOrResumeRow } from './journal-row-schema' +import type { JournalReadRow } from './journal-skipped-row' import { acceptSubmissionFromProviderItem, applyJournalSubmission } from './journal-submission-fold' import { applyJournalDispatchRow } from './journal-dispatch-reducer' import { isWriteFailureSubmission } from '../../../shared/structured-agent-session-dispatch-rejection' @@ -64,6 +65,8 @@ export type JournalReducerState = { latestPersonTurnSequence: number /** The latest person's Stop event and Resume, what the queue's pause is derived from. */ queuePauseMarks: JournalQueuePauseMarks + /** Sequences of a newer build's rows this build skips, so a replacement can carry them. */ + skippedSequences: number[] } export function createJournalReducerState(sessionId: string, epoch: string): JournalReducerState { @@ -83,16 +86,22 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou appliedSettlementIds: new Set(), derivedTurnScope: new JournalDerivedTurnScope(), latestPersonTurnSequence: 0, - queuePauseMarks: createJournalQueuePauseMarks() + queuePauseMarks: createJournalQueuePauseMarks(), + skippedSequences: [] } } -export function applyJournalRow(state: JournalReducerState, row: JournalRow): void { +export function applyJournalRow(state: JournalReducerState, row: JournalReadRow): void { state.lastSequence = Math.max(state.lastSequence, row.seq) state.highestFence = Math.max(state.highestFence, row.fence) if (row.kind === 'epoch') { return } + // Holds its sequence and fence, the envelope facts every row has; its content is unknown here. + if (row.kind === 'skipped') { + state.skippedSequences.push(row.seq) + return + } state.lastActivityAt = Math.max(state.lastActivityAt, row.ts) if (row.kind === 'item') { if (journalItemRevisionIsStale(state, row.itemId, row.revision)) { diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index e65d5919b39b..a21afa453e52 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -4,6 +4,10 @@ // in-place rewrite. A row whose version this build does not understand is // UNREADABLE, not skippable: the caller must degrade to read-only rather than // render a partial timeline or compact past a row it cannot interpret. +// +// A row whose KIND this build does not know, in an envelope it can place, is a +// newer build's addition at the same version: it is SKIPPED (no fold, no +// render) and kept on disk, so a newer kind needs no version bump. import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' import { @@ -21,6 +25,11 @@ import { } from '../../../shared/agent-session-journal-schemas' import { isAdmissibleAgentSessionContextUsage } from '../../../shared/agent-session-context-usage-schema' import type { StructuredAgentSessionStopCause } from '../agent-session-wire/structured-agent-session-stop-cause' +import { + hasJournalRowEnvelope, + skippedJournalRow, + type JournalSkippedRow +} from './journal-skipped-row' /** Producer linkage rides the row BASE rather than the body: the two nested * prompt shapes are `.strict()`, so an unknown key on a body would make the @@ -79,10 +88,9 @@ export type JournalTombstoneRow = JournalRowBase & { queueResume?: true } -/** One Stop that took effect. Temporary carrier: a tombstone's extra key, because a released host - * deletes the journal from the first row kind it does not know (`journal-open.ts` then - * `journal-store-open.ts`) but ignores an unknown key; a row kind of its own once released hosts - * skip unknown kinds instead. */ +/** One Stop that took effect. Temporary carrier: a tombstone's extra key, because a host released + * before unknown kinds were skipped deletes the journal from the first row kind it does not know + * but ignores an unknown key; a row kind of its own once no supported build predates the skip. */ export type JournalStopEvent = { /** Persisted: never rename an arm. Only `user-stop` pauses the queue. */ reason: StructuredAgentSessionStopCause @@ -179,9 +187,11 @@ export type JournalRow = export type JournalRowParse = | { ok: true; row: JournalRow } /** Malformed JSON or a shape this build rejects outright. */ - | { ok: false; unreadable: false } + | { ok: false; unreadable: false; skipped?: undefined } /** A future schema version. The host must not write or compact this journal. */ - | { ok: false; unreadable: true } + | { ok: false; unreadable: true; skipped?: undefined } + /** A kind this build does not know, in a well-formed envelope: skipped, never repaired away. */ + | { ok: false; unreadable: false; skipped: JournalSkippedRow } const ROW_KINDS = new Set([ 'epoch', @@ -230,7 +240,11 @@ export function parseJournalRow(line: string): JournalRowParse { } } dropUnusableContextUsage(upcast) - return isJournalRow(upcast) ? { ok: true, row: upcast } : { ok: false, unreadable: false } + if (isJournalRow(upcast)) { + return { ok: true, row: upcast } + } + const skipped = skippedJournalRow(upcast, ROW_KINDS) + return skipped ? { ok: false, unreadable: false, skipped } : { ok: false, unreadable: false } } /** Linkage ids this build cannot trust, removed from a row it still keeps. @@ -320,14 +334,7 @@ function isJournalRow(record: Record): record is JournalRow { if (typeof record.kind !== 'string' || !ROW_KINDS.has(record.kind)) { return false } - if ( - typeof record.epoch !== 'string' || - !record.epoch || - !Number.isInteger(record.seq) || - (record.seq as number) < 1 || - !Number.isInteger(record.fence) || - typeof record.ts !== 'number' - ) { + if (!hasJournalRowEnvelope(record)) { return false } if (record.kind === 'item') { diff --git a/src/main/native-chat/agent-session-journal/journal-row-table.ts b/src/main/native-chat/agent-session-journal/journal-row-table.ts index 4665a0571150..b6140c3d4423 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-table.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-table.ts @@ -19,6 +19,8 @@ const INSERT_ROW = const SELECT_ROWS_AFTER = `SELECT seq, ts, row_json FROM journal_rows WHERE session_id = ? AND epoch = ? AND seq > ? ORDER BY seq ASC` const SELECT_ROWS_AFTER_LIMITED = `${SELECT_ROWS_AFTER} LIMIT ?` +const SELECT_ROW = `SELECT ts, row_json FROM journal_rows +WHERE session_id = ? AND epoch = ? AND seq = ?` const DELETE_SUFFIX = 'DELETE FROM journal_rows WHERE session_id = ? AND epoch = ? AND seq >= ?' const DELETE_EPOCH = 'DELETE FROM journal_rows WHERE session_id = ? AND epoch = ?' const DELETE_UNPUBLISHED = `DELETE FROM journal_rows WHERE session_id = ? @@ -48,6 +50,27 @@ export function insertJournalRow( return Buffer.byteLength(rowJson, 'utf8') } +/** A row stored as given, for one this build cannot type: a newer build's kind it carries. */ +export function insertJournalStoredRow( + db: Database.Database, + sessionId: string, + row: JournalStoredRow +): void { + db.prepare(INSERT_ROW).run(sessionId, row.epoch, row.seq, row.ts, row.rowJson) +} + +export function readJournalRow( + db: Database.Database, + sessionId: string, + epoch: string, + seq: number +): JournalStoredRow | null { + const row = db.prepare(SELECT_ROW).get(sessionId, epoch, seq) + return row && typeof row.ts === 'number' && typeof row.row_json === 'string' + ? { epoch, seq, ts: row.ts, rowJson: row.row_json } + : null +} + // Why pages, not `.iterate()`: a lazily consumed cursor pins a read snapshot for as long as the // consumer reduces, and a WAL checkpoint cannot pass an open snapshot. Each page is one completed // statement, so the consumer's memory is bounded by a page while no snapshot outlives a fetch. diff --git a/src/main/native-chat/agent-session-journal/journal-skipped-row.ts b/src/main/native-chat/agent-session-journal/journal-skipped-row.ts new file mode 100644 index 000000000000..120633e2d086 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-skipped-row.ts @@ -0,0 +1,71 @@ +// A newer build's row of a kind this build does not know. Its envelope places it, so it holds its +// sequence and is kept on disk; nothing here reads what it says. + +import type { JournalRow, JournalRowParse } from './journal-row-schema' + +/** Where a skipped row sits: its place in the sequence and nothing to fold. In memory only; the + * stored row stays as its writer wrote it. */ +export type JournalSkippedRow = { + kind: 'skipped' + /** The kind the row was stored under. */ + storedKind: string + epoch: string + seq: number + fence: number + ts: number +} + +/** A row as a reader meets it: one this build folds, or a placeholder for one it skips. */ +export type JournalReadRow = JournalRow | JournalSkippedRow + +/** The row a reader folds or skips past; null where it must stop. */ +export function journalReadRow(parsed: JournalRowParse): JournalReadRow | null { + return parsed.ok ? parsed.row : (parsed.skipped ?? null) +} + +/** A row of an unknown kind whose envelope passes every check a known row's must. A known kind + * that fails its own checks is malformed, never skipped. */ +export function skippedJournalRow( + record: Record, + knownKinds: ReadonlySet +): JournalSkippedRow | null { + const { kind } = record + if (typeof kind !== 'string' || kind.length === 0 || knownKinds.has(kind)) { + return null + } + if (!hasJournalRowEnvelope(record)) { + return null + } + const { epoch, seq, fence, ts } = record + return { kind: 'skipped', storedKind: kind, epoch, seq, fence, ts } +} + +/** The fields every row carries whatever its kind: the chat's epoch, a place in it, and its writer. */ +export function hasJournalRowEnvelope( + record: Record +): record is Record & Omit { + return ( + typeof record.epoch === 'string' && + record.epoch.length > 0 && + typeof record.seq === 'number' && + Number.isInteger(record.seq) && + record.seq >= 1 && + typeof record.fence === 'number' && + Number.isInteger(record.fence) && + typeof record.ts === 'number' + ) +} + +/** A skipped row's stored text at a new place: its envelope restamped, everything else kept as its + * writer wrote it. */ +export function restampSkippedJournalRow( + rowJson: string, + skipped: JournalSkippedRow, + place: Pick +): { rowJson: string; row: JournalSkippedRow } | null { + const stored: unknown = JSON.parse(rowJson) + if (typeof stored !== 'object' || stored === null || Array.isArray(stored)) { + return null + } + return { rowJson: JSON.stringify({ ...stored, ...place }), row: { ...skipped, ...place } } +} diff --git a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts index b938efa3c5c9..0da8ef11b1f2 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts @@ -78,6 +78,7 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal host.state().queuePauseMarks, host.state().latestPersonTurnSequence ), + skippedSequences: () => host.state().skippedSequences, cursor: host.cursor, adopt: host.adopt }) diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index 85e6b99f9029..51f05892e557 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts @@ -12,7 +12,7 @@ import type { } from '../../../shared/agent-session-journal-types' import type { JournalHostDatabase } from './journal-host-database' import type { JournalLifecycleMutationInput } from './journal-row-builders' -import type { JournalRow } from './journal-row-schema' +import type { JournalReadRow } from './journal-skipped-row' export type AgentSessionJournalOptions = { identity: AgentSessionJournalIdentity @@ -24,7 +24,7 @@ export type AgentSessionJournalOptions = { } export type JournalReadSince = - | { ok: true; rows: JournalRow[]; cursor: AgentJournalCursor } + | { ok: true; rows: JournalReadRow[]; cursor: AgentJournalCursor } | { ok: false; reset: AgentJournalResetReason } export type ResolveDispatchInput = { diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index 9039760ba43c..c5b463bc6d7f 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -85,11 +85,12 @@ export function journalUserStopInForce( } /** What a rewind's new epoch restates so its pauses read as they did: a lift of /clear's pause (a - * person's turn or a Resume happened), then the Stop still in force, in that order so the lift - * never ends the Stop. */ + * person's turn or a Resume happened) and the Stop still in force, each in the order its source + * row held, so the lift never ends the Stop. */ export type JournalQueuePauseRestatement = { - lifted: boolean - liveStop: JournalStopEvent | null + /** Where /clear's pause was lifted; 0 when it never was. */ + liftedAt: number + liveStop: JournalQueuePauseMarks['latestStop'] } export function journalQueuePauseRestatement( @@ -97,8 +98,8 @@ export function journalQueuePauseRestatement( latestPersonTurnSequence: number ): JournalQueuePauseRestatement { return { - lifted: latestPersonTurnSequence > 0 || marks.resumedSequence > 0, - liveStop: journalUserStopInForce(marks, latestPersonTurnSequence)?.event ?? null + liftedAt: Math.max(latestPersonTurnSequence, marks.resumedSequence), + liveStop: journalUserStopInForce(marks, latestPersonTurnSequence) } } diff --git a/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts b/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts index 42af5f830420..aed18f517287 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts @@ -14,10 +14,8 @@ import type { } from '../../../shared/agent-session-journal-types' import type { AgentSessionJournalBatch } from '../../../shared/agent-session-wire' import { findSequenceGap } from '../agent-session-journal/journal-cursor' -import { - isJournalStopOrResumeRow, - type JournalRow -} from '../agent-session-journal/journal-row-schema' +import { isJournalStopOrResumeRow } from '../agent-session-journal/journal-row-schema' +import type { JournalReadRow } from '../agent-session-journal/journal-skipped-row' export type JournalBatchProjection = | { ok: true; batch: AgentSessionJournalBatch } @@ -26,7 +24,7 @@ export type JournalBatchProjection = | { ok: false; reset: 'journal_gap' } export function projectJournalBatch(input: { - rows: readonly JournalRow[] + rows: readonly JournalReadRow[] snapshot: AgentJournalSnapshot /** Sequence the subscriber has already applied. */ afterSequence: number @@ -43,6 +41,10 @@ export function projectJournalBatch(input: { const touchedItemIds = new Set() const touchedClientMessageIds = new Set() for (const row of input.rows) { + if (row.kind === 'skipped') { + // A newer build's row: it keeps the sequence whole and touches nothing this build renders. + continue + } if (row.kind === 'lifecycle-batch') { for (const mutation of row.mutations) { touchedItemIds.add( diff --git a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts index 20b64c7f6dbb..d9f6b68dfac2 100644 --- a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts +++ b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts @@ -77,7 +77,7 @@ test("an older build keeps every row around a Stop's event and a Resume, and fol if (!since.ok) { throw new Error(`expected rows, got reset ${since.reset}`) } - const rows: JournalRow[] = since.rows + const rows = since.rows.flatMap((row): JournalRow[] => (row.kind === 'skipped' ? [] : [row])) // The older build, after a downgrade, replays the same rows from its own database. const checkout = await materializeReleaseCheckout(BASELINE_REF) From 25bc79919b59ed8633ed0d9164ad0ab2cc513f20 Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:11:31 -0700 Subject: [PATCH 2/5] test(native-chat): a newer build's journal row kind survives reads, writes, rewinds and reopens --- .../journal-unknown-row-kind.test.ts | 293 ++++++++++++++++++ ...-session-stop-event-downgrade.unit.test.ts | 60 +++- 2 files changed, 352 insertions(+), 1 deletion(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts diff --git a/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts new file mode 100644 index 000000000000..99dfbdee76fb --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts @@ -0,0 +1,293 @@ +// A newer build's row kind is skipped and kept: it never reads as corruption, never renders, and +// survives every rewrite this build makes, so the newer build still reads it after an upgrade. + +import { mkdtemp, rm } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { + AGENT_JOURNAL_THREAD_SCOPE, + AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + type AgentJournalItemIdentity, + type AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { projectJournalBatch } from '../agent-session-wire/agent-session-journal-batch' +import { readAgentSessionHistory } from '../agent-session-wire/agent-session-history-page' +import { + closeTestJournalHostDatabase, + createTrackedJournalOpener, + insertTestJournalRowJson, + liveTestJournalRows, + openTestJournalHostDatabase +} from './journal-host-database-test-support' +import { parseJournalRow } from './journal-row-schema' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-newer-kind', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} +const SCOPE = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + +let root: string +const journals = createTrackedJournalOpener() + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-journal-newer-kind-')) +}) + +afterEach(async () => { + await journals.closeAll() + await rm(root, { recursive: true, force: true }) +}) + +function item(ordinal: number): AgentJournalItemIdentity { + return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal } +} + +/** What a newer build would write: a kind this build does not know, in an ordinary envelope. */ +function newerRow(epoch: string, seq: number, extra: Record = {}) { + return { + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + kind: 'future-mark', + epoch, + seq, + fence: 1, + ts: 5_000 + seq, + payload: { said: 'by a newer build', seq }, + ...extra + } +} + +function open() { + return journals.open({ identity: IDENTITY, stateDirectory: root }) +} + +/** Closes the chat, stores `rows` as a newer build or a bad write would, and reopens it as a + * restarted host would. */ +async function restartWith(rows: readonly { seq: number; json: string }[]) { + await journals.closeAll() + const { db } = openTestJournalHostDatabase(root) + for (const row of rows) { + insertTestJournalRowJson(db, IDENTITY.sessionId, row.seq, row.json, 5_000 + row.seq) + } + closeTestJournalHostDatabase(root) + return open() +} + +function stored(): { seq: number; json: Record }[] { + return liveTestJournalRows(openTestJournalHostDatabase(root).db, IDENTITY.sessionId).map( + (row) => ({ seq: row.seq, json: JSON.parse(row.rowJson) }) + ) +} + +function storedNewerRows() { + return stored().filter((row) => row.json.kind === 'future-mark') +} + +/** A journal of its anchor, two items, then a newer build's row at sequence 4. */ +async function journalWithNewerRow() { + const first = await open() + await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) + await first.appendItem(item(1), { kind: 'status', text: 'also before' }, SCOPE) + const json = JSON.stringify(newerRow(first.epoch, 4)) + return { journal: await restartWith([{ seq: 4, json }]), json } +} + +describe('parsing a row of a kind this build does not know', () => { + it('skips one whose envelope places it', () => { + expect(parseJournalRow(JSON.stringify(newerRow('epoch-1', 7)))).toEqual({ + ok: false, + unreadable: false, + skipped: { + kind: 'skipped', + storedKind: 'future-mark', + epoch: 'epoch-1', + seq: 7, + fence: 1, + ts: 5_007 + } + }) + }) + + it.each([ + ['no sequence', { seq: undefined }], + ['a fractional fence', { fence: 1.5 }], + ['an empty epoch', { epoch: '' }], + ['an empty kind', { kind: '' }] + ])('reads one with %s as malformed', (_name, broken) => { + const parsed = parseJournalRow(JSON.stringify(newerRow('epoch-1', 7, broken))) + expect(parsed).toEqual({ ok: false, unreadable: false }) + }) + + it('reads a known kind that fails its own checks as malformed, never skipped', () => { + const parsed = parseJournalRow( + JSON.stringify({ ...newerRow('epoch-1', 7), kind: 'item', itemId: 'x', revision: 1 }) + ) + expect(parsed).toEqual({ ok: false, unreadable: false }) + }) + + it('still latches a future schema version, whatever its kind', () => { + const parsed = parseJournalRow( + JSON.stringify(newerRow('epoch-1', 7, { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION + 1 })) + ) + expect(parsed).toEqual({ ok: false, unreadable: true }) + }) +}) + +describe("a journal holding a newer build's row", () => { + it('opens with every known row, keeps the row through a write and a reopen, and numbers past it', async () => { + const { journal, json } = await journalWithNewerRow() + expect(journal.repair).toEqual({ malformedRows: 0 }) + expect(journal.needsRebuild).toBe(false) + expect(journal.isReadOnly).toBe(false) + expect(journal.cursor().sequence).toBe(4) + expect(journal.snapshot().items.map((entry) => entry.itemId)).toEqual([ + 'codex:thread-1:turn-1:0', + 'codex:thread-1:turn-1:1' + ]) + + await journal.appendItem(item(2), { kind: 'status', text: 'after' }, SCOPE) + expect(journal.cursor().sequence).toBe(5) + + const reopened = await restartWith([]) + expect(reopened.repair).toEqual({ malformedRows: 0 }) + expect(reopened.cursor().sequence).toBe(5) + expect(reopened.snapshot().items).toHaveLength(3) + expect(stored().map((row) => row.seq)).toEqual([1, 2, 3, 4, 5]) + expect(storedNewerRows()).toEqual([{ seq: 4, json: JSON.parse(json) }]) + }) + + it('hands a reader a placeholder that keeps the sequence whole, so catch-up never resets', async () => { + const { journal } = await journalWithNewerRow() + await journal.appendItem(item(2), { kind: 'status', text: 'after' }, SCOPE) + const since = journal.readSince({ epoch: journal.epoch, sequence: 3 }) + if (!since.ok) { + throw new Error(`expected rows, got reset ${since.reset}`) + } + expect(since.rows.map((row) => [row.seq, row.kind])).toEqual([ + [4, 'skipped'], + [5, 'item'] + ]) + const projected = projectJournalBatch({ + rows: since.rows, + snapshot: journal.snapshot(), + afterSequence: 3 + }) + expect(projected.ok && projected.batch.items.map((entry) => entry.itemId)).toEqual([ + 'codex:thread-1:turn-1:2' + ]) + expect(projected.ok && projected.batch.removedItemIds).toEqual([]) + }) + + it('lets a forward history page step past it, even as the newest row', async () => { + const { journal } = await journalWithNewerRow() + const page = readAgentSessionHistory(journal, { + sessionId: IDENTITY.sessionId, + direction: 'after', + cursor: { epoch: journal.epoch, sequence: 3 }, + limit: 10 + }) + expect(page).toMatchObject({ + ok: true, + page: { items: [], hasNewer: false, window: { nextCursor: { sequence: 4 } } } + }) + }) + + it('is carried into the new epoch by a rewind, as written, and again by the next one', async () => { + const { journal, json } = await journalWithNewerRow() + await journal.replaceEpochItems('handle_forked', 1, [ + { identity: item(0), body: { kind: 'status', text: 'republished' } } + ]) + const firstRewind = journal.epoch + expect(storedNewerRows()).toEqual([ + { seq: 3, json: { ...JSON.parse(json), epoch: firstRewind, seq: 3 } } + ]) + + const reopened = await restartWith([]) + expect(reopened.repair).toEqual({ malformedRows: 0 }) + expect(reopened.needsRebuild).toBe(false) + expect(reopened.cursor()).toEqual({ epoch: firstRewind, sequence: 3 }) + + await reopened.replaceEpochItems('handle_forked', 1, []) + expect(storedNewerRows()).toEqual([ + { seq: 2, json: { ...JSON.parse(json), epoch: reopened.epoch, seq: 2 } } + ]) + await reopened.appendItem(item(5), { kind: 'status', text: 'after the rewinds' }, SCOPE) + expect(stored().map((row) => row.seq)).toEqual([1, 2, 3]) + }) + + it.each([ + ['before a Stop', 'newer-first', ['future-mark', 'tombstone']], + ['after a Stop', 'stop-first', ['tombstone', 'future-mark']] + ])('keeps its place %s when a rewind restates the Stop', async (_name, order, expected) => { + const first = await open() + await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) + if (order === 'stop-first') { + await first.appendStopEvent({ reason: 'user-stop', turnId: 'turn-1' }, 1) + } + const at = first.cursor().sequence + 1 + const journal = await restartWith([ + { seq: at, json: JSON.stringify(newerRow(first.epoch, at)) } + ]) + if (order === 'newer-first') { + await journal.appendStopEvent({ reason: 'user-stop', turnId: 'turn-1' }, 1) + } + + await journal.replaceEpochItems('handle_forked', 1, []) + + expect(stored().map((row) => row.json.kind)).toEqual(['epoch', ...expected]) + expect(journal.queuedMessages.pauses('host-a').map((pause) => pause.reason)).toEqual([ + 'stopped' + ]) + }) + + // Intended: a roll starts the chat over from nothing, and takes every row of every kind with it. + it('goes with every other row when the chat is rolled to a new epoch', async () => { + const { journal } = await journalWithNewerRow() + await journal.rollEpoch('handle_forked', 1) + expect(stored().map((row) => row.json.kind)).toEqual(['epoch']) + }) +}) + +describe('real corruption beside a newer row', () => { + it('still drops the suffix from a malformed row, and keeps the newer row before it', async () => { + const first = await open() + await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) + const epoch = first.epoch + const journal = await restartWith([ + { seq: 3, json: JSON.stringify(newerRow(epoch, 3)) }, + { seq: 4, json: '{"not":"a row"}' }, + { seq: 5, json: JSON.stringify(newerRow(epoch, 5)) } + ]) + expect(journal.repair).toEqual({ malformedRows: 1 }) + expect(journal.needsRebuild).toBe(true) + // 1..3 is the surviving prefix; 4 is the disclosure the repair appends. + expect(stored().map((row) => [row.seq, row.json.kind])).toEqual([ + [1, 'epoch'], + [2, 'item'], + [3, 'future-mark'], + [4, 'item'] + ]) + }) + + it('reads a newer kind whose envelope is broken as malformed and drops from it', async () => { + const first = await open() + await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) + const journal = await restartWith([ + { seq: 3, json: JSON.stringify(newerRow(first.epoch, 3, { fence: 'one' })) } + ]) + expect(journal.repair).toEqual({ malformedRows: 1 }) + expect(storedNewerRows()).toEqual([]) + }) + + it('treats a newer row naming the wrong sequence as a gap', async () => { + const first = await open() + await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) + const journal = await restartWith([{ seq: 3, json: JSON.stringify(newerRow(first.epoch, 9)) }]) + expect(journal.needsRebuild).toBe(true) + expect(journal.cursor().sequence).toBe(2) + }) +}) diff --git a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts index d9f6b68dfac2..ad7e6851ffde 100644 --- a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts +++ b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts @@ -4,14 +4,18 @@ import { join } from 'node:path' import { expect, test } from 'vitest' import { AGENT_JOURNAL_THREAD_SCOPE, + AGENT_SESSION_JOURNAL_SCHEMA_VERSION, type AgentJournalItemIdentity, type AgentSessionJournalIdentity } from '../../../src/shared/agent-session-journal-types' import Database from '../../../src/main/sqlite/sync-database' import { journalDatabasePath } from '../../../src/main/native-chat/agent-session-journal/journal-host-database' import { + closeTestJournalHostDatabase, createTrackedJournalOpener, - liveTestJournalRows + insertTestJournalRowJson, + liveTestJournalRows, + openTestJournalHostDatabase } from '../../../src/main/native-chat/agent-session-journal/journal-host-database-test-support' import type { JournalRow } from '../../../src/main/native-chat/agent-session-journal/journal-row-schema' import { importReleaseCheckoutModule, materializeReleaseCheckout } from './release-checkout' @@ -223,3 +227,57 @@ test("an older build opens this build's journal writable and appends to it; the rmSync(directory, { recursive: true, force: true }) } }, 120_000) + +// Why a Stop's event cannot have a row kind of its own yet: this build skips and keeps a kind it +// does not know, but a build from before that deletes the journal from it. Move the baseline to the +// first release that skips, and the older build keeps the row too. +test("this build keeps a newer build's row kind through a write and a reopen; a build before the skip deletes it", async () => { + const directory = mkdtempSync(join(tmpdir(), 'orca-newer-kind-downgrade-')) + const journals = createTrackedJournalOpener() + const newerKinds = () => storedRows(directory).filter((row) => row.includes('"future-mark"')) + try { + const scope = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } + const journal = await journals.open({ identity: IDENTITY, stateDirectory: directory }) + await journal.appendItem(item(0), { kind: 'status', text: 'before' }, scope) + const at = journal.cursor().sequence + 1 + const newer = JSON.stringify({ + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + kind: 'future-mark', + epoch: journal.epoch, + seq: at, + fence: 1, + ts: 2_000, + payload: { said: 'by a newer build' } + }) + await journals.closeAll() + insertTestJournalRowJson( + openTestJournalHostDatabase(directory).db, + IDENTITY.sessionId, + at, + newer + ) + closeTestJournalHostDatabase(directory) + + const reopened = await journals.open({ identity: IDENTITY, stateDirectory: directory }) + expect(reopened.repair).toEqual({ malformedRows: 0 }) + await reopened.appendItem(item(1), { kind: 'status', text: 'after' }, scope) + await journals.closeAll() + expect(newerKinds()).toEqual([newer]) + + const checkout = await materializeReleaseCheckout(WRITABLE_BASELINE_REF) + const support = await importReleaseCheckoutModule( + checkout, + `${JOURNAL}/journal-host-database-test-support.ts` + ) + const older = releaseExport<() => OlderOpener>(support, 'createTrackedJournalOpener')() + try { + await older.open({ identity: IDENTITY, stateDirectory: directory }) + } finally { + await older.closeAll() + } + expect(newerKinds()).toEqual([]) + } finally { + await journals.closeAll() + rmSync(directory, { recursive: true, force: true }) + } +}, 120_000) From bfa82d99abddda0118dc78a6b90c80d73f86951d Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 1 Oct 2026 17:41:14 -0700 Subject: [PATCH 3/5] fix(native-chat): an older Orca keeps an unknown journal row kind read-only unless its writer declared it skippable A row of a kind this build does not know, in a well-formed envelope, now latches the chat read-only with every row kept, the same way a newer row version does. It is read past only when its writer declared `ifUnknown` on the row: `skip` (a rewind drops it) or `carry` (a rewind carries it after the rebuilt history, epoch, seq and fence restamped). Every existing kind changes queue or turn state, so skipping by default would let an older build write from a wrong fold. - journal-row-kind-compatibility.ts: each kind states how older builds read it, typed over every row kind, so a new kind cannot be added without a declaration. - Rewind restates the Resume and Stop as before, then carries `carry` rows in source order; the restatement goes back to { lifted, liveStop }. - Replay treats a row whose body names another sequence than its stored key as malformed at the key, so the next write never collides with it; catch-up reads stop there too. --- .../journal-epoch-controller.ts | 6 +- .../journal-epoch-replacement.test.ts | 4 +- .../journal-epoch-replacement.ts | 51 +++--- .../agent-session-journal/journal-open.ts | 18 +- .../agent-session-journal/journal-reducer.ts | 10 +- .../journal-row-kind-compatibility.ts | 28 ++++ .../journal-row-schema.ts | 58 ++++--- .../journal-skipped-row.ts | 36 ++-- .../journal-store-collaborators.ts | 2 +- .../journal-unknown-row-kind.test.ts | 156 +++++++++++++----- .../queued-message-pause.ts | 13 +- src/shared/agent-session-journal-types.ts | 5 +- ...-session-stop-event-downgrade.unit.test.ts | 71 +++++--- 13 files changed, 296 insertions(+), 162 deletions(-) create mode 100644 src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts b/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts index 2537c93ba70f..93e1a6acf5b0 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts @@ -23,8 +23,8 @@ export class JournalEpochController { highestFence: () => number /** What of the live epoch's Stop and Resume a replacement restates. */ queuePauseRestatement: () => JournalQueuePauseRestatement - /** The live epoch's rows of a newer build's kind, which a replacement carries. */ - skippedSequences: () => readonly number[] + /** The live epoch's rows of a newer build's kind, declared `carry`. */ + carrySequences: () => readonly number[] cursor: () => AgentJournalCursor adopt: (loaded: JournalLoad) => void } @@ -73,7 +73,7 @@ export class JournalEpochController { fence, items, queuePause: this.deps.queuePauseRestatement(), - skippedSequences: this.deps.skippedSequences(), + carrySequences: this.deps.carrySequences(), now: this.deps.now, mintEpoch: this.deps.mintEpoch, onPublished: this.deps.adopt diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts index 5818f87f6b94..34ea93e91378 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts @@ -67,8 +67,8 @@ function replace(input: { reason: 'legacy_import', fence: 1, items: input.items, - queuePause: { liftedAt: 0, liveStop: null }, - skippedSequences: [], + queuePause: { lifted: false, liveStop: null }, + carrySequences: [], now, mintEpoch: () => `epoch-${clock}`, onPublished: input.onPublished ?? (() => undefined) diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts index dd6d470db315..6fd10bd9790a 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts @@ -3,8 +3,8 @@ // One transaction: discard the old epoch's rows, insert the epoch row plus the // replacement items, move the session projection, and retire any repair marker // — this republished history is exactly what the marker was holding out for. -// A newer build's rows this build skips are carried, not discarded: that build -// still reads them after a downgrade and upgrade. +// A newer build's rows its writer declared `carry` are carried, not discarded: +// that build still reads them after a downgrade and upgrade. import type { AgentJournalItemBody, @@ -58,8 +58,8 @@ export function replaceJournalEpoch(input: { /** Restated in the new epoch, or the rewind would release cards the person stopped, or bring * back a /clear pause they already lifted. */ queuePause: JournalQueuePauseRestatement - /** The live epoch's rows of a newer build's kind. */ - skippedSequences: readonly number[] + /** The live epoch's rows of a newer build's kind, declared `carry`. */ + carrySequences: readonly number[] now: () => number mintEpoch: () => string /** Called the instant the transaction commits, before any fallible follow-up. */ @@ -90,43 +90,34 @@ export function replaceJournalEpoch(input: { applyJournalRow(state, row) rows.push(row) } - const { sessionId } = input.identity - const { liftedAt, liveStop } = input.queuePause + const { lifted, liveStop } = input.queuePause const place = () => ({ state, seq: state.lastSequence + 1, fence: input.fence, ts: input.now() }) - const append = (row: JournalRow) => { + if (lifted) { + const row = buildJournalQueueResumeRow(place()) applyJournalRow(state, row) rows.push(row) } - const carried: JournalStoredRow[] = [] - // After the rebuilt history, in the order their source rows held: a newer build may read a - // carried row against a restated Stop or Resume. - const tail: { from: number; write: () => void }[] = [] - if (liftedAt > 0) { - tail.push({ from: liftedAt, write: () => append(buildJournalQueueResumeRow(place())) }) - } if (liveStop) { - const { sequence, event } = liveStop - const write = () => append(buildJournalStopEventRow({ ...place(), event })) - tail.push({ from: sequence, write }) + const row = buildJournalStopEventRow({ ...place(), event: liveStop }) + applyJournalRow(state, row) + rows.push(row) } - for (const stored of readLiveJournalRows(input.database, sessionId, input.skippedSequences)) { + + const { sessionId } = input.identity + // Last, in source order: their writer vouched they mean the same at any later place. + const carried: JournalStoredRow[] = [] + for (const stored of readLiveJournalRows(input.database, sessionId, input.carrySequences)) { const parsed = parseJournalRow(stored.rowJson) const skipped = !parsed.ok && parsed.skipped - if (!skipped) { + if (!skipped || skipped.ifUnknown !== 'carry') { continue } - const write = () => { - const at = { epoch, seq: state.lastSequence + 1, fence: input.fence } - const moved = restampSkippedJournalRow(stored.rowJson, skipped, at) - if (moved) { - applyJournalRow(state, moved.row) - carried.push({ epoch, seq: moved.row.seq, ts: moved.row.ts, rowJson: moved.rowJson }) - } + const at = { epoch, seq: state.lastSequence + 1, fence: input.fence } + const moved = restampSkippedJournalRow(stored.rowJson, skipped, at) + if (moved) { + applyJournalRow(state, moved.row) + carried.push({ epoch, seq: moved.row.seq, ts: moved.row.ts, rowJson: moved.rowJson }) } - tail.push({ from: stored.seq, write }) - } - for (const entry of tail.sort((left, right) => left.from - right.from)) { - entry.write() } input.database.transaction((db) => { diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index c18d5cf99eb4..1d7ce40e1da0 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -27,13 +27,14 @@ const FIRST_JOURNAL_SEQUENCE = 1 export type JournalLoad = { state: JournalReducerState - /** A row from a future schema was met: no writes, no deletion. */ + /** A row this build cannot read was met (a future schema, or a newer kind not declared + * skippable): no writes, no deletion. */ readOnly: boolean /** Set when the surviving prefix is unusable and the caller must roll the epoch. */ corrupt: boolean - /** Rows dropped because their body failed to parse (a future version latches `readOnly`, and - * a newer build's unknown kind is skipped and kept; neither counts here). The store discloses - * these in the timeline. */ + /** Rows dropped because their body failed to parse (an unreadable row latches `readOnly`, and a + * skippable newer kind is kept; neither counts here). The store discloses these in the + * timeline. */ malformedRows: number /** Directory-internal: the first sequence of an unusable suffix. The store * deletes from here before it accepts a write; a probe leaves it alone. */ @@ -94,9 +95,10 @@ export function startJournalRowFold(input: JournalRowFoldInput): { const add = (entry: { seq: number; rowJson: string }): boolean => { const parsed = parseJournalRow(entry.rowJson) - // A newer build's kind still holds its place, so it is sequence-checked like any row. + // A skipped newer kind still holds its place, so it is sequence-checked like any row. const row = journalReadRow(parsed) - if (!row) { + // A body naming another sequence than its key is malformed there: writes number past the key. + if (!row || row.seq !== entry.seq) { truncateFrom = entry.seq latched = !parsed.ok && parsed.unreadable malformedRows = latched ? 0 : 1 @@ -153,7 +155,7 @@ export function startJournalRowFold(input: JournalRowFoldInput): { return { add, finish } } -/** Rows after a cursor, in sequence order, a newer build's kinds as placeholders that keep the +/** Rows after a cursor, in sequence order, skippable newer kinds as placeholders that keep the * sequence whole. Stops at the first row this build cannot parse, exactly as replay does. */ export function readJournalRowsAfterCursor( db: Database.Database, @@ -165,7 +167,7 @@ export function readJournalRowsAfterCursor( const rows: JournalReadRow[] = [] for (const stored of readJournalRowsAfter(db, sessionId, epoch, afterSequence, limit)) { const row = journalReadRow(parseJournalRow(stored.rowJson)) - if (!row) { + if (!row || row.seq !== stored.seq) { break } rows.push(row) diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 693a139cc970..75da8afbf4ec 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -65,8 +65,8 @@ export type JournalReducerState = { latestPersonTurnSequence: number /** The latest person's Stop event and Resume, what the queue's pause is derived from. */ queuePauseMarks: JournalQueuePauseMarks - /** Sequences of a newer build's rows this build skips, so a replacement can carry them. */ - skippedSequences: number[] + /** Sequences of a newer build's rows declared `carry`, which a replacement carries. */ + carrySequences: number[] } export function createJournalReducerState(sessionId: string, epoch: string): JournalReducerState { @@ -87,7 +87,7 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou derivedTurnScope: new JournalDerivedTurnScope(), latestPersonTurnSequence: 0, queuePauseMarks: createJournalQueuePauseMarks(), - skippedSequences: [] + carrySequences: [] } } @@ -99,7 +99,9 @@ export function applyJournalRow(state: JournalReducerState, row: JournalReadRow) } // Holds its sequence and fence, the envelope facts every row has; its content is unknown here. if (row.kind === 'skipped') { - state.skippedSequences.push(row.seq) + if (row.ifUnknown === 'carry') { + state.carrySequences.push(row.seq) + } return } state.lastActivityAt = Math.max(state.lastActivityAt, row.ts) diff --git a/src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts b/src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts new file mode 100644 index 000000000000..6512b51b8727 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts @@ -0,0 +1,28 @@ +// The row kinds this build knows, and what a build that predates each does with it. When a kind +// may be declared skippable is the contract in `journal-row-schema.ts`'s header. + +import type { JournalRow } from './journal-row-schema' + +/** What a row's writer declares, as `ifUnknown`, a build that does not know its kind does with it. + * Persisted: never rename a value. */ +export const JOURNAL_ROW_IF_UNKNOWN = ['skip', 'carry'] as const +/** `skip`: read past it, and a rewrite drops it. `carry`: read past it, and a rewrite carries it. */ +export type JournalRowIfUnknown = (typeof JOURNAL_ROW_IF_UNKNOWN)[number] + +/** How each kind reads on an older supported build: `read` (every supported build knows it), its + * rows' declared `ifUnknown`, or `read-only`. A new kind must say. */ +const JOURNAL_ROW_KIND_ON_OLDER_BUILDS: Record< + JournalRow['kind'], + 'read' | 'read-only' | JournalRowIfUnknown +> = { + epoch: 'read', + item: 'read', + tombstone: 'read', + submission: 'read', + dispatch: 'read', + 'lifecycle-batch': 'read' +} + +export const JOURNAL_ROW_KINDS: ReadonlySet = new Set( + Object.keys(JOURNAL_ROW_KIND_ON_OLDER_BUILDS) +) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index a21afa453e52..8639658cd4ac 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -5,9 +5,15 @@ // UNREADABLE, not skippable: the caller must degrade to read-only rather than // render a partial timeline or compact past a row it cannot interpret. // -// A row whose KIND this build does not know, in an envelope it can place, is a -// newer build's addition at the same version: it is SKIPPED (no fold, no -// render) and kept on disk, so a newer kind needs no version bump. +// A row whose KIND this build does not know is UNREADABLE the same way and kept +// on disk, unless its writer declared `ifUnknown` on it: `skip` (read past it; +// a rewrite drops it) or `carry` (read past it; a rewrite carries it after the +// rebuilt history, its epoch, seq and fence restamped). A new kind may declare +// either only if a build that never sees it makes no different write, and a +// `carry` row only if it means the same at any later place in a new epoch (no +// sequence or epoch references). Otherwise it stays undeclared, or is written +// at a bumped `v`, and older builds go read-only. Each kind records its choice +// in `journal-row-kind-compatibility.ts`. import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' import { @@ -25,9 +31,10 @@ import { } from '../../../shared/agent-session-journal-schemas' import { isAdmissibleAgentSessionContextUsage } from '../../../shared/agent-session-context-usage-schema' import type { StructuredAgentSessionStopCause } from '../agent-session-wire/structured-agent-session-stop-cause' +import { JOURNAL_ROW_KINDS } from './journal-row-kind-compatibility' import { hasJournalRowEnvelope, - skippedJournalRow, + unknownKindJournalRow, type JournalSkippedRow } from './journal-skipped-row' @@ -88,9 +95,10 @@ export type JournalTombstoneRow = JournalRowBase & { queueResume?: true } -/** One Stop that took effect. Temporary carrier: a tombstone's extra key, because a host released - * before unknown kinds were skipped deletes the journal from the first row kind it does not know - * but ignores an unknown key; a row kind of its own once no supported build predates the skip. */ +/** One Stop that took effect. Temporary carrier: a tombstone's extra key, because an older host + * ignores an unknown key but deletes the journal from the first row kind it does not know. A Stop + * changes the queue and the turn's end, so no build may skip it: a kind of its own ships its reader + * first and is written once no supported build lacks that reader, or is written at a bumped `v`. */ export type JournalStopEvent = { /** Persisted: never rename an arm. Only `user-stop` pauses the queue. */ reason: StructuredAgentSessionStopCause @@ -188,27 +196,20 @@ export type JournalRowParse = | { ok: true; row: JournalRow } /** Malformed JSON or a shape this build rejects outright. */ | { ok: false; unreadable: false; skipped?: undefined } - /** A future schema version. The host must not write or compact this journal. */ + /** A future schema version, or a kind this build does not know that its writer did not declare + * `ifUnknown`. The host must not write or compact this journal. */ | { ok: false; unreadable: true; skipped?: undefined } - /** A kind this build does not know, in a well-formed envelope: skipped, never repaired away. */ + /** A kind this build does not know that its writer declared may be read past. */ | { ok: false; unreadable: false; skipped: JournalSkippedRow } -const ROW_KINDS = new Set([ - 'epoch', - 'item', - 'tombstone', - 'submission', - 'dispatch', - 'lifecycle-batch' -]) - export function serializeJournalRow(row: JournalRow): string { return JSON.stringify(row) } /** - * Parse one persisted line. Older versions are upcast; newer versions are - * reported as unreadable so the caller fails closed. + * Parse one persisted line. Older versions are upcast; newer versions, and newer + * kinds their writer did not declare skippable, are reported as unreadable so the + * caller fails closed. */ export function parseJournalRow(line: string): JournalRowParse { let parsed: unknown @@ -243,8 +244,13 @@ export function parseJournalRow(line: string): JournalRowParse { if (isJournalRow(upcast)) { return { ok: true, row: upcast } } - const skipped = skippedJournalRow(upcast, ROW_KINDS) - return skipped ? { ok: false, unreadable: false, skipped } : { ok: false, unreadable: false } + const unknownKind = unknownKindJournalRow(upcast) + if (unknownKind === 'unreadable') { + return { ok: false, unreadable: true } + } + return unknownKind + ? { ok: false, unreadable: false, skipped: unknownKind } + : { ok: false, unreadable: false } } /** Linkage ids this build cannot trust, removed from a row it still keeps. @@ -331,7 +337,7 @@ function isPlainObject(value: unknown): value is Record { * schema — their nested shapes are dereferenced unguarded all the way to the * rendered surface, so a JSON-valid corruption must fail here, not there. */ function isJournalRow(record: Record): record is JournalRow { - if (typeof record.kind !== 'string' || !ROW_KINDS.has(record.kind)) { + if (typeof record.kind !== 'string' || !JOURNAL_ROW_KINDS.has(record.kind)) { return false } if (!hasJournalRowEnvelope(record)) { @@ -377,7 +383,11 @@ function isJournalRow(record: Record): record is JournalRow { record.mutations.every(isLifecycleMutation) ) } - return typeof record.reason === 'string' && isPlainObject(record.providerHandle) + return ( + record.kind === 'epoch' && + typeof record.reason === 'string' && + isPlainObject(record.providerHandle) + ) } function isLifecycleMutation(value: unknown): value is JournalLifecycleMutation { diff --git a/src/main/native-chat/agent-session-journal/journal-skipped-row.ts b/src/main/native-chat/agent-session-journal/journal-skipped-row.ts index 120633e2d086..69078fd62006 100644 --- a/src/main/native-chat/agent-session-journal/journal-skipped-row.ts +++ b/src/main/native-chat/agent-session-journal/journal-skipped-row.ts @@ -1,6 +1,12 @@ -// A newer build's row of a kind this build does not know. Its envelope places it, so it holds its -// sequence and is kept on disk; nothing here reads what it says. +// A newer build's row of a kind this build does not know, which its writer declared an older build +// may read past. Its envelope places it, so it holds its sequence and is kept on disk; nothing here +// reads what it says. +import { + JOURNAL_ROW_IF_UNKNOWN, + JOURNAL_ROW_KINDS, + type JournalRowIfUnknown +} from './journal-row-kind-compatibility' import type { JournalRow, JournalRowParse } from './journal-row-schema' /** Where a skipped row sits: its place in the sequence and nothing to fold. In memory only; the @@ -9,6 +15,7 @@ export type JournalSkippedRow = { kind: 'skipped' /** The kind the row was stored under. */ storedKind: string + ifUnknown: JournalRowIfUnknown epoch: string seq: number fence: number @@ -23,27 +30,30 @@ export function journalReadRow(parsed: JournalRowParse): JournalReadRow | null { return parsed.ok ? parsed.row : (parsed.skipped ?? null) } -/** A row of an unknown kind whose envelope passes every check a known row's must. A known kind - * that fails its own checks is malformed, never skipped. */ -export function skippedJournalRow( - record: Record, - knownKinds: ReadonlySet -): JournalSkippedRow | null { - const { kind } = record - if (typeof kind !== 'string' || kind.length === 0 || knownKinds.has(kind)) { +/** A row of a kind this build does not know: skipped where its writer declared that safe, + * `unreadable` where it did not, and null where no envelope places it (malformed). */ +export function unknownKindJournalRow( + record: Record +): JournalSkippedRow | 'unreadable' | null { + const { kind, ifUnknown } = record + if (typeof kind !== 'string' || kind.length === 0 || JOURNAL_ROW_KINDS.has(kind)) { return null } if (!hasJournalRowEnvelope(record)) { return null } + const declared = JOURNAL_ROW_IF_UNKNOWN.find((value) => value === ifUnknown) + if (!declared) { + return 'unreadable' + } const { epoch, seq, fence, ts } = record - return { kind: 'skipped', storedKind: kind, epoch, seq, fence, ts } + return { kind: 'skipped', storedKind: kind, ifUnknown: declared, epoch, seq, fence, ts } } /** The fields every row carries whatever its kind: the chat's epoch, a place in it, and its writer. */ export function hasJournalRowEnvelope( record: Record -): record is Record & Omit { +): record is Record & Pick { return ( typeof record.epoch === 'string' && record.epoch.length > 0 && @@ -56,7 +66,7 @@ export function hasJournalRowEnvelope( ) } -/** A skipped row's stored text at a new place: its envelope restamped, everything else kept as its +/** A carried row's stored text at a new place: its envelope restamped, everything else kept as its * writer wrote it. */ export function restampSkippedJournalRow( rowJson: string, diff --git a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts index 0da8ef11b1f2..662e6cd4f7cc 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts @@ -78,7 +78,7 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal host.state().queuePauseMarks, host.state().latestPersonTurnSequence ), - skippedSequences: () => host.state().skippedSequences, + carrySequences: () => host.state().carrySequences, cursor: host.cursor, adopt: host.adopt }) diff --git a/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts index 99dfbdee76fb..9a502442255c 100644 --- a/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts @@ -1,5 +1,6 @@ -// A newer build's row kind is skipped and kept: it never reads as corruption, never renders, and -// survives every rewrite this build makes, so the newer build still reads it after an upgrade. +// A newer build's row kind is never repaired away. Undeclared, it latches this build read-only with +// every row kept. Declared `skip` or `carry` by its writer, it is read past and the chat stays +// writable; a rewind drops a `skip` row and carries a `carry` row into the new epoch. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -88,22 +89,23 @@ function storedNewerRows() { } /** A journal of its anchor, two items, then a newer build's row at sequence 4. */ -async function journalWithNewerRow() { +async function journalWithNewerRow(extra: Record = {}) { const first = await open() await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) await first.appendItem(item(1), { kind: 'status', text: 'also before' }, SCOPE) - const json = JSON.stringify(newerRow(first.epoch, 4)) + const json = JSON.stringify(newerRow(first.epoch, 4, extra)) return { journal: await restartWith([{ seq: 4, json }]), json } } describe('parsing a row of a kind this build does not know', () => { - it('skips one whose envelope places it', () => { - expect(parseJournalRow(JSON.stringify(newerRow('epoch-1', 7)))).toEqual({ + it.each(['skip', 'carry'] as const)('reads past one its writer declared %s', (ifUnknown) => { + expect(parseJournalRow(JSON.stringify(newerRow('epoch-1', 7, { ifUnknown })))).toEqual({ ok: false, unreadable: false, skipped: { kind: 'skipped', storedKind: 'future-mark', + ifUnknown, epoch: 'epoch-1', seq: 7, fence: 1, @@ -112,34 +114,75 @@ describe('parsing a row of a kind this build does not know', () => { }) }) + it.each([ + ['no declaration', {}], + ['a declaration this build does not know', { ifUnknown: 'drop' }], + ['a declaration of the wrong type', { ifUnknown: true }] + ])('reads one with %s as unreadable', (_name, declared) => { + const parsed = parseJournalRow(JSON.stringify(newerRow('epoch-1', 7, declared))) + expect(parsed).toEqual({ ok: false, unreadable: true }) + }) + it.each([ ['no sequence', { seq: undefined }], ['a fractional fence', { fence: 1.5 }], ['an empty epoch', { epoch: '' }], ['an empty kind', { kind: '' }] - ])('reads one with %s as malformed', (_name, broken) => { - const parsed = parseJournalRow(JSON.stringify(newerRow('epoch-1', 7, broken))) - expect(parsed).toEqual({ ok: false, unreadable: false }) + ])('reads one with %s as malformed, declared or not', (_name, broken) => { + for (const declared of [{}, { ifUnknown: 'skip' }]) { + const parsed = parseJournalRow( + JSON.stringify(newerRow('epoch-1', 7, { ...declared, ...broken })) + ) + expect(parsed).toEqual({ ok: false, unreadable: false }) + } }) - it('reads a known kind that fails its own checks as malformed, never skipped', () => { + it('reads a known kind that fails its own checks as malformed, whatever it declares', () => { const parsed = parseJournalRow( - JSON.stringify({ ...newerRow('epoch-1', 7), kind: 'item', itemId: 'x', revision: 1 }) + JSON.stringify({ + ...newerRow('epoch-1', 7, { ifUnknown: 'skip' }), + kind: 'item', + itemId: 'x', + revision: 1 + }) ) expect(parsed).toEqual({ ok: false, unreadable: false }) }) - it('still latches a future schema version, whatever its kind', () => { + it('still latches a future schema version, whatever it declares', () => { const parsed = parseJournalRow( - JSON.stringify(newerRow('epoch-1', 7, { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION + 1 })) + JSON.stringify( + newerRow('epoch-1', 7, { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION + 1, ifUnknown: 'skip' }) + ) ) expect(parsed).toEqual({ ok: false, unreadable: true }) }) }) -describe("a journal holding a newer build's row", () => { - it('opens with every known row, keeps the row through a write and a reopen, and numbers past it', async () => { +describe("a journal holding a newer build's undeclared row", () => { + it('opens read-only with the rows before it, refuses writes, and deletes nothing', async () => { const { journal, json } = await journalWithNewerRow() + const before = stored() + expect(journal.isReadOnly).toBe(true) + expect(journal.repair).toEqual({ malformedRows: 0 }) + expect(journal.snapshot().items.map((entry) => entry.itemId)).toEqual([ + 'codex:thread-1:turn-1:0', + 'codex:thread-1:turn-1:1' + ]) + await expect( + journal.appendItem(item(2), { kind: 'status', text: 'after' }, SCOPE) + ).rejects.toMatchObject({ code: 'journal_read_only' }) + + const reopened = await restartWith([]) + expect(reopened.isReadOnly).toBe(true) + expect(stored()).toEqual(before) + expect(storedNewerRows()).toEqual([{ seq: 4, json: JSON.parse(json) }]) + }) +}) + +describe("a journal holding a newer build's row declared skippable", () => { + it('opens with every known row, keeps the row through a write and a reopen, and numbers past it', async () => { + const { journal, json } = await journalWithNewerRow({ ifUnknown: 'skip' }) expect(journal.repair).toEqual({ malformedRows: 0 }) expect(journal.needsRebuild).toBe(false) expect(journal.isReadOnly).toBe(false) @@ -161,7 +204,7 @@ describe("a journal holding a newer build's row", () => { }) it('hands a reader a placeholder that keeps the sequence whole, so catch-up never resets', async () => { - const { journal } = await journalWithNewerRow() + const { journal } = await journalWithNewerRow({ ifUnknown: 'skip' }) await journal.appendItem(item(2), { kind: 'status', text: 'after' }, SCOPE) const since = journal.readSince({ epoch: journal.epoch, sequence: 3 }) if (!since.ok) { @@ -183,7 +226,7 @@ describe("a journal holding a newer build's row", () => { }) it('lets a forward history page step past it, even as the newest row', async () => { - const { journal } = await journalWithNewerRow() + const { journal } = await journalWithNewerRow({ ifUnknown: 'skip' }) const page = readAgentSessionHistory(journal, { sessionId: IDENTITY.sessionId, direction: 'after', @@ -196,8 +239,16 @@ describe("a journal holding a newer build's row", () => { }) }) - it('is carried into the new epoch by a rewind, as written, and again by the next one', async () => { - const { journal, json } = await journalWithNewerRow() + it('is dropped by a rewind when declared skip', async () => { + const { journal } = await journalWithNewerRow({ ifUnknown: 'skip' }) + await journal.replaceEpochItems('handle_forked', 1, [ + { identity: item(0), body: { kind: 'status', text: 'republished' } } + ]) + expect(stored().map((row) => row.json.kind)).toEqual(['epoch', 'item']) + }) + + it('is carried into the new epoch by a rewind when declared carry, and again by the next one', async () => { + const { journal, json } = await journalWithNewerRow({ ifUnknown: 'carry' }) await journal.replaceEpochItems('handle_forked', 1, [ { identity: item(0), body: { kind: 'status', text: 'republished' } } ]) @@ -219,48 +270,49 @@ describe("a journal holding a newer build's row", () => { expect(stored().map((row) => row.seq)).toEqual([1, 2, 3]) }) - it.each([ - ['before a Stop', 'newer-first', ['future-mark', 'tombstone']], - ['after a Stop', 'stop-first', ['tombstone', 'future-mark']] - ])('keeps its place %s when a rewind restates the Stop', async (_name, order, expected) => { + it("carries a row after the restated Stop, under the rewinder's fence", async () => { const first = await open() await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) - if (order === 'stop-first') { - await first.appendStopEvent({ reason: 'user-stop', turnId: 'turn-1' }, 1) - } const at = first.cursor().sequence + 1 const journal = await restartWith([ - { seq: at, json: JSON.stringify(newerRow(first.epoch, at)) } + { seq: at, json: JSON.stringify(newerRow(first.epoch, at, { ifUnknown: 'carry' })) } ]) - if (order === 'newer-first') { - await journal.appendStopEvent({ reason: 'user-stop', turnId: 'turn-1' }, 1) - } + await journal.appendStopEvent({ reason: 'user-stop', turnId: 'turn-1' }, 1) - await journal.replaceEpochItems('handle_forked', 1, []) + await journal.replaceEpochItems('handle_forked', 3, []) - expect(stored().map((row) => row.json.kind)).toEqual(['epoch', ...expected]) + expect(stored().map((row) => [row.json.kind, row.json.fence])).toEqual([ + ['epoch', 3], + ['tombstone', 3], + ['future-mark', 3] + ]) expect(journal.queuedMessages.pauses('host-a').map((pause) => pause.reason)).toEqual([ 'stopped' ]) + const reopened = await restartWith([]) + expect(reopened.repair).toEqual({ malformedRows: 0 }) + expect(reopened.cursor().sequence).toBe(3) + await reopened.appendItem(item(1), { kind: 'status', text: 'after' }, { ...SCOPE, fence: 3 }) + expect(reopened.cursor().sequence).toBe(4) }) // Intended: a roll starts the chat over from nothing, and takes every row of every kind with it. it('goes with every other row when the chat is rolled to a new epoch', async () => { - const { journal } = await journalWithNewerRow() + const { journal } = await journalWithNewerRow({ ifUnknown: 'carry' }) await journal.rollEpoch('handle_forked', 1) expect(stored().map((row) => row.json.kind)).toEqual(['epoch']) }) }) describe('real corruption beside a newer row', () => { - it('still drops the suffix from a malformed row, and keeps the newer row before it', async () => { + it('still drops the suffix from a malformed row, and keeps the skippable row before it', async () => { const first = await open() await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) const epoch = first.epoch const journal = await restartWith([ - { seq: 3, json: JSON.stringify(newerRow(epoch, 3)) }, + { seq: 3, json: JSON.stringify(newerRow(epoch, 3, { ifUnknown: 'skip' })) }, { seq: 4, json: '{"not":"a row"}' }, - { seq: 5, json: JSON.stringify(newerRow(epoch, 5)) } + { seq: 5, json: JSON.stringify(newerRow(epoch, 5, { ifUnknown: 'skip' })) } ]) expect(journal.repair).toEqual({ malformedRows: 1 }) expect(journal.needsRebuild).toBe(true) @@ -283,11 +335,29 @@ describe('real corruption beside a newer row', () => { expect(storedNewerRows()).toEqual([]) }) - it('treats a newer row naming the wrong sequence as a gap', async () => { - const first = await open() - await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) - const journal = await restartWith([{ seq: 3, json: JSON.stringify(newerRow(first.epoch, 9)) }]) - expect(journal.needsRebuild).toBe(true) - expect(journal.cursor().sequence).toBe(2) - }) + it.each([ + ['a known row', { kind: 'tombstone', itemId: 'gone', revision: 1 }], + ['a skippable newer row', { ifUnknown: 'skip' }] + ])( + 'drops %s whose body names another sequence than its key, and writes past it', + async (_name, shape) => { + const first = await open() + await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) + // Key 3, body 2: trusting the body would number the next write onto key 3. + const json = JSON.stringify(newerRow(first.epoch, 2, shape)) + const journal = await restartWith([{ seq: 3, json }]) + expect(journal.repair).toEqual({ malformedRows: 1 }) + expect(journal.needsRebuild).toBe(true) + expect(stored().map((row) => [row.seq, row.json.kind])).toEqual([ + [1, 'epoch'], + [2, 'item'], + [3, 'item'] + ]) + + await journal.appendItem(item(1), { kind: 'status', text: 'after' }, SCOPE) + const reopened = await restartWith([]) + expect(reopened.cursor().sequence).toBe(4) + expect(reopened.snapshot().items).toHaveLength(3) + } + ) }) diff --git a/src/main/native-chat/agent-session-journal/queued-message-pause.ts b/src/main/native-chat/agent-session-journal/queued-message-pause.ts index c5b463bc6d7f..9039760ba43c 100644 --- a/src/main/native-chat/agent-session-journal/queued-message-pause.ts +++ b/src/main/native-chat/agent-session-journal/queued-message-pause.ts @@ -85,12 +85,11 @@ export function journalUserStopInForce( } /** What a rewind's new epoch restates so its pauses read as they did: a lift of /clear's pause (a - * person's turn or a Resume happened) and the Stop still in force, each in the order its source - * row held, so the lift never ends the Stop. */ + * person's turn or a Resume happened), then the Stop still in force, in that order so the lift + * never ends the Stop. */ export type JournalQueuePauseRestatement = { - /** Where /clear's pause was lifted; 0 when it never was. */ - liftedAt: number - liveStop: JournalQueuePauseMarks['latestStop'] + lifted: boolean + liveStop: JournalStopEvent | null } export function journalQueuePauseRestatement( @@ -98,8 +97,8 @@ export function journalQueuePauseRestatement( latestPersonTurnSequence: number ): JournalQueuePauseRestatement { return { - liftedAt: Math.max(latestPersonTurnSequence, marks.resumedSequence), - liveStop: journalUserStopInForce(marks, latestPersonTurnSequence) + lifted: latestPersonTurnSequence > 0 || marks.resumedSequence > 0, + liveStop: journalUserStopInForce(marks, latestPersonTurnSequence)?.event ?? null } } diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index 548b263bbfec..cb9a44043c76 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -4,8 +4,9 @@ // so no class instances, Maps, or Dates. // // Rows are append-only. `schemaVersion` is upcast at read time and never -// rewritten in place, so a host that cannot read a row refuses to write the -// journal rather than skipping or compacting past it. +// rewritten in place, so a host that cannot read a row (a newer version, or a +// newer kind its writer did not declare skippable) refuses to write the journal +// rather than skipping or compacting past it. import type { UnreadAgentSessionFailureFact } from './agent-session-failure' import type { AgentSessionFailureRowWords } from './agent-session-failure-words' diff --git a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts index ad7e6851ffde..bc9068f60f13 100644 --- a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts +++ b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts @@ -13,6 +13,7 @@ import { journalDatabasePath } from '../../../src/main/native-chat/agent-session import { closeTestJournalHostDatabase, createTrackedJournalOpener, + deleteTestJournalRow, insertTestJournalRowJson, liveTestJournalRows, openTestJournalHostDatabase @@ -228,41 +229,61 @@ test("an older build opens this build's journal writable and appends to it; the } }, 120_000) -// Why a Stop's event cannot have a row kind of its own yet: this build skips and keeps a kind it -// does not know, but a build from before that deletes the journal from it. Move the baseline to the -// first release that skips, and the older build keeps the row too. -test("this build keeps a newer build's row kind through a write and a reopen; a build before the skip deletes it", async () => { +// Why a Stop's event cannot have a row kind of its own yet: a build before this one deletes the +// journal from a kind it does not know. This build keeps every row and goes read-only on an +// undeclared kind, and reads past one its writer declared `skip` or `carry`; a Stop changes the +// queue and the turn's end, so it may never declare either. A Stop kind ships its reader first and +// is written once no supported build lacks that reader, or is written at a bumped `v`. +test("this build keeps a newer build's row kind: read-only undeclared, writable declared; a build before it deletes it", async () => { const directory = mkdtempSync(join(tmpdir(), 'orca-newer-kind-downgrade-')) const journals = createTrackedJournalOpener() const newerKinds = () => storedRows(directory).filter((row) => row.includes('"future-mark"')) + const scope = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } try { - const scope = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } const journal = await journals.open({ identity: IDENTITY, stateDirectory: directory }) await journal.appendItem(item(0), { kind: 'status', text: 'before' }, scope) const at = journal.cursor().sequence + 1 - const newer = JSON.stringify({ - v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, - kind: 'future-mark', - epoch: journal.epoch, - seq: at, - fence: 1, - ts: 2_000, - payload: { said: 'by a newer build' } - }) + const newerRow = (declared: Record) => + JSON.stringify({ + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + kind: 'future-mark', + epoch: journal.epoch, + seq: at, + fence: 1, + ts: 2_000, + payload: { said: 'by a newer build' }, + ...declared + }) await journals.closeAll() - insertTestJournalRowJson( - openTestJournalHostDatabase(directory).db, - IDENTITY.sessionId, - at, - newer - ) - closeTestJournalHostDatabase(directory) + // Each case starts from the same journal: its rows before `at`, then the newer row. + const storeNewerRow = (json: string) => { + const { db } = openTestJournalHostDatabase(directory) + deleteTestJournalRow(db, IDENTITY.sessionId, at) + deleteTestJournalRow(db, IDENTITY.sessionId, at + 1) + insertTestJournalRowJson(db, IDENTITY.sessionId, at, json) + closeTestJournalHostDatabase(directory) + } - const reopened = await journals.open({ identity: IDENTITY, stateDirectory: directory }) - expect(reopened.repair).toEqual({ malformedRows: 0 }) - await reopened.appendItem(item(1), { kind: 'status', text: 'after' }, scope) + const undeclared = newerRow({}) + storeNewerRow(undeclared) + const readOnly = await journals.open({ identity: IDENTITY, stateDirectory: directory }) + expect(readOnly.isReadOnly).toBe(true) + await expect( + readOnly.appendItem(item(1), { kind: 'status', text: 'after' }, scope) + ).rejects.toMatchObject({ code: 'journal_read_only' }) await journals.closeAll() - expect(newerKinds()).toEqual([newer]) + expect(newerKinds()).toEqual([undeclared]) + + for (const ifUnknown of ['skip', 'carry']) { + const declared = newerRow({ ifUnknown }) + storeNewerRow(declared) + const writable = await journals.open({ identity: IDENTITY, stateDirectory: directory }) + expect(writable.isReadOnly).toBe(false) + expect(writable.repair).toEqual({ malformedRows: 0 }) + await writable.appendItem(item(1), { kind: 'status', text: 'after' }, scope) + await journals.closeAll() + expect(newerKinds()).toEqual([declared]) + } const checkout = await materializeReleaseCheckout(WRITABLE_BASELINE_REF) const support = await importReleaseCheckoutModule( From f30c080b5f220ebd0ccd2a4515120287ece6362e Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 1 Oct 2026 18:01:00 -0700 Subject: [PATCH 4/5] refactor(native-chat): drop the writer opt-in; an unknown journal row kind only latches read-only An older Orca now treats a row of a kind it does not know exactly like a row from a newer schema version: every row stays on disk and the chat opens read-only until an update. The writer-declared skip/carry opt-in, its in-memory placeholder, the carry through rewinds and the per-kind registry are removed: no current or planned kind could use them, and they can come with the first kind that may safely be read past. Kept: an unknown kind needs the envelope every row keeps (epoch, sequence, fence, timestamp), else it is damage as before; a row whose body names another sequence than its stored key is malformed at the key; the epoch row's validator names its kind. The schema header states the rule for adding a kind: keep the envelope, and either ship the reader first or bump `v`. --- .../agent-session-journal/journal-cursor.ts | 4 +- .../journal-epoch-controller.ts | 3 - .../journal-epoch-replacement.test.ts | 1 - .../journal-epoch-replacement.ts | 47 +-- .../agent-session-journal/journal-open.ts | 31 +- .../agent-session-journal/journal-reducer.ts | 17 +- .../journal-row-kind-compatibility.ts | 28 -- .../journal-row-schema.ts | 79 +++-- .../journal-row-table.ts | 23 -- .../journal-skipped-row.ts | 81 ----- .../journal-store-collaborators.ts | 1 - .../journal-store-contracts.ts | 4 +- .../journal-unknown-row-kind.test.ts | 327 +++++------------- .../agent-session-journal-batch.ts | 12 +- src/shared/agent-session-journal-types.ts | 4 +- ...-session-stop-event-downgrade.unit.test.ts | 76 ++-- 16 files changed, 198 insertions(+), 540 deletions(-) delete mode 100644 src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts delete mode 100644 src/main/native-chat/agent-session-journal/journal-skipped-row.ts diff --git a/src/main/native-chat/agent-session-journal/journal-cursor.ts b/src/main/native-chat/agent-session-journal/journal-cursor.ts index 9f7b8fd5e766..bd43623ae497 100644 --- a/src/main/native-chat/agent-session-journal/journal-cursor.ts +++ b/src/main/native-chat/agent-session-journal/journal-cursor.ts @@ -9,7 +9,7 @@ import type { AgentJournalResetReason } from '../../../shared/agent-session-journal-types' import type { JournalReadSince } from './journal-store-contracts' -import type { JournalReadRow } from './journal-skipped-row' +import type { JournalRow } from './journal-row-schema' export type JournalCursorRange = { epoch: string @@ -77,7 +77,7 @@ export function findSequenceGap( export function readJournalSince( source: { state: { epoch: string; lastSequence: number; oldestSequence: number } - rowsAfter: (afterSequence: number) => JournalReadRow[] + rowsAfter: (afterSequence: number) => JournalRow[] readOnly: boolean }, cursor: AgentJournalCursor, diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts b/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts index 93e1a6acf5b0..51d3a4b2fd53 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-controller.ts @@ -23,8 +23,6 @@ export class JournalEpochController { highestFence: () => number /** What of the live epoch's Stop and Resume a replacement restates. */ queuePauseRestatement: () => JournalQueuePauseRestatement - /** The live epoch's rows of a newer build's kind, declared `carry`. */ - carrySequences: () => readonly number[] cursor: () => AgentJournalCursor adopt: (loaded: JournalLoad) => void } @@ -73,7 +71,6 @@ export class JournalEpochController { fence, items, queuePause: this.deps.queuePauseRestatement(), - carrySequences: this.deps.carrySequences(), now: this.deps.now, mintEpoch: this.deps.mintEpoch, onPublished: this.deps.adopt diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts index 34ea93e91378..2992459a787a 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.test.ts @@ -68,7 +68,6 @@ function replace(input: { fence: 1, items: input.items, queuePause: { lifted: false, liveStop: null }, - carrySequences: [], now, mintEpoch: () => `epoch-${clock}`, onPublished: input.onPublished ?? (() => undefined) diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts index 6fd10bd9790a..70efc8912837 100644 --- a/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts +++ b/src/main/native-chat/agent-session-journal/journal-epoch-replacement.ts @@ -3,8 +3,6 @@ // One transaction: discard the old epoch's rows, insert the epoch row plus the // replacement items, move the session projection, and retire any repair marker // — this republished history is exactly what the marker was holding out for. -// A newer build's rows its writer declared `carry` are carried, not discarded: -// that build still reads them after a downgrade and upgrade. import type { AgentJournalItemBody, @@ -26,18 +24,10 @@ import type { JournalQueuePauseRestatement } from './queued-message-pause' import { deleteJournalEpochRows, insertJournalRow, - insertJournalStoredRow, publishJournalSessionEpoch, - readJournalRow, - readJournalSessionEpoch, - type JournalStoredRow + readJournalSessionEpoch } from './journal-row-table' -import { - parseJournalRow, - type AgentJournalEpochReason, - type JournalRow -} from './journal-row-schema' -import { restampSkippedJournalRow } from './journal-skipped-row' +import type { AgentJournalEpochReason, JournalRow } from './journal-row-schema' import { assertJournalFence } from './journal-write-guards' export type JournalReplacementItem = AgentJournalProducerLinkage & { @@ -58,8 +48,6 @@ export function replaceJournalEpoch(input: { /** Restated in the new epoch, or the rewind would release cards the person stopped, or bring * back a /clear pause they already lifted. */ queuePause: JournalQueuePauseRestatement - /** The live epoch's rows of a newer build's kind, declared `carry`. */ - carrySequences: readonly number[] now: () => number mintEpoch: () => string /** Called the instant the transaction commits, before any fallible follow-up. */ @@ -104,22 +92,6 @@ export function replaceJournalEpoch(input: { } const { sessionId } = input.identity - // Last, in source order: their writer vouched they mean the same at any later place. - const carried: JournalStoredRow[] = [] - for (const stored of readLiveJournalRows(input.database, sessionId, input.carrySequences)) { - const parsed = parseJournalRow(stored.rowJson) - const skipped = !parsed.ok && parsed.skipped - if (!skipped || skipped.ifUnknown !== 'carry') { - continue - } - const at = { epoch, seq: state.lastSequence + 1, fence: input.fence } - const moved = restampSkippedJournalRow(stored.rowJson, skipped, at) - if (moved) { - applyJournalRow(state, moved.row) - carried.push({ epoch, seq: moved.row.seq, ts: moved.row.ts, rowJson: moved.rowJson }) - } - } - input.database.transaction((db) => { const retired = readJournalSessionEpoch(db, sessionId) if (retired !== null) { @@ -129,9 +101,6 @@ export function replaceJournalEpoch(input: { for (const row of rows) { insertJournalRow(db, sessionId, row) } - for (const row of carried) { - insertJournalStoredRow(db, sessionId, row) - } publishJournalSessionEpoch(db, input.identity, epoch) }) @@ -141,15 +110,3 @@ export function replaceJournalEpoch(input: { state.oldestSequence = 1 input.onPublished({ state, readOnly: false, corrupt: false, malformedRows: 0 }) } - -/** The live epoch's rows at these sequences, as stored. */ -function readLiveJournalRows( - database: JournalHostDatabase, - sessionId: string, - sequences: readonly number[] -): JournalStoredRow[] { - const live = readJournalSessionEpoch(database.db, sessionId) - return live === null - ? [] - : sequences.flatMap((seq) => readJournalRow(database.db, sessionId, live, seq) ?? []) -} diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts index 1d7ce40e1da0..ff811983a253 100644 --- a/src/main/native-chat/agent-session-journal/journal-open.ts +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -20,20 +20,19 @@ import { import { JOURNAL_REPAIR_DISCLOSURE_ITEM_ID } from './journal-repair-disclosure' import { pendingJournalRepairSequence } from './journal-repair-marker' import { isJournalStopOrResumeRow, parseJournalRow, type JournalRow } from './journal-row-schema' -import { journalReadRow, type JournalReadRow } from './journal-skipped-row' /** Every epoch row is sequence 1, and no compaction moves that floor. */ const FIRST_JOURNAL_SEQUENCE = 1 export type JournalLoad = { state: JournalReducerState - /** A row this build cannot read was met (a future schema, or a newer kind not declared - * skippable): no writes, no deletion. */ + /** A row from a future schema, or of a kind this build does not know, was met: no writes, no + * deletion. */ readOnly: boolean /** Set when the surviving prefix is unusable and the caller must roll the epoch. */ corrupt: boolean - /** Rows dropped because their body failed to parse (an unreadable row latches `readOnly`, and a - * skippable newer kind is kept; neither counts here). The store discloses these in the + /** Rows dropped because they failed to parse or name another sequence than their key (an + * unreadable row latches `readOnly`, never counted here). The store discloses these in the * timeline. */ malformedRows: number /** Directory-internal: the first sequence of an unusable suffix. The store @@ -95,15 +94,14 @@ export function startJournalRowFold(input: JournalRowFoldInput): { const add = (entry: { seq: number; rowJson: string }): boolean => { const parsed = parseJournalRow(entry.rowJson) - // A skipped newer kind still holds its place, so it is sequence-checked like any row. - const row = journalReadRow(parsed) // A body naming another sequence than its key is malformed there: writes number past the key. - if (!row || row.seq !== entry.seq) { + if (!parsed.ok || parsed.row.seq !== entry.seq) { truncateFrom = entry.seq latched = !parsed.ok && parsed.unreadable malformedRows = latched ? 0 : 1 return false } + const row = parsed.row // Parse past a gap so an unreadable future row still latches read-only. if (gapSequence !== undefined) { return true @@ -124,9 +122,6 @@ export function startJournalRowFold(input: JournalRowFoldInput): { return true } applyJournalRow(state, row) - if (row.kind === 'skipped') { - return true - } const disclosure = row.kind === 'item' && row.itemId === JOURNAL_REPAIR_DISCLOSURE_ITEM_ID // A Stop or Resume is no history, so it never reads as a rebuilt or provider-backed epoch. if (!disclosure && !isJournalStopOrResumeRow(row)) { @@ -155,22 +150,22 @@ export function startJournalRowFold(input: JournalRowFoldInput): { return { add, finish } } -/** Rows after a cursor, in sequence order, skippable newer kinds as placeholders that keep the - * sequence whole. Stops at the first row this build cannot parse, exactly as replay does. */ +/** Rows after a cursor, in sequence order. Stops at the first row this build + * cannot parse, exactly as replay does. */ export function readJournalRowsAfterCursor( db: Database.Database, sessionId: string, epoch: string, afterSequence: number, limit?: number -): JournalReadRow[] { - const rows: JournalReadRow[] = [] +): JournalRow[] { + const rows: JournalRow[] = [] for (const stored of readJournalRowsAfter(db, sessionId, epoch, afterSequence, limit)) { - const row = journalReadRow(parseJournalRow(stored.rowJson)) - if (!row || row.seq !== stored.seq) { + const parsed = parseJournalRow(stored.rowJson) + if (!parsed.ok) { break } - rows.push(row) + rows.push(parsed.row) } return rows } diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts index 75da8afbf4ec..a7d6f150e70c 100644 --- a/src/main/native-chat/agent-session-journal/journal-reducer.ts +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -26,8 +26,7 @@ import { structuredAgentSessionPayloadFingerprint } from '../../../shared/struct import { JournalDerivedTurnScope } from './journal-derived-turn-scope' import { removeJournalItem, statedOrDerivedTurnScope, upsertJournalItem } from './journal-item-fold' import { journalItemRevisionIsStale } from './journal-item-revision' -import { isJournalStopOrResumeRow } from './journal-row-schema' -import type { JournalReadRow } from './journal-skipped-row' +import { isJournalStopOrResumeRow, type JournalRow } from './journal-row-schema' import { acceptSubmissionFromProviderItem, applyJournalSubmission } from './journal-submission-fold' import { applyJournalDispatchRow } from './journal-dispatch-reducer' import { isWriteFailureSubmission } from '../../../shared/structured-agent-session-dispatch-rejection' @@ -65,8 +64,6 @@ export type JournalReducerState = { latestPersonTurnSequence: number /** The latest person's Stop event and Resume, what the queue's pause is derived from. */ queuePauseMarks: JournalQueuePauseMarks - /** Sequences of a newer build's rows declared `carry`, which a replacement carries. */ - carrySequences: number[] } export function createJournalReducerState(sessionId: string, epoch: string): JournalReducerState { @@ -86,24 +83,16 @@ export function createJournalReducerState(sessionId: string, epoch: string): Jou appliedSettlementIds: new Set(), derivedTurnScope: new JournalDerivedTurnScope(), latestPersonTurnSequence: 0, - queuePauseMarks: createJournalQueuePauseMarks(), - carrySequences: [] + queuePauseMarks: createJournalQueuePauseMarks() } } -export function applyJournalRow(state: JournalReducerState, row: JournalReadRow): void { +export function applyJournalRow(state: JournalReducerState, row: JournalRow): void { state.lastSequence = Math.max(state.lastSequence, row.seq) state.highestFence = Math.max(state.highestFence, row.fence) if (row.kind === 'epoch') { return } - // Holds its sequence and fence, the envelope facts every row has; its content is unknown here. - if (row.kind === 'skipped') { - if (row.ifUnknown === 'carry') { - state.carrySequences.push(row.seq) - } - return - } state.lastActivityAt = Math.max(state.lastActivityAt, row.ts) if (row.kind === 'item') { if (journalItemRevisionIsStale(state, row.itemId, row.revision)) { diff --git a/src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts b/src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts deleted file mode 100644 index 6512b51b8727..000000000000 --- a/src/main/native-chat/agent-session-journal/journal-row-kind-compatibility.ts +++ /dev/null @@ -1,28 +0,0 @@ -// The row kinds this build knows, and what a build that predates each does with it. When a kind -// may be declared skippable is the contract in `journal-row-schema.ts`'s header. - -import type { JournalRow } from './journal-row-schema' - -/** What a row's writer declares, as `ifUnknown`, a build that does not know its kind does with it. - * Persisted: never rename a value. */ -export const JOURNAL_ROW_IF_UNKNOWN = ['skip', 'carry'] as const -/** `skip`: read past it, and a rewrite drops it. `carry`: read past it, and a rewrite carries it. */ -export type JournalRowIfUnknown = (typeof JOURNAL_ROW_IF_UNKNOWN)[number] - -/** How each kind reads on an older supported build: `read` (every supported build knows it), its - * rows' declared `ifUnknown`, or `read-only`. A new kind must say. */ -const JOURNAL_ROW_KIND_ON_OLDER_BUILDS: Record< - JournalRow['kind'], - 'read' | 'read-only' | JournalRowIfUnknown -> = { - epoch: 'read', - item: 'read', - tombstone: 'read', - submission: 'read', - dispatch: 'read', - 'lifecycle-batch': 'read' -} - -export const JOURNAL_ROW_KINDS: ReadonlySet = new Set( - Object.keys(JOURNAL_ROW_KIND_ON_OLDER_BUILDS) -) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 8639658cd4ac..3788ea1288b5 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -5,15 +5,13 @@ // UNREADABLE, not skippable: the caller must degrade to read-only rather than // render a partial timeline or compact past a row it cannot interpret. // -// A row whose KIND this build does not know is UNREADABLE the same way and kept -// on disk, unless its writer declared `ifUnknown` on it: `skip` (read past it; -// a rewrite drops it) or `carry` (read past it; a rewrite carries it after the -// rebuilt history, its epoch, seq and fence restamped). A new kind may declare -// either only if a build that never sees it makes no different write, and a -// `carry` row only if it means the same at any later place in a new epoch (no -// sequence or epoch references). Otherwise it stays undeclared, or is written -// at a bumped `v`, and older builds go read-only. Each kind records its choice -// in `journal-row-kind-compatibility.ts`. +// A row whose KIND this build does not know is UNREADABLE the same way, and kept +// on disk, when it has the envelope every row keeps: a non-empty `epoch`, an +// integer `seq` >= 1, an integer `fence` and a numeric `ts`. So a new kind keeps +// that envelope; changing the envelope is a `v` bump. Builds from before this +// rule delete the journal from an unknown kind, so a new kind either ships its +// reader first and is written only once no supported build lacks that reader, +// or is written at a bumped `v`. import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure' import { @@ -31,12 +29,6 @@ import { } from '../../../shared/agent-session-journal-schemas' import { isAdmissibleAgentSessionContextUsage } from '../../../shared/agent-session-context-usage-schema' import type { StructuredAgentSessionStopCause } from '../agent-session-wire/structured-agent-session-stop-cause' -import { JOURNAL_ROW_KINDS } from './journal-row-kind-compatibility' -import { - hasJournalRowEnvelope, - unknownKindJournalRow, - type JournalSkippedRow -} from './journal-skipped-row' /** Producer linkage rides the row BASE rather than the body: the two nested * prompt shapes are `.strict()`, so an unknown key on a body would make the @@ -95,10 +87,10 @@ export type JournalTombstoneRow = JournalRowBase & { queueResume?: true } -/** One Stop that took effect. Temporary carrier: a tombstone's extra key, because an older host - * ignores an unknown key but deletes the journal from the first row kind it does not know. A Stop - * changes the queue and the turn's end, so no build may skip it: a kind of its own ships its reader - * first and is written once no supported build lacks that reader, or is written at a bumped `v`. */ +/** One Stop that took effect. Temporary carrier: a tombstone's extra key, which every host ignores, + * where a host from before the header's rule deletes the journal from a kind it does not know. A + * kind of its own ships its reader first and is written once no supported build lacks that + * reader, or is written at a bumped `v`: an older host must never fold past a person's Stop. */ export type JournalStopEvent = { /** Persisted: never rename an arm. Only `user-stop` pauses the queue. */ reason: StructuredAgentSessionStopCause @@ -195,21 +187,27 @@ export type JournalRow = export type JournalRowParse = | { ok: true; row: JournalRow } /** Malformed JSON or a shape this build rejects outright. */ - | { ok: false; unreadable: false; skipped?: undefined } - /** A future schema version, or a kind this build does not know that its writer did not declare - * `ifUnknown`. The host must not write or compact this journal. */ - | { ok: false; unreadable: true; skipped?: undefined } - /** A kind this build does not know that its writer declared may be read past. */ - | { ok: false; unreadable: false; skipped: JournalSkippedRow } + | { ok: false; unreadable: false } + /** A future schema version, or a kind this build does not know. The host must not write or + * compact this journal. */ + | { ok: false; unreadable: true } + +const ROW_KINDS = new Set([ + 'epoch', + 'item', + 'tombstone', + 'submission', + 'dispatch', + 'lifecycle-batch' +]) export function serializeJournalRow(row: JournalRow): string { return JSON.stringify(row) } /** - * Parse one persisted line. Older versions are upcast; newer versions, and newer - * kinds their writer did not declare skippable, are reported as unreadable so the - * caller fails closed. + * Parse one persisted line. Older versions are upcast; newer versions and newer + * kinds are reported as unreadable so the caller fails closed. */ export function parseJournalRow(line: string): JournalRowParse { let parsed: unknown @@ -244,13 +242,10 @@ export function parseJournalRow(line: string): JournalRowParse { if (isJournalRow(upcast)) { return { ok: true, row: upcast } } - const unknownKind = unknownKindJournalRow(upcast) - if (unknownKind === 'unreadable') { - return { ok: false, unreadable: true } - } - return unknownKind - ? { ok: false, unreadable: false, skipped: unknownKind } - : { ok: false, unreadable: false } + // A newer build's kind is placed by the envelope every row keeps; one without it is damage. + const { kind } = upcast + const unknownKind = typeof kind === 'string' && kind !== '' && !ROW_KINDS.has(kind) + return { ok: false, unreadable: unknownKind && hasJournalRowEnvelope(upcast) } } /** Linkage ids this build cannot trust, removed from a row it still keeps. @@ -337,7 +332,7 @@ function isPlainObject(value: unknown): value is Record { * schema — their nested shapes are dereferenced unguarded all the way to the * rendered surface, so a JSON-valid corruption must fail here, not there. */ function isJournalRow(record: Record): record is JournalRow { - if (typeof record.kind !== 'string' || !JOURNAL_ROW_KINDS.has(record.kind)) { + if (typeof record.kind !== 'string' || !ROW_KINDS.has(record.kind)) { return false } if (!hasJournalRowEnvelope(record)) { @@ -390,6 +385,18 @@ function isJournalRow(record: Record): record is JournalRow { ) } +/** The fields every row keeps whatever its kind: its epoch, its place in it, its writer, its time. */ +function hasJournalRowEnvelope(record: Record): boolean { + return ( + typeof record.epoch === 'string' && + record.epoch !== '' && + Number.isInteger(record.seq) && + Number(record.seq) >= 1 && + Number.isInteger(record.fence) && + typeof record.ts === 'number' + ) +} + function isLifecycleMutation(value: unknown): value is JournalLifecycleMutation { if (!isPlainObject(value) || typeof value.itemId !== 'string') { return false diff --git a/src/main/native-chat/agent-session-journal/journal-row-table.ts b/src/main/native-chat/agent-session-journal/journal-row-table.ts index b6140c3d4423..4665a0571150 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-table.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-table.ts @@ -19,8 +19,6 @@ const INSERT_ROW = const SELECT_ROWS_AFTER = `SELECT seq, ts, row_json FROM journal_rows WHERE session_id = ? AND epoch = ? AND seq > ? ORDER BY seq ASC` const SELECT_ROWS_AFTER_LIMITED = `${SELECT_ROWS_AFTER} LIMIT ?` -const SELECT_ROW = `SELECT ts, row_json FROM journal_rows -WHERE session_id = ? AND epoch = ? AND seq = ?` const DELETE_SUFFIX = 'DELETE FROM journal_rows WHERE session_id = ? AND epoch = ? AND seq >= ?' const DELETE_EPOCH = 'DELETE FROM journal_rows WHERE session_id = ? AND epoch = ?' const DELETE_UNPUBLISHED = `DELETE FROM journal_rows WHERE session_id = ? @@ -50,27 +48,6 @@ export function insertJournalRow( return Buffer.byteLength(rowJson, 'utf8') } -/** A row stored as given, for one this build cannot type: a newer build's kind it carries. */ -export function insertJournalStoredRow( - db: Database.Database, - sessionId: string, - row: JournalStoredRow -): void { - db.prepare(INSERT_ROW).run(sessionId, row.epoch, row.seq, row.ts, row.rowJson) -} - -export function readJournalRow( - db: Database.Database, - sessionId: string, - epoch: string, - seq: number -): JournalStoredRow | null { - const row = db.prepare(SELECT_ROW).get(sessionId, epoch, seq) - return row && typeof row.ts === 'number' && typeof row.row_json === 'string' - ? { epoch, seq, ts: row.ts, rowJson: row.row_json } - : null -} - // Why pages, not `.iterate()`: a lazily consumed cursor pins a read snapshot for as long as the // consumer reduces, and a WAL checkpoint cannot pass an open snapshot. Each page is one completed // statement, so the consumer's memory is bounded by a page while no snapshot outlives a fetch. diff --git a/src/main/native-chat/agent-session-journal/journal-skipped-row.ts b/src/main/native-chat/agent-session-journal/journal-skipped-row.ts deleted file mode 100644 index 69078fd62006..000000000000 --- a/src/main/native-chat/agent-session-journal/journal-skipped-row.ts +++ /dev/null @@ -1,81 +0,0 @@ -// A newer build's row of a kind this build does not know, which its writer declared an older build -// may read past. Its envelope places it, so it holds its sequence and is kept on disk; nothing here -// reads what it says. - -import { - JOURNAL_ROW_IF_UNKNOWN, - JOURNAL_ROW_KINDS, - type JournalRowIfUnknown -} from './journal-row-kind-compatibility' -import type { JournalRow, JournalRowParse } from './journal-row-schema' - -/** Where a skipped row sits: its place in the sequence and nothing to fold. In memory only; the - * stored row stays as its writer wrote it. */ -export type JournalSkippedRow = { - kind: 'skipped' - /** The kind the row was stored under. */ - storedKind: string - ifUnknown: JournalRowIfUnknown - epoch: string - seq: number - fence: number - ts: number -} - -/** A row as a reader meets it: one this build folds, or a placeholder for one it skips. */ -export type JournalReadRow = JournalRow | JournalSkippedRow - -/** The row a reader folds or skips past; null where it must stop. */ -export function journalReadRow(parsed: JournalRowParse): JournalReadRow | null { - return parsed.ok ? parsed.row : (parsed.skipped ?? null) -} - -/** A row of a kind this build does not know: skipped where its writer declared that safe, - * `unreadable` where it did not, and null where no envelope places it (malformed). */ -export function unknownKindJournalRow( - record: Record -): JournalSkippedRow | 'unreadable' | null { - const { kind, ifUnknown } = record - if (typeof kind !== 'string' || kind.length === 0 || JOURNAL_ROW_KINDS.has(kind)) { - return null - } - if (!hasJournalRowEnvelope(record)) { - return null - } - const declared = JOURNAL_ROW_IF_UNKNOWN.find((value) => value === ifUnknown) - if (!declared) { - return 'unreadable' - } - const { epoch, seq, fence, ts } = record - return { kind: 'skipped', storedKind: kind, ifUnknown: declared, epoch, seq, fence, ts } -} - -/** The fields every row carries whatever its kind: the chat's epoch, a place in it, and its writer. */ -export function hasJournalRowEnvelope( - record: Record -): record is Record & Pick { - return ( - typeof record.epoch === 'string' && - record.epoch.length > 0 && - typeof record.seq === 'number' && - Number.isInteger(record.seq) && - record.seq >= 1 && - typeof record.fence === 'number' && - Number.isInteger(record.fence) && - typeof record.ts === 'number' - ) -} - -/** A carried row's stored text at a new place: its envelope restamped, everything else kept as its - * writer wrote it. */ -export function restampSkippedJournalRow( - rowJson: string, - skipped: JournalSkippedRow, - place: Pick -): { rowJson: string; row: JournalSkippedRow } | null { - const stored: unknown = JSON.parse(rowJson) - if (typeof stored !== 'object' || stored === null || Array.isArray(stored)) { - return null - } - return { rowJson: JSON.stringify({ ...stored, ...place }), row: { ...skipped, ...place } } -} diff --git a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts index 662e6cd4f7cc..b938efa3c5c9 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-collaborators.ts @@ -78,7 +78,6 @@ export function createJournalStoreCollaborators(host: JournalStoreHost): Journal host.state().queuePauseMarks, host.state().latestPersonTurnSequence ), - carrySequences: () => host.state().carrySequences, cursor: host.cursor, adopt: host.adopt }) diff --git a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts index 51f05892e557..85e6b99f9029 100644 --- a/src/main/native-chat/agent-session-journal/journal-store-contracts.ts +++ b/src/main/native-chat/agent-session-journal/journal-store-contracts.ts @@ -12,7 +12,7 @@ import type { } from '../../../shared/agent-session-journal-types' import type { JournalHostDatabase } from './journal-host-database' import type { JournalLifecycleMutationInput } from './journal-row-builders' -import type { JournalReadRow } from './journal-skipped-row' +import type { JournalRow } from './journal-row-schema' export type AgentSessionJournalOptions = { identity: AgentSessionJournalIdentity @@ -24,7 +24,7 @@ export type AgentSessionJournalOptions = { } export type JournalReadSince = - | { ok: true; rows: JournalReadRow[]; cursor: AgentJournalCursor } + | { ok: true; rows: JournalRow[]; cursor: AgentJournalCursor } | { ok: false; reset: AgentJournalResetReason } export type ResolveDispatchInput = { diff --git a/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts index 9a502442255c..2ea4e1c57de4 100644 --- a/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts @@ -1,6 +1,6 @@ -// A newer build's row kind is never repaired away. Undeclared, it latches this build read-only with -// every row kept. Declared `skip` or `carry` by its writer, it is read past and the chat stays -// writable; a rewind drops a `skip` row and carries a `carry` row into the new epoch. +// A newer build's row kind is never repaired away: like a newer row version, it latches this build +// read-only with every row kept, so the newer build still reads the whole chat after an upgrade. +// A row whose stored sequence and body disagree is damage at its stored sequence. import { mkdtemp, rm } from 'node:fs/promises' import { tmpdir } from 'node:os' @@ -12,8 +12,6 @@ import { type AgentJournalItemIdentity, type AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' -import { projectJournalBatch } from '../agent-session-wire/agent-session-journal-batch' -import { readAgentSessionHistory } from '../agent-session-wire/agent-session-history-page' import { closeTestJournalHostDatabase, createTrackedJournalOpener, @@ -48,7 +46,11 @@ function item(ordinal: number): AgentJournalItemIdentity { return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal } } -/** What a newer build would write: a kind this build does not know, in an ordinary envelope. */ +function itemIds(journal: { snapshot: () => { items: { itemId: string }[] } }): string[] { + return journal.snapshot().items.map((entry) => entry.itemId) +} + +/** What a newer build would write: a kind this build does not know, in the envelope every row keeps. */ function newerRow(epoch: string, seq: number, extra: Record = {}) { return { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, @@ -66,298 +68,159 @@ function open() { return journals.open({ identity: IDENTITY, stateDirectory: root }) } -/** Closes the chat, stores `rows` as a newer build or a bad write would, and reopens it as a - * restarted host would. */ -async function restartWith(rows: readonly { seq: number; json: string }[]) { +/** Closes the chat, applies `edit` to its stored rows as a newer build or a bad write would, and + * reopens it as a restarted host would. */ +async function restartAfter(edit: (put: (seq: number, json: string) => void) => void = () => {}) { await journals.closeAll() const { db } = openTestJournalHostDatabase(root) - for (const row of rows) { - insertTestJournalRowJson(db, IDENTITY.sessionId, row.seq, row.json, 5_000 + row.seq) - } + edit((seq, json) => { + db.prepare('DELETE FROM journal_rows WHERE session_id = ? AND seq = ?').run( + IDENTITY.sessionId, + seq + ) + insertTestJournalRowJson(db, IDENTITY.sessionId, seq, json, 5_000 + seq) + }) closeTestJournalHostDatabase(root) return open() } -function stored(): { seq: number; json: Record }[] { +function stored(): { seq: number; rowJson: string }[] { return liveTestJournalRows(openTestJournalHostDatabase(root).db, IDENTITY.sessionId).map( - (row) => ({ seq: row.seq, json: JSON.parse(row.rowJson) }) + (row) => ({ seq: row.seq, rowJson: row.rowJson }) ) } -function storedNewerRows() { - return stored().filter((row) => row.json.kind === 'future-mark') +function storedKinds(): [number, unknown][] { + return stored().map((row) => [row.seq, JSON.parse(row.rowJson).kind]) } -/** A journal of its anchor, two items, then a newer build's row at sequence 4. */ -async function journalWithNewerRow(extra: Record = {}) { +/** A journal of its anchor and one item, closed: the next row lands at sequence 3. */ +async function journalWithOneItem(): Promise { const first = await open() await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) - await first.appendItem(item(1), { kind: 'status', text: 'also before' }, SCOPE) - const json = JSON.stringify(newerRow(first.epoch, 4, extra)) - return { journal: await restartWith([{ seq: 4, json }]), json } + return first.epoch } describe('parsing a row of a kind this build does not know', () => { - it.each(['skip', 'carry'] as const)('reads past one its writer declared %s', (ifUnknown) => { - expect(parseJournalRow(JSON.stringify(newerRow('epoch-1', 7, { ifUnknown })))).toEqual({ + it('reads one in the envelope every row keeps as unreadable', () => { + expect(parseJournalRow(JSON.stringify(newerRow('epoch-1', 7)))).toEqual({ ok: false, - unreadable: false, - skipped: { - kind: 'skipped', - storedKind: 'future-mark', - ifUnknown, - epoch: 'epoch-1', - seq: 7, - fence: 1, - ts: 5_007 - } + unreadable: true }) }) - it.each([ - ['no declaration', {}], - ['a declaration this build does not know', { ifUnknown: 'drop' }], - ['a declaration of the wrong type', { ifUnknown: true }] - ])('reads one with %s as unreadable', (_name, declared) => { - const parsed = parseJournalRow(JSON.stringify(newerRow('epoch-1', 7, declared))) - expect(parsed).toEqual({ ok: false, unreadable: true }) - }) - it.each([ ['no sequence', { seq: undefined }], + ['a sequence of 0', { seq: 0 }], + ['a string sequence', { seq: '7' }], ['a fractional fence', { fence: 1.5 }], + ['a string timestamp', { ts: '5007' }], ['an empty epoch', { epoch: '' }], - ['an empty kind', { kind: '' }] - ])('reads one with %s as malformed, declared or not', (_name, broken) => { - for (const declared of [{}, { ifUnknown: 'skip' }]) { - const parsed = parseJournalRow( - JSON.stringify(newerRow('epoch-1', 7, { ...declared, ...broken })) - ) - expect(parsed).toEqual({ ok: false, unreadable: false }) - } + ['an empty kind', { kind: '' }], + ['a kind that is not a string', { kind: 7 }] + ])('reads one with %s as malformed', (_name, broken) => { + const parsed = parseJournalRow(JSON.stringify(newerRow('epoch-1', 7, broken))) + expect(parsed).toEqual({ ok: false, unreadable: false }) }) - it('reads a known kind that fails its own checks as malformed, whatever it declares', () => { + it('reads a known kind that fails its own checks as malformed', () => { const parsed = parseJournalRow( - JSON.stringify({ - ...newerRow('epoch-1', 7, { ifUnknown: 'skip' }), - kind: 'item', - itemId: 'x', - revision: 1 - }) + JSON.stringify({ ...newerRow('epoch-1', 7), kind: 'item', itemId: 'x', revision: 1 }) ) expect(parsed).toEqual({ ok: false, unreadable: false }) }) - it('still latches a future schema version, whatever it declares', () => { + it('reads a future schema version as unreadable before it looks at the kind', () => { const parsed = parseJournalRow( - JSON.stringify( - newerRow('epoch-1', 7, { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION + 1, ifUnknown: 'skip' }) - ) + JSON.stringify(newerRow('epoch-1', 7, { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION + 1 })) ) expect(parsed).toEqual({ ok: false, unreadable: true }) }) }) -describe("a journal holding a newer build's undeclared row", () => { - it('opens read-only with the rows before it, refuses writes, and deletes nothing', async () => { - const { journal, json } = await journalWithNewerRow() - const before = stored() +describe("a journal holding a newer build's row kind", () => { + it('opens read-only with the rows before it, refuses writes, keeps every row, and reads whole once the kind is known', async () => { + const first = await open() + for (const ordinal of [0, 1, 2, 3]) { + await first.appendItem(item(ordinal), { kind: 'status', text: `item ${ordinal}` }, SCOPE) + } + const epoch = first.epoch + await journals.closeAll() + const known = stored().find((row) => row.seq === 3)?.rowJson ?? '' + const newer = JSON.stringify(newerRow(epoch, 3)) + + const journal = await restartAfter((put) => put(3, newer)) + const latched = stored() expect(journal.isReadOnly).toBe(true) expect(journal.repair).toEqual({ malformedRows: 0 }) - expect(journal.snapshot().items.map((entry) => entry.itemId)).toEqual([ - 'codex:thread-1:turn-1:0', - 'codex:thread-1:turn-1:1' - ]) + expect(itemIds(journal)).toEqual(['codex:thread-1:turn-1:0']) await expect( - journal.appendItem(item(2), { kind: 'status', text: 'after' }, SCOPE) + journal.appendItem(item(4), { kind: 'status', text: 'after' }, SCOPE) ).rejects.toMatchObject({ code: 'journal_read_only' }) - const reopened = await restartWith([]) + const reopened = await restartAfter() expect(reopened.isReadOnly).toBe(true) - expect(stored()).toEqual(before) - expect(storedNewerRows()).toEqual([{ seq: 4, json: JSON.parse(json) }]) - }) -}) - -describe("a journal holding a newer build's row declared skippable", () => { - it('opens with every known row, keeps the row through a write and a reopen, and numbers past it', async () => { - const { journal, json } = await journalWithNewerRow({ ifUnknown: 'skip' }) - expect(journal.repair).toEqual({ malformedRows: 0 }) - expect(journal.needsRebuild).toBe(false) - expect(journal.isReadOnly).toBe(false) - expect(journal.cursor().sequence).toBe(4) - expect(journal.snapshot().items.map((entry) => entry.itemId)).toEqual([ - 'codex:thread-1:turn-1:0', - 'codex:thread-1:turn-1:1' - ]) - - await journal.appendItem(item(2), { kind: 'status', text: 'after' }, SCOPE) - expect(journal.cursor().sequence).toBe(5) - - const reopened = await restartWith([]) - expect(reopened.repair).toEqual({ malformedRows: 0 }) - expect(reopened.cursor().sequence).toBe(5) - expect(reopened.snapshot().items).toHaveLength(3) - expect(stored().map((row) => row.seq)).toEqual([1, 2, 3, 4, 5]) - expect(storedNewerRows()).toEqual([{ seq: 4, json: JSON.parse(json) }]) - }) - - it('hands a reader a placeholder that keeps the sequence whole, so catch-up never resets', async () => { - const { journal } = await journalWithNewerRow({ ifUnknown: 'skip' }) - await journal.appendItem(item(2), { kind: 'status', text: 'after' }, SCOPE) - const since = journal.readSince({ epoch: journal.epoch, sequence: 3 }) - if (!since.ok) { - throw new Error(`expected rows, got reset ${since.reset}`) - } - expect(since.rows.map((row) => [row.seq, row.kind])).toEqual([ - [4, 'skipped'], - [5, 'item'] - ]) - const projected = projectJournalBatch({ - rows: since.rows, - snapshot: journal.snapshot(), - afterSequence: 3 - }) - expect(projected.ok && projected.batch.items.map((entry) => entry.itemId)).toEqual([ - 'codex:thread-1:turn-1:2' - ]) - expect(projected.ok && projected.batch.removedItemIds).toEqual([]) - }) - - it('lets a forward history page step past it, even as the newest row', async () => { - const { journal } = await journalWithNewerRow({ ifUnknown: 'skip' }) - const page = readAgentSessionHistory(journal, { - sessionId: IDENTITY.sessionId, - direction: 'after', - cursor: { epoch: journal.epoch, sequence: 3 }, - limit: 10 - }) - expect(page).toMatchObject({ - ok: true, - page: { items: [], hasNewer: false, window: { nextCursor: { sequence: 4 } } } - }) + expect(stored()).toEqual(latched) + expect(stored().find((row) => row.seq === 3)?.rowJson).toBe(newer) + + // Stand-in for the newer build after an upgrade: the same place holds a kind it reads. + const upgraded = await restartAfter((put) => put(3, known)) + expect(upgraded.isReadOnly).toBe(false) + expect(itemIds(upgraded)).toHaveLength(4) + await upgraded.appendItem(item(4), { kind: 'status', text: 'after' }, SCOPE) + expect(upgraded.cursor().sequence).toBe(6) }) - it('is dropped by a rewind when declared skip', async () => { - const { journal } = await journalWithNewerRow({ ifUnknown: 'skip' }) - await journal.replaceEpochItems('handle_forked', 1, [ - { identity: item(0), body: { kind: 'status', text: 'republished' } } - ]) - expect(stored().map((row) => row.json.kind)).toEqual(['epoch', 'item']) - }) - - it('is carried into the new epoch by a rewind when declared carry, and again by the next one', async () => { - const { journal, json } = await journalWithNewerRow({ ifUnknown: 'carry' }) - await journal.replaceEpochItems('handle_forked', 1, [ - { identity: item(0), body: { kind: 'status', text: 'republished' } } - ]) - const firstRewind = journal.epoch - expect(storedNewerRows()).toEqual([ - { seq: 3, json: { ...JSON.parse(json), epoch: firstRewind, seq: 3 } } - ]) - - const reopened = await restartWith([]) - expect(reopened.repair).toEqual({ malformedRows: 0 }) - expect(reopened.needsRebuild).toBe(false) - expect(reopened.cursor()).toEqual({ epoch: firstRewind, sequence: 3 }) - - await reopened.replaceEpochItems('handle_forked', 1, []) - expect(storedNewerRows()).toEqual([ - { seq: 2, json: { ...JSON.parse(json), epoch: reopened.epoch, seq: 2 } } - ]) - await reopened.appendItem(item(5), { kind: 'status', text: 'after the rewinds' }, SCOPE) - expect(stored().map((row) => row.seq)).toEqual([1, 2, 3]) - }) - - it("carries a row after the restated Stop, under the rewinder's fence", async () => { - const first = await open() - await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) - const at = first.cursor().sequence + 1 - const journal = await restartWith([ - { seq: at, json: JSON.stringify(newerRow(first.epoch, at, { ifUnknown: 'carry' })) } - ]) - await journal.appendStopEvent({ reason: 'user-stop', turnId: 'turn-1' }, 1) - - await journal.replaceEpochItems('handle_forked', 3, []) - - expect(stored().map((row) => [row.json.kind, row.json.fence])).toEqual([ - ['epoch', 3], - ['tombstone', 3], - ['future-mark', 3] - ]) - expect(journal.queuedMessages.pauses('host-a').map((pause) => pause.reason)).toEqual([ - 'stopped' - ]) - const reopened = await restartWith([]) - expect(reopened.repair).toEqual({ malformedRows: 0 }) - expect(reopened.cursor().sequence).toBe(3) - await reopened.appendItem(item(1), { kind: 'status', text: 'after' }, { ...SCOPE, fence: 3 }) - expect(reopened.cursor().sequence).toBe(4) - }) - - // Intended: a roll starts the chat over from nothing, and takes every row of every kind with it. - it('goes with every other row when the chat is rolled to a new epoch', async () => { - const { journal } = await journalWithNewerRow({ ifUnknown: 'carry' }) - await journal.rollEpoch('handle_forked', 1) - expect(stored().map((row) => row.json.kind)).toEqual(['epoch']) - }) -}) - -describe('real corruption beside a newer row', () => { - it('still drops the suffix from a malformed row, and keeps the skippable row before it', async () => { - const first = await open() - await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) - const epoch = first.epoch - const journal = await restartWith([ - { seq: 3, json: JSON.stringify(newerRow(epoch, 3, { ifUnknown: 'skip' })) }, - { seq: 4, json: '{"not":"a row"}' }, - { seq: 5, json: JSON.stringify(newerRow(epoch, 5, { ifUnknown: 'skip' })) } - ]) - expect(journal.repair).toEqual({ malformedRows: 1 }) - expect(journal.needsRebuild).toBe(true) - // 1..3 is the surviving prefix; 4 is the disclosure the repair appends. - expect(stored().map((row) => [row.seq, row.json.kind])).toEqual([ + it('stays read-only with nothing deleted when the row names another sequence than its key', async () => { + const epoch = await journalWithOneItem() + const journal = await restartAfter((put) => put(3, JSON.stringify(newerRow(epoch, 9)))) + expect(journal.isReadOnly).toBe(true) + expect(storedKinds()).toEqual([ [1, 'epoch'], [2, 'item'], - [3, 'future-mark'], - [4, 'item'] + [3, 'future-mark'] ]) }) it('reads a newer kind whose envelope is broken as malformed and drops from it', async () => { - const first = await open() - await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) - const journal = await restartWith([ - { seq: 3, json: JSON.stringify(newerRow(first.epoch, 3, { fence: 'one' })) } - ]) + const epoch = await journalWithOneItem() + const journal = await restartAfter((put) => + put(3, JSON.stringify(newerRow(epoch, 3, { fence: 'one' }))) + ) + expect(journal.isReadOnly).toBe(false) expect(journal.repair).toEqual({ malformedRows: 1 }) - expect(storedNewerRows()).toEqual([]) + // 3 is the disclosure the repair appends where the broken row was. + expect(storedKinds()).toEqual([ + [1, 'epoch'], + [2, 'item'], + [3, 'item'] + ]) }) +}) +describe('a row whose body names another sequence than its stored key', () => { it.each([ - ['a known row', { kind: 'tombstone', itemId: 'gone', revision: 1 }], - ['a skippable newer row', { ifUnknown: 'skip' }] + ['an earlier', 2], + ['a later', 9] ])( - 'drops %s whose body names another sequence than its key, and writes past it', - async (_name, shape) => { - const first = await open() - await first.appendItem(item(0), { kind: 'status', text: 'before' }, SCOPE) - // Key 3, body 2: trusting the body would number the next write onto key 3. - const json = JSON.stringify(newerRow(first.epoch, 2, shape)) - const journal = await restartWith([{ seq: 3, json }]) + 'is dropped at its key when it names %s one, and the next write succeeds', + async (_name, seq) => { + const epoch = await journalWithOneItem() + const tombstone = { ...newerRow(epoch, seq), kind: 'tombstone', itemId: 'gone', revision: 1 } + const journal = await restartAfter((put) => put(3, JSON.stringify(tombstone))) expect(journal.repair).toEqual({ malformedRows: 1 }) expect(journal.needsRebuild).toBe(true) - expect(stored().map((row) => [row.seq, row.json.kind])).toEqual([ + // The valid row at 2 stays; 3 is the disclosure the repair appends. + expect(storedKinds()).toEqual([ [1, 'epoch'], [2, 'item'], [3, 'item'] ]) await journal.appendItem(item(1), { kind: 'status', text: 'after' }, SCOPE) - const reopened = await restartWith([]) + const reopened = await restartAfter() expect(reopened.cursor().sequence).toBe(4) - expect(reopened.snapshot().items).toHaveLength(3) + expect(itemIds(reopened)).toHaveLength(3) } ) }) diff --git a/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts b/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts index aed18f517287..42af5f830420 100644 --- a/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts +++ b/src/main/native-chat/agent-session-wire/agent-session-journal-batch.ts @@ -14,8 +14,10 @@ import type { } from '../../../shared/agent-session-journal-types' import type { AgentSessionJournalBatch } from '../../../shared/agent-session-wire' import { findSequenceGap } from '../agent-session-journal/journal-cursor' -import { isJournalStopOrResumeRow } from '../agent-session-journal/journal-row-schema' -import type { JournalReadRow } from '../agent-session-journal/journal-skipped-row' +import { + isJournalStopOrResumeRow, + type JournalRow +} from '../agent-session-journal/journal-row-schema' export type JournalBatchProjection = | { ok: true; batch: AgentSessionJournalBatch } @@ -24,7 +26,7 @@ export type JournalBatchProjection = | { ok: false; reset: 'journal_gap' } export function projectJournalBatch(input: { - rows: readonly JournalReadRow[] + rows: readonly JournalRow[] snapshot: AgentJournalSnapshot /** Sequence the subscriber has already applied. */ afterSequence: number @@ -41,10 +43,6 @@ export function projectJournalBatch(input: { const touchedItemIds = new Set() const touchedClientMessageIds = new Set() for (const row of input.rows) { - if (row.kind === 'skipped') { - // A newer build's row: it keeps the sequence whole and touches nothing this build renders. - continue - } if (row.kind === 'lifecycle-batch') { for (const mutation of row.mutations) { touchedItemIds.add( diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts index cb9a44043c76..e5f8125a5c0f 100644 --- a/src/shared/agent-session-journal-types.ts +++ b/src/shared/agent-session-journal-types.ts @@ -5,8 +5,8 @@ // // Rows are append-only. `schemaVersion` is upcast at read time and never // rewritten in place, so a host that cannot read a row (a newer version, or a -// newer kind its writer did not declare skippable) refuses to write the journal -// rather than skipping or compacting past it. +// newer kind) refuses to write the journal rather than skipping or compacting +// past it. import type { UnreadAgentSessionFailureFact } from './agent-session-failure' import type { AgentSessionFailureRowWords } from './agent-session-failure-words' diff --git a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts index bc9068f60f13..e585191c1a05 100644 --- a/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts +++ b/tests/e2e/cross-version-wire/agent-session-stop-event-downgrade.unit.test.ts @@ -13,7 +13,6 @@ import { journalDatabasePath } from '../../../src/main/native-chat/agent-session import { closeTestJournalHostDatabase, createTrackedJournalOpener, - deleteTestJournalRow, insertTestJournalRowJson, liveTestJournalRows, openTestJournalHostDatabase @@ -82,7 +81,7 @@ test("an older build keeps every row around a Stop's event and a Resume, and fol if (!since.ok) { throw new Error(`expected rows, got reset ${since.reset}`) } - const rows = since.rows.flatMap((row): JournalRow[] => (row.kind === 'skipped' ? [] : [row])) + const rows: JournalRow[] = since.rows // The older build, after a downgrade, replays the same rows from its own database. const checkout = await materializeReleaseCheckout(BASELINE_REF) @@ -229,61 +228,48 @@ test("an older build opens this build's journal writable and appends to it; the } }, 120_000) -// Why a Stop's event cannot have a row kind of its own yet: a build before this one deletes the -// journal from a kind it does not know. This build keeps every row and goes read-only on an -// undeclared kind, and reads past one its writer declared `skip` or `carry`; a Stop changes the -// queue and the turn's end, so it may never declare either. A Stop kind ships its reader first and -// is written once no supported build lacks that reader, or is written at a bumped `v`. -test("this build keeps a newer build's row kind: read-only undeclared, writable declared; a build before it deletes it", async () => { +// Why a Stop's event cannot have a row kind of its own yet: this build keeps a kind it does not know +// and goes read-only, but a build from before that deletes the journal from it. So a Stop kind ships +// its reader first and is written once no supported build lacks that reader, or is written at a +// bumped `v`. Move the baseline to the first release with this rule, and the older build keeps the +// row too. +test("this build keeps a newer build's row kind and goes read-only; a build before it deletes it", async () => { const directory = mkdtempSync(join(tmpdir(), 'orca-newer-kind-downgrade-')) const journals = createTrackedJournalOpener() const newerKinds = () => storedRows(directory).filter((row) => row.includes('"future-mark"')) - const scope = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } try { + const scope = { fence: 1, turnScope: AGENT_JOURNAL_THREAD_SCOPE } const journal = await journals.open({ identity: IDENTITY, stateDirectory: directory }) await journal.appendItem(item(0), { kind: 'status', text: 'before' }, scope) const at = journal.cursor().sequence + 1 - const newerRow = (declared: Record) => - JSON.stringify({ - v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, - kind: 'future-mark', - epoch: journal.epoch, - seq: at, - fence: 1, - ts: 2_000, - payload: { said: 'by a newer build' }, - ...declared - }) + const newer = JSON.stringify({ + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + kind: 'future-mark', + epoch: journal.epoch, + seq: at, + fence: 1, + ts: 2_000, + payload: { said: 'by a newer build' } + }) await journals.closeAll() - // Each case starts from the same journal: its rows before `at`, then the newer row. - const storeNewerRow = (json: string) => { - const { db } = openTestJournalHostDatabase(directory) - deleteTestJournalRow(db, IDENTITY.sessionId, at) - deleteTestJournalRow(db, IDENTITY.sessionId, at + 1) - insertTestJournalRowJson(db, IDENTITY.sessionId, at, json) - closeTestJournalHostDatabase(directory) - } + insertTestJournalRowJson( + openTestJournalHostDatabase(directory).db, + IDENTITY.sessionId, + at, + newer + ) + closeTestJournalHostDatabase(directory) + const rowsBefore = storedRows(directory) - const undeclared = newerRow({}) - storeNewerRow(undeclared) - const readOnly = await journals.open({ identity: IDENTITY, stateDirectory: directory }) - expect(readOnly.isReadOnly).toBe(true) + const reopened = await journals.open({ identity: IDENTITY, stateDirectory: directory }) + expect(reopened.isReadOnly).toBe(true) + expect(reopened.repair).toEqual({ malformedRows: 0 }) await expect( - readOnly.appendItem(item(1), { kind: 'status', text: 'after' }, scope) + reopened.appendItem(item(1), { kind: 'status', text: 'after' }, scope) ).rejects.toMatchObject({ code: 'journal_read_only' }) await journals.closeAll() - expect(newerKinds()).toEqual([undeclared]) - - for (const ifUnknown of ['skip', 'carry']) { - const declared = newerRow({ ifUnknown }) - storeNewerRow(declared) - const writable = await journals.open({ identity: IDENTITY, stateDirectory: directory }) - expect(writable.isReadOnly).toBe(false) - expect(writable.repair).toEqual({ malformedRows: 0 }) - await writable.appendItem(item(1), { kind: 'status', text: 'after' }, scope) - await journals.closeAll() - expect(newerKinds()).toEqual([declared]) - } + expect(storedRows(directory)).toEqual(rowsBefore) + expect(newerKinds()).toEqual([newer]) const checkout = await materializeReleaseCheckout(WRITABLE_BASELINE_REF) const support = await importReleaseCheckoutModule( From 7704a6b3a0d2e19f1b36268099aab99a099175ae Mon Sep 17 00:00:00 2001 From: Brennan Benson <79079362+brennanb2025@users.noreply.github.com> Date: Thu, 1 Oct 2026 18:18:43 -0700 Subject: [PATCH 5/5] refactor(native-chat): derive the journal's known row kinds from the row union Each kind's own-field check now lives in one table keyed by every kind JournalRow holds, and the set of kinds this build knows is derived from that table. A kind added to the union without a check fails to compile, rather than latching this build's own chats read-only as a newer build's kind. A test reads one valid row of every kind. --- .../journal-row-schema.ts | 100 +++++++----------- .../journal-unknown-row-kind.test.ts | 52 ++++++++- 2 files changed, 90 insertions(+), 62 deletions(-) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 3788ea1288b5..9276b1a8c05f 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -192,15 +192,6 @@ export type JournalRowParse = * compact this journal. */ | { ok: false; unreadable: true } -const ROW_KINDS = new Set([ - 'epoch', - 'item', - 'tombstone', - 'submission', - 'dispatch', - 'lifecycle-batch' -]) - export function serializeJournalRow(row: JournalRow): string { return JSON.stringify(row) } @@ -244,7 +235,7 @@ export function parseJournalRow(line: string): JournalRowParse { } // A newer build's kind is placed by the envelope every row keeps; one without it is damage. const { kind } = upcast - const unknownKind = typeof kind === 'string' && kind !== '' && !ROW_KINDS.has(kind) + const unknownKind = typeof kind === 'string' && kind !== '' && !KNOWN_ROW_KINDS.has(kind) return { ok: false, unreadable: unknownKind && hasJournalRowEnvelope(upcast) } } @@ -332,58 +323,45 @@ function isPlainObject(value: unknown): value is Record { * schema — their nested shapes are dereferenced unguarded all the way to the * rendered surface, so a JSON-valid corruption must fail here, not there. */ function isJournalRow(record: Record): record is JournalRow { - if (typeof record.kind !== 'string' || !ROW_KINDS.has(record.kind)) { - return false - } - if (!hasJournalRowEnvelope(record)) { - return false - } - if (record.kind === 'item') { - return ( - typeof record.itemId === 'string' && - Number.isInteger(record.revision) && - isAdmissibleAgentJournalItemBody(record.body) - ) - } - if (record.kind === 'tombstone') { - return typeof record.itemId === 'string' && Number.isInteger(record.revision) - } - if (record.kind === 'submission') { - return ( - typeof record.clientMessageId === 'string' && - record.clientMessageId.length > 0 && - typeof record.payloadFingerprint === 'string' && - isPlainObject(record.providerHandle) && - isAdmissibleAgentJournalMessageBody(record.body) - ) - } - if (record.kind === 'dispatch') { - return ( - typeof record.clientMessageId === 'string' && - record.clientMessageId.length > 0 && - typeof record.state === 'string' && - record.state.length > 0 && - (record.providerItemId === null || typeof record.providerItemId === 'string') && - (record.reason === null || typeof record.reason === 'string') - ) - } - if (record.kind === 'lifecycle-batch') { - return ( - typeof record.settlementId === 'string' && - record.settlementId.length > 0 && - Array.isArray(record.mutations) && - record.mutations.length > 0 && - record.mutations.length <= MAX_JOURNAL_LIFECYCLE_BATCH_MUTATIONS && - Buffer.byteLength(JSON.stringify(record), 'utf8') + 1 <= MAX_JOURNAL_LIFECYCLE_BATCH_BYTES && - record.mutations.every(isLifecycleMutation) - ) - } - return ( - record.kind === 'epoch' && - typeof record.reason === 'string' && - isPlainObject(record.providerHandle) - ) + const fieldCheck = typeof record.kind === 'string' ? KNOWN_ROW_KINDS.get(record.kind) : undefined + return fieldCheck !== undefined && hasJournalRowEnvelope(record) && fieldCheck(record) +} + +/** Each kind's own fields, keyed by every kind the union holds: a kind without a check here fails + * to compile, never reads as a newer build's kind. */ +const ROW_FIELD_CHECK_BY_KIND: Record< + JournalRow['kind'], + (record: Record) => boolean +> = { + epoch: (record) => typeof record.reason === 'string' && isPlainObject(record.providerHandle), + item: (record) => + typeof record.itemId === 'string' && + Number.isInteger(record.revision) && + isAdmissibleAgentJournalItemBody(record.body), + tombstone: (record) => typeof record.itemId === 'string' && Number.isInteger(record.revision), + submission: (record) => + typeof record.clientMessageId === 'string' && + record.clientMessageId.length > 0 && + typeof record.payloadFingerprint === 'string' && + isPlainObject(record.providerHandle) && + isAdmissibleAgentJournalMessageBody(record.body), + dispatch: (record) => + typeof record.clientMessageId === 'string' && + record.clientMessageId.length > 0 && + typeof record.state === 'string' && + record.state.length > 0 && + (record.providerItemId === null || typeof record.providerItemId === 'string') && + (record.reason === null || typeof record.reason === 'string'), + 'lifecycle-batch': (record) => + typeof record.settlementId === 'string' && + record.settlementId.length > 0 && + Array.isArray(record.mutations) && + record.mutations.length > 0 && + record.mutations.length <= MAX_JOURNAL_LIFECYCLE_BATCH_MUTATIONS && + Buffer.byteLength(JSON.stringify(record), 'utf8') + 1 <= MAX_JOURNAL_LIFECYCLE_BATCH_BYTES && + record.mutations.every(isLifecycleMutation) } +const KNOWN_ROW_KINDS = new Map(Object.entries(ROW_FIELD_CHECK_BY_KIND)) /** The fields every row keeps whatever its kind: its epoch, its place in it, its writer, its time. */ function hasJournalRowEnvelope(record: Record): boolean { diff --git a/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts index 2ea4e1c57de4..794ca52ab7d5 100644 --- a/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-unknown-row-kind.test.ts @@ -19,7 +19,7 @@ import { liveTestJournalRows, openTestJournalHostDatabase } from './journal-host-database-test-support' -import { parseJournalRow } from './journal-row-schema' +import { parseJournalRow, type JournalRow } from './journal-row-schema' const IDENTITY: AgentSessionJournalIdentity = { sessionId: 'session-newer-kind', @@ -101,6 +101,56 @@ async function journalWithOneItem(): Promise { return first.epoch } +/** One valid row of every kind this build writes; a kind added to the union without one here + * fails to compile. */ +function rowOfEveryKind(): { [Kind in JournalRow['kind']]: Extract } { + const base = { v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, epoch: 'epoch-1', fence: 1, ts: 1 } + const providerHandle = IDENTITY.providerHandle + return { + epoch: { ...base, seq: 1, kind: 'epoch', reason: 'session_created', providerHandle }, + item: { + ...base, + seq: 2, + kind: 'item', + itemId: 'i', + revision: 1, + body: { kind: 'status', text: 'x' } + }, + tombstone: { ...base, seq: 3, kind: 'tombstone', itemId: 'i', revision: 2 }, + submission: { + ...base, + seq: 4, + kind: 'submission', + clientMessageId: 'c', + payloadFingerprint: 'f', + providerHandle, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] } + }, + dispatch: { + ...base, + seq: 5, + kind: 'dispatch', + clientMessageId: 'c', + state: 'accepted', + providerItemId: null, + reason: null + }, + 'lifecycle-batch': { + ...base, + seq: 6, + kind: 'lifecycle-batch', + settlementId: 's', + mutations: [{ kind: 'tombstone', itemId: 'i', revision: 3 }] + } + } +} + +it("reads a row of every kind this build writes, never as a newer build's kind", () => { + for (const row of Object.values(rowOfEveryKind())) { + expect(parseJournalRow(JSON.stringify(row))).toEqual({ ok: true, row }) + } +}) + describe('parsing a row of a kind this build does not know', () => { it('reads one in the envelope every row keeps as unreadable', () => { expect(parseJournalRow(JSON.stringify(newerRow('epoch-1', 7)))).toEqual({