diff --git a/src/main/native-chat/agent-session-journal/journal-blob-store.ts b/src/main/native-chat/agent-session-journal/journal-blob-store.ts new file mode 100644 index 00000000000..a084c5641d8 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-blob-store.ts @@ -0,0 +1,74 @@ +// Content-addressed store for the remainder of a bounded payload. +// +// Blobs are named by their sha256, so writing the same output twice costs one +// file and re-import is idempotent. They live beside the journal (host-side +// per-workspace state, never inside the user's working tree) and share the +// epoch's retention: compaction prunes every blob no retained row references. + +import { mkdir, readFile, readdir, rm, stat } from 'node:fs/promises' +import { join } from 'node:path' +import { durableWriteTempPath, writeFileDurable } from '../../durable-file-write' + +const BLOB_DIR = 'blobs' + +function blobPath(journalDir: string, digest: string): string { + return join(journalDir, BLOB_DIR, digest) +} + +/** Persist `payload` under its digest. Returns the digest so the caller can + * stamp it on the row it is about to append. */ +export async function putJournalBlob( + journalDir: string, + digest: string, + payload: string +): Promise { + const target = blobPath(journalDir, digest) + // Content addressing makes a rewrite pointless: identical digest, identical bytes. + if (await pathExists(target)) { + return digest + } + await mkdir(join(journalDir, BLOB_DIR), { recursive: true }) + await writeFileDurable(durableWriteTempPath(target), target, payload) + return digest +} + +export async function readJournalBlob(journalDir: string, digest: string): Promise { + try { + return await readFile(blobPath(journalDir, digest), 'utf-8') + } catch { + return null + } +} + +/** Drop every blob outside `retained`. Called from compaction, under the + * current lease fence, after the snapshot is durable — so a crash mid-prune + * leaves extra blobs rather than dangling references. */ +export async function pruneJournalBlobs( + journalDir: string, + retained: ReadonlySet +): Promise { + let removed = 0 + let names: string[] + try { + names = await readdir(join(journalDir, BLOB_DIR)) + } catch { + return 0 + } + for (const name of names) { + if (retained.has(name)) { + continue + } + await rm(join(journalDir, BLOB_DIR, name), { force: true }).catch(() => {}) + removed += 1 + } + return removed +} + +async function pathExists(path: string): Promise { + try { + await stat(path) + return true + } catch { + return false + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-compaction.ts b/src/main/native-chat/agent-session-journal/journal-compaction.ts new file mode 100644 index 00000000000..dd8f4f38066 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-compaction.ts @@ -0,0 +1,103 @@ +// Retention and compaction. +// +// The snapshot carries the retained tail with it, so publishing both is ONE +// atomic write and there is no window where the folded state exists without the +// rows a reconnecting client still needs. Truncating the log afterwards is +// idempotent: a crash before it leaves the log a superset of the tail. +// +// The retained tail must cover the longest reconnect window Orca supports, or a +// client that was merely asleep gets a full snapshot reload instead of a resume. + +import { + referencedBlobDigests, + renderJournalState, + type JournalReducerState +} from './journal-reducer' +import { pruneJournalBlobs } from './journal-blob-store' +import { + rewriteJournalLog, + writeJournalSnapshotFile, + type JournalSnapshotFile +} from './journal-log-file' +import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' +import type { JournalRow } from './journal-row-schema' + +export type JournalCompactionPolicy = { + /** Always keep at least this many rows, however old they are. */ + minTailRows: number + /** Keep every row observed within this window. */ + retainTailMs: number +} + +/** Two hours of tail comfortably covers a phone that slept through a commute, + * which is the longest reconnect Orca resumes rather than reloads. */ +export const DEFAULT_JOURNAL_COMPACTION_POLICY: JournalCompactionPolicy = { + minTailRows: 512, + retainTailMs: 2 * 60 * 60 * 1000 +} + +export type JournalCompactionResult = { + tailRows: JournalRow[] + compactedThrough: number + oldestSequence: number +} + +export async function compactJournal(input: { + journalDir: string + state: JournalReducerState + tailRows: readonly JournalRow[] + policy?: JournalCompactionPolicy + now: number +}): Promise { + const policy = input.policy ?? DEFAULT_JOURNAL_COMPACTION_POLICY + const retained = retainTail(input.tailRows, policy, input.now) + const rendered = renderJournalState(input.state) + const compactedThrough = input.state.lastSequence + + const snapshot: JournalSnapshotFile = { + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + epoch: input.state.epoch, + compactedThrough, + items: rendered.items, + submissions: rendered.submissions, + receipts: [...input.state.receipts.values()].map((receipt) => ({ + clientMessageId: receipt.clientMessageId, + providerItemId: receipt.providerItemId, + epoch: receipt.cursor.epoch, + sequence: receipt.cursor.sequence, + acceptedAt: receipt.acceptedAt + })), + aliases: [...input.state.aliases.entries()].map(([providerItemId, itemId]) => ({ + providerItemId, + itemId + })), + tail: retained + } + + await writeJournalSnapshotFile(input.journalDir, snapshot) + await rewriteJournalLog(input.journalDir, retained) + // Blobs are pruned last: a crash before this leaks bytes, whereas pruning + // first would strand a snapshot pointing at a payload that no longer exists. + await pruneJournalBlobs(input.journalDir, referencedBlobDigests(input.state)) + + return { + tailRows: retained, + compactedThrough, + oldestSequence: retained[0]?.seq ?? compactedThrough + 1 + } +} + +function retainTail( + rows: readonly JournalRow[], + policy: JournalCompactionPolicy, + now: number +): JournalRow[] { + if (rows.length <= policy.minTailRows) { + return [...rows] + } + const floor = now - policy.retainTailMs + const byAge = rows.findIndex((row) => row.ts >= floor) + const byCount = rows.length - policy.minTailRows + const start = byAge < 0 ? byCount : Math.min(byAge, byCount) + return rows.slice(start) +} diff --git a/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts new file mode 100644 index 00000000000..0f8d85d34fc --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-crash-boundary.test.ts @@ -0,0 +1,382 @@ +// The crash boundary: the host wrote a submission row, dispatched, and died +// before it learned whether the provider took the message. Replay must reconcile +// without duplicating the user's message and without losing it. + +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 { + agentJournalItemKey, + agentJournalSubmissionKey +} from '../../../shared/agent-session-journal-item-key' +import type { + AgentJournalItemIdentity, + AgentJournalMessageItem, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { digestPayload } from './journal-payload-bounds' +import { + reconcileSubmissions, + type ProviderHistoryItem, + type ProviderHistoryWindow +} from './journal-submission-reconciler' +import { openAgentSessionJournal } from './journal-store' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +const TURN_ID = '019fd8ca-edbe-7c43-b231-4c7aea3a2d89' + +const ACCEPTED_IDENTITY: AgentJournalItemIdentity = { + provider: 'codex', + threadId: 'thread-1', + turnId: TURN_ID, + ordinal: 0 +} + +let root: string +let clock = 1_000 + +function tick(): number { + clock += 1 + return clock +} + +function userMessage(text: string): AgentJournalMessageItem { + return { kind: 'message', role: 'user', blocks: [{ type: 'text', text }] } +} + +async function open() { + return openAgentSessionJournal({ + identity: IDENTITY, + journalDir: root, + now: tick, + mintEpoch: () => `epoch-${clock}` + }) +} + +/** A Codex `userMessage` history item; `clientId` is the echoed client message id. */ +function history(input: { + itemId: string + clientId: string | null + text: string + ordinal: number +}): ProviderHistoryItem { + return { + providerItemId: input.itemId, + clientMessageId: input.clientId, + payloadFingerprint: digestPayload(input.text), + identity: { provider: 'codex', threadId: 'thread-1', turnId: TURN_ID, ordinal: input.ordinal } + } +} + +function window( + items: ProviderHistoryItem[], + overrides: Partial = {} +): ProviderHistoryWindow { + return { items, boundaryConsistent: true, turnInFlight: false, ...overrides } +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-journal-crash-')) + clock = 1_000 +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('crash between provider accept and journal commit', () => { + it('reconciles the echo into the existing bubble instead of duplicating it', async () => { + const journal = await open() + await journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: digestPayload('deploy the thing'), + body: userMessage('deploy the thing'), + fence: 1 + }) + // Host dies here: the provider accepted, but no dispatch row was written. + + const restarted = await open() + expect(restarted.pendingSubmissions().map((entry) => entry.clientMessageId)).toEqual(['cm_1']) + await restarted.markPendingSubmissionsUnknown(2) + expect(restarted.submissions()[0]?.dispatchState).toBe('unknown') + + const [outcome] = reconcileSubmissions({ + submissions: restarted.submissions(), + history: window([ + history({ itemId: 'item-1', clientId: 'cm_1', text: 'deploy the thing', ordinal: 0 }) + ]) + }) + expect(outcome).toMatchObject({ outcome: 'accepted', providerItemId: 'item-1' }) + + if (outcome?.outcome !== 'accepted') { + throw new Error('expected the echoed submission to reconcile as accepted') + } + await restarted.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'accepted', + providerIdentity: outcome.identity, + fence: 2, + recovered: true + }) + // The provider's own copy of the message arrives next, under the identity + // reconciliation adopted. It must land in the bubble the user already sees. + await restarted.appendItem(outcome.identity, userMessage('deploy the thing'), { fence: 2 }) + + const items = restarted.snapshot().items + expect(items).toHaveLength(1) + expect(items[0]?.itemId).toBe(agentJournalSubmissionKey('cm_1')) + expect(restarted.receiptFor('cm_1')?.providerItemId).toBe(agentJournalItemKey(outcome.identity)) + }) + + it('reports a rejected submission as never delivered, and never re-sends it', async () => { + const journal = await open() + await journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: digestPayload('never landed'), + body: userMessage('never landed'), + fence: 1 + }) + + const restarted = await open() + await restarted.markPendingSubmissionsUnknown(2) + const [outcome] = reconcileSubmissions({ + submissions: restarted.submissions(), + history: window([]) + }) + expect(outcome).toEqual({ + clientMessageId: 'cm_1', + outcome: 'rejected', + reason: 'not_delivered' + }) + + await restarted.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'rejected', + reason: 'not_delivered', + fence: 2, + recovered: true + }) + // The bubble survives with an explicit terminal state — the message is not + // silently retried and not silently dropped. + expect(restarted.snapshot().items).toHaveLength(1) + expect(restarted.snapshot().submissions[0]?.dispatchState).toBe('rejected') + expect(restarted.receiptFor('cm_1')).toBeNull() + }) + + it('survives replay of an already-reconciled journal without changing the answer', async () => { + const journal = await open() + await journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: digestPayload('once'), + body: userMessage('once'), + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'accepted', + providerIdentity: ACCEPTED_IDENTITY, + fence: 1 + }) + const settled = journal.snapshot() + + const reopened = await open() + expect(reopened.snapshot()).toEqual(settled) + expect(reopened.pendingSubmissions()).toHaveLength(0) + expect( + reconcileSubmissions({ submissions: reopened.submissions(), history: window([]) }) + ).toEqual([]) + }) + + it('keeps the receipt after the row that minted it was compacted away', async () => { + const journal = await openAgentSessionJournal({ + identity: IDENTITY, + journalDir: root, + now: tick, + mintEpoch: () => `epoch-${clock}`, + compaction: { minTailRows: 1, retainTailMs: 0 } + }) + await journal.appendSubmission({ + clientMessageId: 'cm_1', + payloadFingerprint: digestPayload('kept'), + body: userMessage('kept'), + fence: 1 + }) + await journal.resolveDispatch({ + clientMessageId: 'cm_1', + state: 'accepted', + providerIdentity: ACCEPTED_IDENTITY, + fence: 1 + }) + await journal.compact() + + const reopened = await open() + expect(reopened.receiptFor('cm_1')?.providerItemId).toBe(agentJournalItemKey(ACCEPTED_IDENTITY)) + }) +}) + +describe('reconciliation matching', () => { + const submissions = [ + { + clientMessageId: 'cm_1', + fence: 1, + payloadFingerprint: digestPayload('same text'), + dispatchState: 'unknown' as const, + providerItemId: null, + reason: null, + submittedAt: 1, + resolvedAt: null + }, + { + clientMessageId: 'cm_2', + fence: 1, + payloadFingerprint: digestPayload('same text'), + dispatchState: 'unknown' as const, + providerItemId: null, + reason: null, + submittedAt: 2, + resolvedAt: null + } + ] + + it('matches each submission to its own echo when the provider carries client ids', () => { + const outcomes = reconcileSubmissions({ + submissions, + history: window([ + history({ itemId: 'item-1', clientId: 'cm_1', text: 'same text', ordinal: 0 }), + history({ itemId: 'item-3', clientId: 'cm_2', text: 'same text', ordinal: 2 }) + ]) + }) + expect(outcomes).toEqual([ + expect.objectContaining({ clientMessageId: 'cm_1', providerItemId: 'item-1' }), + expect.objectContaining({ clientMessageId: 'cm_2', providerItemId: 'item-3' }) + ]) + }) + + it('refuses to guess between two identical payloads with no id to tell them apart', () => { + const outcomes = reconcileSubmissions({ + submissions, + history: window([ + history({ itemId: 'item-1', clientId: null, text: 'same text', ordinal: 0 }), + history({ itemId: 'item-3', clientId: null, text: 'same text', ordinal: 2 }) + ]) + }) + expect(outcomes.map((outcome) => outcome.outcome)).toEqual(['unknown', 'unknown']) + expect(outcomes[0]).toMatchObject({ reason: 'ambiguous_match' }) + }) + + it('uses a unique fingerprint only as a tiebreak when no id is echoed', () => { + const [outcome] = reconcileSubmissions({ + submissions: [submissions[0]!], + history: window([ + history({ itemId: 'item-1', clientId: null, text: 'same text', ordinal: 0 }), + history({ itemId: 'item-2', clientId: null, text: 'something else', ordinal: 1 }) + ]) + }) + expect(outcome).toMatchObject({ outcome: 'accepted', providerItemId: 'item-1' }) + }) + + it('never matches on text alone when the fingerprint disagrees', () => { + const [outcome] = reconcileSubmissions({ + submissions: [submissions[0]!], + history: window([ + { + providerItemId: 'item-1', + clientMessageId: null, + payloadFingerprint: digestPayload('different payload, same rendered text'), + identity: { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 0 } + } + ]) + }) + expect(outcome).toMatchObject({ outcome: 'rejected', reason: 'not_delivered' }) + }) + + it('lets a strong client-id match win an item a weaker fingerprint would have claimed', () => { + const outcomes = reconcileSubmissions({ + submissions, + history: window([ + history({ itemId: 'item-1', clientId: 'cm_2', text: 'same text', ordinal: 0 }) + ]) + }) + expect(outcomes).toEqual([ + expect.objectContaining({ clientMessageId: 'cm_1', outcome: 'rejected' }), + expect.objectContaining({ clientMessageId: 'cm_2', providerItemId: 'item-1' }) + ]) + }) + + it('re-matches a submission on the journal key it already adopted, not the raw provider id', () => { + const [outcome] = reconcileSubmissions({ + submissions: [{ ...submissions[0]!, providerItemId: agentJournalItemKey(ACCEPTED_IDENTITY) }], + history: window([ + // The provider renumbered its raw id; the identity-derived key did not move. + history({ itemId: 'item-7', clientId: null, text: 'unrelated', ordinal: 0 }) + ]) + }) + expect(outcome).toMatchObject({ outcome: 'accepted', providerItemId: 'item-7' }) + }) + + it('never hands one provider item to two submissions', () => { + const outcomes = reconcileSubmissions({ + submissions: [ + { ...submissions[0]!, providerItemId: agentJournalItemKey(ACCEPTED_IDENTITY) }, + submissions[1]! + ], + // One item, wanted by both passes: cm_1 adopted its key, cm_2 is echoed on + // it. Adopting it twice would render the same provider message twice. + history: window([ + history({ itemId: 'item-7', clientId: 'cm_2', text: 'same text', ordinal: 0 }) + ]) + }) + expect(outcomes).toEqual([ + expect.objectContaining({ clientMessageId: 'cm_1', providerItemId: 'item-7' }), + expect.objectContaining({ clientMessageId: 'cm_2', outcome: 'rejected' }) + ]) + }) + + it('stays unknown when the history boundary cannot be trusted', () => { + const [outcome] = reconcileSubmissions({ + submissions: [submissions[0]!], + history: window([], { boundaryConsistent: false }) + }) + expect(outcome).toEqual({ + clientMessageId: 'cm_1', + outcome: 'unknown', + reason: 'history_boundary_inconsistent' + }) + }) + + it('stays unknown while a turn is still running', () => { + const [outcome] = reconcileSubmissions({ + submissions: [submissions[0]!], + history: window([], { turnInFlight: true }) + }) + expect(outcome).toEqual({ + clientMessageId: 'cm_1', + outcome: 'unknown', + reason: 'turn_in_flight' + }) + }) + + it('leaves settled submissions alone', () => { + expect( + reconcileSubmissions({ + submissions: [ + { + ...submissions[0]!, + dispatchState: 'accepted', + providerItemId: agentJournalItemKey(ACCEPTED_IDENTITY) + }, + { ...submissions[1]!, dispatchState: 'rejected' } + ], + history: window([]) + }) + ).toEqual([]) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-cursor.test.ts b/src/main/native-chat/agent-session-journal/journal-cursor.test.ts new file mode 100644 index 00000000000..704973e7ebd --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-cursor.test.ts @@ -0,0 +1,106 @@ +import { describe, expect, it } from 'vitest' +import { + findSequenceGap, + resolveJournalResume, + sameJournalCursor, + type JournalCursorRange +} from './journal-cursor' + +const RANGE: JournalCursorRange = { epoch: 'e1', lastSequence: 40, oldestSequence: 11 } + +describe('resolveJournalResume', () => { + it('resumes from a cursor inside the retained tail', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e1', sequence: 25 })).toEqual({ + ok: true, + afterSequence: 25 + }) + }) + + it('resumes from a cursor sitting exactly on the compaction boundary', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e1', sequence: 10 })).toEqual({ + ok: true, + afterSequence: 10 + }) + }) + + it('resumes from a cursor at the tip with nothing to send', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e1', sequence: 40 })).toEqual({ + ok: true, + afterSequence: 40 + }) + }) + + it('forces a reload when the epoch rolled', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e0', sequence: 25 })).toEqual({ + ok: false, + reset: 'epoch_changed' + }) + }) + + it('forces a reload when the epoch rolled even at a sequence this epoch also holds', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e0', sequence: 40 }).ok).toBe(false) + }) + + it('forces a reload when the client is ahead of the journal', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e1', sequence: 41 })).toEqual({ + ok: false, + reset: 'cursor_ahead' + }) + }) + + it('forces a reload when the cursor fell below the compaction floor', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e1', sequence: 9 })).toEqual({ + ok: false, + reset: 'cursor_compacted' + }) + }) + + it('resumes a fresh client from sequence 0 on an uncompacted journal', () => { + const fresh: JournalCursorRange = { epoch: 'e1', lastSequence: 3, oldestSequence: 1 } + expect(resolveJournalResume(fresh, { epoch: 'e1', sequence: 0 })).toEqual({ + ok: true, + afterSequence: 0 + }) + }) + + it('rejects sequence 0 once the journal has compacted past it', () => { + expect(resolveJournalResume(RANGE, { epoch: 'e1', sequence: 0 })).toEqual({ + ok: false, + reset: 'cursor_compacted' + }) + }) +}) + +describe('sameJournalCursor', () => { + it('requires both the epoch and the sequence to match', () => { + expect(sameJournalCursor({ epoch: 'e1', sequence: 3 }, { epoch: 'e1', sequence: 3 })).toBe(true) + expect(sameJournalCursor({ epoch: 'e1', sequence: 3 }, { epoch: 'e2', sequence: 3 })).toBe( + false + ) + expect(sameJournalCursor({ epoch: 'e1', sequence: 3 }, { epoch: 'e1', sequence: 4 })).toBe( + false + ) + }) +}) + +describe('findSequenceGap', () => { + it('accepts a contiguous run', () => { + expect(findSequenceGap([7, 8, 9], 7)).toBeNull() + }) + + it('accepts an empty run', () => { + expect(findSequenceGap([], 7)).toBeNull() + }) + + it('reports the first missing sequence', () => { + expect(findSequenceGap([7, 8, 10], 7)).toEqual({ gapAt: 9 }) + }) + + it('reports a run that starts above the expected first sequence', () => { + expect(findSequenceGap([8, 9], 7)).toEqual({ gapAt: 7 }) + }) + + it('reports a reused sequence', () => { + expect(findSequenceGap([7, 7, 8], 7)).toEqual({ gapAt: 8 }) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-cursor.ts b/src/main/native-chat/agent-session-journal/journal-cursor.ts new file mode 100644 index 00000000000..4b39fd2e761 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-cursor.ts @@ -0,0 +1,71 @@ +// Epoch-qualified cursor resume. +// +// A cursor is only meaningful inside its epoch: rollover invalidates every one +// of them, and the client takes a clean snapshot reload rather than being +// handed rows that belong to a rebuilt timeline. + +import type { + AgentJournalCursor, + AgentJournalResetReason +} from '../../../shared/agent-session-journal-types' + +export type JournalCursorRange = { + epoch: string + /** Sequence of the newest appended row; 0 when the epoch is empty. */ + lastSequence: number + /** Lowest sequence still individually replayable after compaction. */ + oldestSequence: number +} + +export type JournalResume = + /** Replay rows with `seq > afterSequence`. */ + { ok: true; afterSequence: number } | { ok: false; reset: AgentJournalResetReason } + +/** Total order over cursors within one epoch; cross-epoch comparison is + * meaningless, so callers must check the epoch first. */ +export function sameJournalCursor(a: AgentJournalCursor, b: AgentJournalCursor): boolean { + return a.epoch === b.epoch && a.sequence === b.sequence +} + +/** + * Decide whether a reconnecting client can resume from `cursor`. + * + * An epoch mismatch, a cursor ahead of the journal (the client saw rows this + * host no longer has — a rolled-back or rebuilt prefix), and a cursor below the + * compaction floor all resolve to a snapshot reload. None of them is recoverable + * by shipping a partial batch. + */ +export function resolveJournalResume( + range: JournalCursorRange, + cursor: AgentJournalCursor +): JournalResume { + if (cursor.epoch !== range.epoch) { + return { ok: false, reset: 'epoch_changed' } + } + if (cursor.sequence > range.lastSequence) { + return { ok: false, reset: 'cursor_ahead' } + } + // `oldestSequence - 1` is the compaction boundary: a client sitting exactly on + // it has seen everything folded into the snapshot and can take the tail. + if (cursor.sequence < range.oldestSequence - 1) { + return { ok: false, reset: 'cursor_compacted' } + } + return { ok: true, afterSequence: cursor.sequence } +} + +/** Contiguity check over a replayed row sequence. A gap means the journal lost + * a row; the reader must treat it as corrupt and force epoch rollover rather + * than render a partial timeline. */ +export function findSequenceGap( + sequences: readonly number[], + expectedFirst: number +): { gapAt: number } | null { + let expected = expectedFirst + for (const sequence of sequences) { + if (sequence !== expected) { + return { gapAt: expected } + } + expected += 1 + } + return null +} diff --git a/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts b/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts new file mode 100644 index 00000000000..b6b9a5ea9be --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-epoch-rollover.ts @@ -0,0 +1,48 @@ +// Opening a new epoch. +// +// The snapshot is what names the live epoch, so it is published BEFORE the log +// is reset. A crash mid-rollover therefore leaves stale-epoch rows behind the +// new snapshot, which `loadJournal` drops — the reverse order would leave a +// journal whose log no longer matches any epoch anyone can name. + +import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' +import type { AgentSessionProviderHandle } from '../../../shared/agent-session-journal-types' +import { compactJournal } from './journal-compaction' +import { + applyJournalRow, + createJournalReducerState, + type JournalReducerState +} from './journal-reducer' +import type { AgentJournalEpochReason, JournalRow } from './journal-row-schema' + +export async function publishNewEpoch(input: { + journalDir: string + sessionId: string + providerHandle: AgentSessionProviderHandle + epoch: string + reason: AgentJournalEpochReason + fence: number + now: number +}): Promise<{ state: JournalReducerState; row: JournalRow }> { + const row: JournalRow = { + kind: 'epoch', + reason: input.reason, + providerHandle: input.providerHandle, + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + epoch: input.epoch, + seq: 1, + fence: input.fence, + ts: input.now + } + const state = createJournalReducerState(input.sessionId, input.epoch) + await compactJournal({ + journalDir: input.journalDir, + state, + tailRows: [row], + policy: { minTailRows: 1, retainTailMs: Number.POSITIVE_INFINITY }, + now: input.now + }) + applyJournalRow(state, row) + state.oldestSequence = 1 + return { state, row } +} diff --git a/src/main/native-chat/agent-session-journal/journal-item-identity.test.ts b/src/main/native-chat/agent-session-journal/journal-item-identity.test.ts new file mode 100644 index 00000000000..b4c74e77456 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-item-identity.test.ts @@ -0,0 +1,145 @@ +import { describe, expect, it } from 'vitest' +import { + agentJournalItemKey, + agentJournalSubmissionKey +} from '../../../shared/agent-session-journal-item-key' +import type { AgentJournalItemIdentity } from '../../../shared/agent-session-journal-types' + +// Fixtures mirror the shapes the providers actually emit: a resumed Codex +// thread renumbers its items positionally, and a forked Claude session copies +// history with the ORIGINAL item uuids. + +const THREAD = '019fd8ca-edbe-7c43-b231-4c7aea3a2d89' +const TURN_A = '019fd8ca-edbe-7c43-b231-4c7aea3a2d89' +const TURN_B = '019fd8cb-1c40-7a02-9f31-0f1a54b7c211' + +describe('codex identity survives positional renumbering', () => { + it('keys the same logical item identically before and after a resume', () => { + // First run: the app server labels the second turn's user message item-3. + const live: AgentJournalItemIdentity = { + provider: 'codex', + threadId: THREAD, + turnId: TURN_B, + ordinal: 0 + } + // After `thread/resume` the same item comes back as item-1 of the replayed + // history. Ordinal-within-turn is unchanged, so the key is unchanged. + const resumed: AgentJournalItemIdentity = { + provider: 'codex', + threadId: THREAD, + turnId: TURN_B, + ordinal: 0 + } + expect(agentJournalItemKey(resumed)).toBe(agentJournalItemKey(live)) + }) + + it('separates two items inside one turn', () => { + const first = agentJournalItemKey({ + provider: 'codex', + threadId: THREAD, + turnId: TURN_A, + ordinal: 0 + }) + const second = agentJournalItemKey({ + provider: 'codex', + threadId: THREAD, + turnId: TURN_A, + ordinal: 1 + }) + expect(first).not.toBe(second) + }) + + it('disambiguates a fork that copies turns keeping their original turn ids', () => { + const original = agentJournalItemKey({ + provider: 'codex', + threadId: THREAD, + turnId: TURN_A, + ordinal: 0 + }) + const forked = agentJournalItemKey({ + provider: 'codex', + threadId: '019fd900-77aa-7c19-8bd0-2b3c4d5e6f70', + turnId: TURN_A, + ordinal: 0 + }) + expect(forked).not.toBe(original) + }) +}) + +describe('claude identity', () => { + it('keys on (session id, uuid)', () => { + const key = agentJournalItemKey({ + provider: 'claude', + sessionId: '29eb22a4-6a5f-4f21-9b0c-1d7f3a2e5c88', + uuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04' + }) + expect(key).toBe( + 'claude:29eb22a4-6a5f-4f21-9b0c-1d7f3a2e5c88:c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04' + ) + }) + + it('reconciles a forked transcript onto the parent item rather than duplicating it', () => { + // `--fork-session` mints a new session id but copies records verbatim, so a + // copied record still names the session it was written in. + const parent = agentJournalItemKey({ + provider: 'claude', + sessionId: '29eb22a4-6a5f-4f21-9b0c-1d7f3a2e5c88', + uuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04' + }) + const copiedIntoFork = agentJournalItemKey({ + provider: 'claude', + sessionId: '29eb22a4-6a5f-4f21-9b0c-1d7f3a2e5c88', + uuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04' + }) + expect(copiedIntoFork).toBe(parent) + }) + + it('keeps a genuinely new item in the fork distinct', () => { + const parent = agentJournalItemKey({ + provider: 'claude', + sessionId: '29eb22a4-6a5f-4f21-9b0c-1d7f3a2e5c88', + uuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04' + }) + const minted = agentJournalItemKey({ + provider: 'claude', + sessionId: '7b1e5d33-0f28-42ac-8d59-9a4c6e2b1f70', + uuid: 'f8b2c9a1-3e77-4c60-b1a2-5d0e7f4a9c33' + }) + expect(minted).not.toBe(parent) + }) +}) + +describe('key encoding', () => { + it('cannot be collided by a separator inside an id', () => { + const a = agentJournalItemKey({ + provider: 'legacy', + agent: 'codex', + sessionId: 'a:b', + recordId: 'c' + }) + const b = agentJournalItemKey({ + provider: 'legacy', + agent: 'codex', + sessionId: 'a', + recordId: 'b:c' + }) + expect(a).not.toBe(b) + }) + + it('separates the provider namespaces', () => { + const orca = agentJournalItemKey({ provider: 'orca', clientMessageId: 'x' }) + const legacy = agentJournalItemKey({ + provider: 'legacy', + agent: 'claude', + sessionId: 'x', + recordId: 'x' + }) + expect(orca).not.toBe(legacy) + }) + + it('derives the submission slot from the same function the reducer uses', () => { + expect(agentJournalSubmissionKey('cm_42')).toBe( + agentJournalItemKey({ provider: 'orca', clientMessageId: 'cm_42' }) + ) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-identity.ts b/src/main/native-chat/agent-session-journal/journal-legacy-identity.ts new file mode 100644 index 00000000000..c7903e604d3 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-legacy-identity.ts @@ -0,0 +1,87 @@ +// Identity anchors for bridge-era transcript lines. +// +// The transcript decoders return a render model, not an identity, so the import +// reads identity from the SAME raw line the decoder consumed rather than +// inferring it from decoded text. +// +// Claude gets its real identity namespace: the project jsonl IS the provider's +// store, and `uuid` survives `--fork-session` unchanged, so a later structured +// session reconciles against these keys directly. +// +// Codex, Grok, and omp get the `legacy` namespace. A Codex rollout file records +// `response_item` ids (`msg_…`, `rs_…`, `ctc_…`) which are a different namespace +// from the app-server's positional `item-N` ordinals, and rollout records carry +// no turn id at all — so a rollout line cannot be expressed as a stable +// `(threadId, turnId, ordinal)` key without guessing. Legacy items are therefore +// import-scoped, and a later structured resume rolls the epoch and rebuilds. + +import type { AgentType } from '../../../shared/agent-status-types' +import type { AgentJournalItemIdentity } from '../../../shared/agent-session-journal-types' +import type { NativeChatTranscriptAgent } from '../../../shared/native-chat-agent-support' + +export type LegacyIdentityTracker = { + /** Identity for whatever the decoder emits from this raw line. Called for + * every line in file order, including ones the decoder discards. */ + identify(line: string, lineIndex: number): AgentJournalItemIdentity +} + +export function createLegacyIdentityTracker(input: { + transcriptAgent: NativeChatTranscriptAgent + agent: AgentType + sessionId: string +}): LegacyIdentityTracker { + if (input.transcriptAgent === 'claude') { + return { identify: (line, index) => claudeIdentity(line, index, input.agent, input.sessionId) } + } + return { + identify: (line, index) => ({ + provider: 'legacy', + agent: input.agent, + sessionId: input.sessionId, + recordId: legacyRecordId(line, index) + }) + } +} + +function claudeIdentity( + line: string, + lineIndex: number, + agent: AgentType, + sessionId: string +): AgentJournalItemIdentity { + const record = parseRecord(line) + const uuid = stringField(record, 'uuid') + if (!uuid) { + return { provider: 'legacy', agent, sessionId, recordId: `#${lineIndex}` } + } + // The record's own session id wins: a forked transcript keeps the original + // item uuids, and pairing them with the fork's id would mint new identities. + return { provider: 'claude', sessionId: stringField(record, 'sessionId') ?? sessionId, uuid } +} + +/** `payload.id` when the record carries one, else the line's position. Position + * is deterministic for a given import of a given file, which is all the legacy + * namespace promises. */ +function legacyRecordId(line: string, lineIndex: number): string { + const record = parseRecord(line) + const payload = record?.payload + const id = stringField(payload, 'id') ?? stringField(record, 'id') ?? stringField(record, 'uuid') + return id ?? `#${lineIndex}` +} + +function parseRecord(line: string): Record | null { + try { + const parsed: unknown = JSON.parse(line) + return parsed && typeof parsed === 'object' ? (parsed as Record) : null + } catch { + return null + } +} + +function stringField(source: unknown, key: string): string | null { + if (!source || typeof source !== 'object') { + return null + } + const value = (source as Record)[key] + return typeof value === 'string' && value ? value : null +} diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts new file mode 100644 index 00000000000..87a64a08e7a --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.test.ts @@ -0,0 +1,405 @@ +// Legacy import runs the existing per-agent transcript decoders and keys the +// results by identity read off the same raw lines. Fixtures are shaped like the +// files the providers actually write. + +import { mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' +import { readJournalBlob } from './journal-blob-store' +import { createLegacyIdentityTracker } from './journal-legacy-identity' +import { importLegacyTranscriptIntoJournal } from './journal-legacy-import' +import { DEFAULT_JOURNAL_PAYLOAD_LIMITS } from './journal-payload-bounds' +import { openAgentSessionJournal, type AgentSessionJournal } from './journal-store' + +const CLAUDE_SESSION = '29eb22a4-6a5f-4f21-9b0c-1d7f3a2e5c88' +const CODEX_SESSION = '019fd532-7c11-7a90-b6de-4e1a2c3d5f60' + +let root: string +let clock = 1_000 + +function tick(): number { + clock += 1 + return clock +} + +function identity(agent: 'claude' | 'codex', sessionId: string): AgentSessionJournalIdentity { + return { + sessionId, + workspaceId: 'ws-1', + hostId: 'host-1', + agent, + providerHandle: + agent === 'claude' + ? { kind: 'claude', sessionId, leafUuid: null } + : { kind: 'codex', threadId: sessionId } + } +} + +async function open(agent: 'claude' | 'codex', sessionId: string): Promise { + return openAgentSessionJournal({ + identity: identity(agent, sessionId), + journalDir: root, + now: tick, + mintEpoch: () => `epoch-${clock}` + }) +} + +function legacyKey(recordId: string): string { + return agentJournalItemKey({ + provider: 'legacy', + agent: 'codex', + sessionId: CODEX_SESSION, + recordId + }) +} + +async function writeFixture(name: string, lines: unknown[]): Promise { + const path = join(root, name) + await writeFile(path, `${lines.map((line) => JSON.stringify(line)).join('\n')}\n`, 'utf-8') + return path +} + +const CLAUDE_LINES = [ + { type: 'file-history-snapshot', messageId: 'boot', snapshot: {} }, + { + parentUuid: null, + isSidechain: false, + type: 'user', + message: { role: 'user', content: [{ type: 'text', text: 'add a retry' }] }, + uuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04', + timestamp: '2026-08-05T10:00:00.000Z', + cwd: '/Users/dev/project', + sessionId: CLAUDE_SESSION, + version: '2.1.220', + gitBranch: 'main' + }, + { + parentUuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04', + isSidechain: false, + type: 'assistant', + requestId: 'req_01', + message: { + role: 'assistant', + content: [{ type: 'text', text: 'On it.' }], + id: 'msg_ignored_in_favour_of_uuid' + }, + uuid: 'b7c9e1f2-8a30-4d55-91ab-6f0e2c4d8b11', + timestamp: '2026-08-05T10:00:04.000Z', + sessionId: CLAUDE_SESSION, + version: '2.1.220' + }, + { + parentUuid: 'b7c9e1f2-8a30-4d55-91ab-6f0e2c4d8b11', + isSidechain: false, + type: 'assistant', + message: { + role: 'assistant', + content: [{ type: 'tool_use', id: 'toolu_01', name: 'Edit', input: { file_path: 'a.ts' } }] + }, + uuid: 'd2f4a6b8-1c02-4e77-83bd-5a9c7e1f3d20', + timestamp: '2026-08-05T10:00:07.000Z', + sessionId: CLAUDE_SESSION + }, + { + parentUuid: 'd2f4a6b8-1c02-4e77-83bd-5a9c7e1f3d20', + isSidechain: false, + isMeta: true, + type: 'user', + message: { + role: 'user', + content: [{ type: 'tool_result', tool_use_id: 'toolu_01', content: 'edited 1 file' }] + }, + uuid: 'e3a5b7c9-2d13-4f88-94ce-6b0d8f2a4e31', + timestamp: '2026-08-05T10:00:08.000Z', + sessionId: CLAUDE_SESSION + }, + { type: 'last-prompt', leafUuid: 'e3a5b7c9-2d13-4f88-94ce-6b0d8f2a4e31' } +] + +// Shapes taken from a real rollout file: `event_msg` records carry no id, and +// `response_item` records do — which is exactly the split the tracker handles. +const CODEX_LINES = [ + { + type: 'session_meta', + timestamp: '2026-08-05T10:00:00.000Z', + payload: { + id: CODEX_SESSION, + session_id: CODEX_SESSION, + cwd: '/Users/dev/project', + originator: 'codex_cli_rs', + cli_version: '0.146.1' + } + }, + { + type: 'event_msg', + timestamp: '2026-08-05T10:00:01.000Z', + payload: { type: 'task_started', turn_id: '019fd8ca-edbe-7c43-b231-4c7aea3a2d89' } + }, + { + type: 'event_msg', + timestamp: '2026-08-05T10:00:02.000Z', + payload: { type: 'user_message', message: 'add a retry', kind: 'plain' } + }, + { + type: 'response_item', + timestamp: '2026-08-05T10:00:04.000Z', + payload: { + type: 'reasoning', + id: 'rs_06235749b04250a3016a7404b3a25c8199b882f2f8288fefd0', + summary: [{ type: 'summary_text', text: 'Checking the retry policy.' }] + } + }, + { + type: 'event_msg', + timestamp: '2026-08-05T10:00:05.000Z', + payload: { type: 'agent_message', message: 'On it.' } + }, + { + type: 'event_msg', + timestamp: '2026-08-05T10:00:06.000Z', + payload: { type: 'token_count', info: { total_token_usage: { input_tokens: 12 } } } + } +] + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-journal-import-')) + clock = 1_000 +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('claude import', () => { + it('keys items by (session id, uuid) from the raw record', async () => { + const filePath = await writeFixture('claude.jsonl', CLAUDE_LINES) + const journal = await open('claude', CLAUDE_SESSION) + const result = await importLegacyTranscriptIntoJournal({ + journal, + agent: 'claude', + sessionId: CLAUDE_SESSION, + fence: 1, + options: { filePath } + }) + + expect(result.ok).toBe(true) + expect(journal.snapshot().items.map((entry) => entry.itemId)).toEqual([ + agentJournalItemKey({ + provider: 'claude', + sessionId: CLAUDE_SESSION, + uuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04' + }), + agentJournalItemKey({ + provider: 'claude', + sessionId: CLAUDE_SESSION, + uuid: 'b7c9e1f2-8a30-4d55-91ab-6f0e2c4d8b11' + }), + agentJournalItemKey({ + provider: 'claude', + sessionId: CLAUDE_SESSION, + uuid: 'd2f4a6b8-1c02-4e77-83bd-5a9c7e1f3d20' + }), + agentJournalItemKey({ + provider: 'claude', + sessionId: CLAUDE_SESSION, + uuid: 'e3a5b7c9-2d13-4f88-94ce-6b0d8f2a4e31' + }) + ]) + }) + + it('stays aligned when the decoder drops lines the tracker still walks', async () => { + const filePath = await writeFixture('claude.jsonl', CLAUDE_LINES) + const journal = await open('claude', CLAUDE_SESSION) + await importLegacyTranscriptIntoJournal({ + journal, + agent: 'claude', + sessionId: CLAUDE_SESSION, + fence: 1, + options: { filePath } + }) + const items = journal.snapshot().items + // The first record is a file-history snapshot the decoder discards; if the + // anchors were misaligned, the first bubble would carry its identity. + expect(items[0]?.body).toEqual({ + kind: 'message', + role: 'user', + blocks: [{ type: 'text', text: 'add a retry' }] + }) + expect(items[2]?.body).toMatchObject({ kind: 'tool-call', name: 'Edit' }) + }) + + it('is idempotent: a second import reproduces the same timeline in a new epoch', async () => { + const filePath = await writeFixture('claude.jsonl', CLAUDE_LINES) + const journal = await open('claude', CLAUDE_SESSION) + const options = { filePath } + await importLegacyTranscriptIntoJournal({ + journal, + agent: 'claude', + sessionId: CLAUDE_SESSION, + fence: 1, + options + }) + const first = journal.snapshot() + const firstEpoch = journal.epoch + + await importLegacyTranscriptIntoJournal({ + journal, + agent: 'claude', + sessionId: CLAUDE_SESSION, + fence: 1, + options + }) + const second = journal.snapshot() + + expect(journal.epoch).not.toBe(firstEpoch) + expect(second.items.map((entry) => entry.itemId)).toEqual( + first.items.map((entry) => entry.itemId) + ) + expect(second.items.map((entry) => entry.body)).toEqual(first.items.map((entry) => entry.body)) + }) + + it('reconciles a forked transcript onto the parent uuids rather than duplicating them', () => { + const tracker = createLegacyIdentityTracker({ + transcriptAgent: 'claude', + agent: 'claude', + // The fork's own session id, which is NOT what the copied records carry. + sessionId: '7b1e5d33-0f28-42ac-8d59-9a4c6e2b1f70' + }) + const copied = JSON.stringify(CLAUDE_LINES[1]) + expect(tracker.identify(copied, 0)).toEqual({ + provider: 'claude', + sessionId: CLAUDE_SESSION, + uuid: 'c1a5f0de-2b44-4a11-9f0e-7c2d31b6aa04' + }) + }) +}) + +describe('codex import', () => { + it('keys rollout records in the import-scoped namespace, not as app-server ordinals', async () => { + const filePath = await writeFixture('rollout.jsonl', CODEX_LINES) + const journal = await open('codex', CODEX_SESSION) + await importLegacyTranscriptIntoJournal({ + journal, + agent: 'codex', + sessionId: CODEX_SESSION, + fence: 1, + options: { filePath } + }) + + // A rollout record with its own id keeps it; an `event_msg` has none, so it + // falls back to its line position — deterministic for a given file. + expect(journal.snapshot().items.map((entry) => entry.itemId)).toEqual([ + legacyKey('#2'), + legacyKey('rs_06235749b04250a3016a7404b3a25c8199b882f2f8288fefd0'), + legacyKey('#4') + ]) + expect(journal.snapshot().items.map((entry) => entry.body)).toEqual([ + { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'add a retry' }] }, + { + kind: 'message', + role: 'reasoning', + blocks: [{ type: 'text', text: 'Checking the retry policy.' }] + }, + { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: 'On it.' }] } + ]) + }) + + it('survives a resumed rollout that renumbers positional item ids', () => { + const tracker = createLegacyIdentityTracker({ + transcriptAgent: 'codex', + agent: 'codex', + sessionId: CODEX_SESSION + }) + const original = tracker.identify(JSON.stringify(CODEX_LINES[3]), 4) + // Same record replayed at a different position in a resumed file. + const replayed = tracker.identify(JSON.stringify(CODEX_LINES[3]), 11) + expect(replayed).toEqual(original) + expect(original).toMatchObject({ + recordId: 'rs_06235749b04250a3016a7404b3a25c8199b882f2f8288fefd0' + }) + }) + + it('falls back to line position only when a record carries no id', () => { + const tracker = createLegacyIdentityTracker({ + transcriptAgent: 'codex', + agent: 'codex', + sessionId: CODEX_SESSION + }) + expect(tracker.identify(JSON.stringify({ type: 'event_msg', payload: {} }), 3)).toEqual({ + provider: 'legacy', + agent: 'codex', + sessionId: CODEX_SESSION, + recordId: '#3' + }) + }) +}) + +describe('payload bounds on import', () => { + it('marks a clipped tool result and parks the remainder in the blob store', async () => { + const output = 'y'.repeat(64 * 1024) + const filePath = await writeFixture('claude-big.jsonl', [ + { + parentUuid: null, + isSidechain: false, + type: 'user', + message: { + role: 'user', + content: [{ type: 'tool_result', tool_use_id: 'toolu_9', content: output }] + }, + uuid: 'aa11bb22-cc33-4d44-8e55-6f7788990011', + timestamp: '2026-08-05T10:00:09.000Z', + sessionId: CLAUDE_SESSION + } + ]) + const journal = await open('claude', CLAUDE_SESSION) + await importLegacyTranscriptIntoJournal({ + journal, + agent: 'claude', + sessionId: CLAUDE_SESSION, + fence: 1, + options: { filePath, limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, inlineHeadBytes: 1_024 } } + }) + + const item = journal.snapshot().items[0] + expect(item?.body).toMatchObject({ kind: 'tool-call', state: 'completed' }) + const body = item?.body + if (body?.kind !== 'tool-call' || !body.output) { + throw new Error('expected a bounded tool-call output') + } + expect(body.output.truncated).toBe(true) + expect(body.output.byteLength).toBe(64 * 1024) + expect(body.output.head).toHaveLength(1_024) + expect(await readJournalBlob(root, body.output.digest)).toBe(output) + }) +}) + +describe('import failures', () => { + it('reports a missing transcript without touching the journal', async () => { + const journal = await open('claude', CLAUDE_SESSION) + const before = journal.epoch + const result = await importLegacyTranscriptIntoJournal({ + journal, + agent: 'claude', + sessionId: CLAUDE_SESSION, + fence: 1, + options: { filePath: join(root, 'missing.jsonl') } + }) + expect(result).toMatchObject({ ok: false }) + expect(journal.epoch).toBe(before) + }) + + it('rejects an agent with no transcript decoder', async () => { + const journal = await open('claude', CLAUDE_SESSION) + const result = await importLegacyTranscriptIntoJournal({ + journal, + agent: 'gemini', + sessionId: CLAUDE_SESSION, + fence: 1, + options: { filePath: join(root, 'claude.jsonl') } + }) + expect(result).toMatchObject({ ok: false }) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-legacy-import.ts b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts new file mode 100644 index 00000000000..a95d0ace7b8 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-legacy-import.ts @@ -0,0 +1,200 @@ +// Hydrating a journal from a bridge-era transcript. +// +// This reuses the existing per-agent transcript decoders verbatim — a second +// parser would drift from the one the live view already uses. The decoders +// return a render model with no identity, so the import wraps them: the wrapper +// reads an identity anchor off the SAME raw line, then delegates the content. +// +// Import always opens a fresh epoch. The imported timeline is a best-effort +// reconstruction with import-scoped identities for most providers, so it must +// never be spliced into a sequence space that a structured session is also +// writing; a later structured resume rolls the epoch again and rebuilds. + +import { createReadStream } from 'node:fs' +import type { AgentType } from '../../../shared/agent-status-types' +import type { + AgentJournalCursor, + AgentJournalItemBody, + AgentJournalItemIdentity +} from '../../../shared/agent-session-journal-types' +import type { NativeChatBlock, NativeChatMessage } from '../../../shared/native-chat-types' +import { resolveNativeChatTranscriptAgent } from '../../../shared/native-chat-agent-support' +import { resolveSessionFilePath, type ResolveSessionFileOptions } from '../session-file-resolver' +import { + decodeClaudeTranscriptLine, + decodeCodexTranscriptLine, + decodeGrokTranscriptLine, + decodeOmpTranscriptLine +} from '../transcript-line-decoders' +import { decodeTranscriptStream } from '../transcript-stream-lines' +import { putJournalBlob } from './journal-blob-store' +import { createLegacyIdentityTracker } from './journal-legacy-identity' +import { + boundInlineText, + boundPayload, + DEFAULT_JOURNAL_PAYLOAD_LIMITS, + type JournalPayloadLimits +} from './journal-payload-bounds' +import type { AgentSessionJournal } from './journal-store' + +export type LegacyImportOptions = ResolveSessionFileOptions & { + /** Resolve directly to this file, skipping path discovery. */ + filePath?: string + limits?: JournalPayloadLimits +} + +export type LegacyImportResult = + | { ok: true; epoch: string; cursor: AgentJournalCursor; imported: number } + | { ok: false; error: string } + +export async function importLegacyTranscriptIntoJournal(input: { + journal: AgentSessionJournal + agent: AgentType + sessionId: string + fence: number + options?: LegacyImportOptions +}): Promise { + const options = input.options ?? {} + const limits = options.limits ?? DEFAULT_JOURNAL_PAYLOAD_LIMITS + const transcriptAgent = resolveNativeChatTranscriptAgent(input.agent) + if (!transcriptAgent) { + return { ok: false, error: `Unsupported agent for journal import: ${input.agent}` } + } + const filePath = + options.filePath ?? (await resolveSessionFilePath(input.agent, input.sessionId, options)) + if (!filePath) { + return { ok: false, error: `No transcript found for ${input.agent} session ${input.sessionId}` } + } + + let decoded: { messages: NativeChatMessage[]; identities: AgentJournalItemIdentity[] } + try { + decoded = await decodeWithIdentities({ + filePath, + transcriptAgent, + agent: input.agent, + sessionId: input.sessionId + }) + } catch (err) { + return { ok: false, error: err instanceof Error ? err.message : String(err) } + } + + await input.journal.rollEpoch('legacy_import', input.fence) + let cursor = input.journal.cursor() + for (const [index, message] of decoded.messages.entries()) { + const identity = decoded.identities[index] + if (!identity) { + continue + } + const mapped = legacyItemBody(message, limits) + for (const blob of mapped.blobs) { + await putJournalBlob(input.journal.directory, blob.digest, blob.payload) + } + const appended = await input.journal.appendItem(identity, mapped.body, { + fence: input.fence, + observedAt: message.timestamp ?? undefined + }) + cursor = appended.cursor + } + return { ok: true, epoch: input.journal.epoch, cursor, imported: decoded.messages.length } +} + +const TRANSCRIPT_DECODERS = { + claude: decodeClaudeTranscriptLine, + codex: decodeCodexTranscriptLine, + grok: decodeGrokTranscriptLine, + omp: decodeOmpTranscriptLine +} as const + +/** Run the real decoder while recording an identity anchor per emitted message, + * index-aligned with `messages`. */ +async function decodeWithIdentities(input: { + filePath: string + transcriptAgent: keyof typeof TRANSCRIPT_DECODERS + agent: AgentType + sessionId: string +}): Promise<{ messages: NativeChatMessage[]; identities: AgentJournalItemIdentity[] }> { + const tracker = createLegacyIdentityTracker({ + transcriptAgent: input.transcriptAgent, + agent: input.agent, + sessionId: input.sessionId + }) + const decode = TRANSCRIPT_DECODERS[input.transcriptAgent] + const identities: AgentJournalItemIdentity[] = [] + let lineIndex = 0 + + const stream = createReadStream(input.filePath, { encoding: 'utf-8' }) + const { messages } = await decodeTranscriptStream( + stream, + input.filePath, + 0, + (line, fallbackId) => { + const identity = tracker.identify(line, lineIndex) + lineIndex += 1 + const message = decode(line, fallbackId) + if (message) { + identities.push(identity) + } + return message + }, + true + ) + return { messages, identities } +} + +type MappedLegacyItem = { + body: AgentJournalItemBody + blobs: { digest: string; payload: string }[] +} + +/** + * A message whose only content is a tool invocation becomes a tool-call item so + * the reducer renders it as one. Everything else stays a message item with its + * blocks bounded in place. + */ +function legacyItemBody( + message: NativeChatMessage, + limits: JournalPayloadLimits +): MappedLegacyItem { + const only = message.blocks.length === 1 ? message.blocks[0] : undefined + if (only?.type === 'tool-call') { + return { + body: { kind: 'tool-call', name: only.name, input: only.input, state: 'completed' }, + blobs: [] + } + } + if (only?.type === 'tool-result') { + const output = boundPayload(only.output, limits) + return { + body: { + kind: 'tool-call', + name: 'tool-result', + input: null, + state: only.isError ? 'failed' : 'completed', + output + }, + blobs: output.truncated ? [{ digest: output.digest, payload: only.output }] : [] + } + } + return { + body: { + kind: 'message', + role: message.role, + blocks: message.blocks.map((block) => boundBlock(block, limits)) + }, + blobs: [] + } +} + +/** Inline block text keeps only a bounded head plus an explicit marker. No blob + * is written: the marker carries the digest and byte length, and the source + * transcript remains the full copy — a blob here would be unreferenced by the + * render model and pruned at the next compaction. */ +function boundBlock(block: NativeChatBlock, limits: JournalPayloadLimits): NativeChatBlock { + if (block.type === 'text') { + return { ...block, text: boundInlineText(block.text, limits).text } + } + if (block.type === 'tool-result') { + return { ...block, output: boundInlineText(block.output, limits).text } + } + return block +} diff --git a/src/main/native-chat/agent-session-journal/journal-log-file.ts b/src/main/native-chat/agent-session-journal/journal-log-file.ts new file mode 100644 index 00000000000..795b7436c64 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-log-file.ts @@ -0,0 +1,139 @@ +// On-disk layout for one session's journal. +// +// /log.jsonl append-only rows, fsynced before the caller is told the write landed +// /snapshot.json folded state at a compaction boundary PLUS the retained tail +// /blobs/ bounded-payload remainders +// +// The snapshot carries its own tail so compaction is one atomic write. A crash +// between publishing the snapshot and truncating the log leaves the log a +// superset of the tail, and recovery unions the two by sequence — never a hole. + +import { appendFile, mkdir, open, readFile } from 'node:fs/promises' +import { join } from 'node:path' +import { durableWriteTempPath, writeFileDurable } from '../../durable-file-write' +import type { + AgentJournalRenderItem, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import { parseJournalRow, serializeJournalRow, type JournalRow } from './journal-row-schema' + +export const JOURNAL_LOG_FILE = 'log.jsonl' +export const JOURNAL_SNAPSHOT_FILE = 'snapshot.json' + +export type JournalSnapshotFile = { + v: number + epoch: string + /** Highest sequence folded into `items`; the tail starts after it. */ + compactedThrough: number + items: AgentJournalRenderItem[] + submissions: AgentJournalSubmission[] + /** Receipts outlive the rows that minted them: a client reconnecting after + * compaction must still get the same answer instead of re-sending. */ + receipts: { + clientMessageId: string + providerItemId: string + epoch: string + sequence: number + acceptedAt: number + }[] + /** Provider item id → submission slot, preserved so a post-compaction echo + * still reconciles into the bubble it belongs to. */ + aliases: { providerItemId: string; itemId: string }[] + tail: JournalRow[] +} + +export type JournalReadResult = { + rows: JournalRow[] + /** True when a line used a schema version this build cannot read. Reading + * STOPS there — the row must not be skipped — and the host degrades to + * read-only: no writes, no compaction, no deletion. */ + unreadable: boolean + /** Lines that failed to parse for reasons other than schema version. */ + malformed: number +} + +export async function ensureJournalDir(journalDir: string): Promise { + await mkdir(journalDir, { recursive: true }) +} + +export async function readJournalSnapshotFile( + journalDir: string +): Promise { + try { + const raw = await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8') + const parsed = JSON.parse(raw) as JournalSnapshotFile + return typeof parsed?.epoch === 'string' ? parsed : null + } catch { + return null + } +} + +export async function writeJournalSnapshotFile( + journalDir: string, + snapshot: JournalSnapshotFile +): Promise { + const target = join(journalDir, JOURNAL_SNAPSHOT_FILE) + await writeFileDurable(durableWriteTempPath(target), target, JSON.stringify(snapshot)) +} + +export async function readJournalLog(journalDir: string): Promise { + let raw: string + try { + raw = await readFile(join(journalDir, JOURNAL_LOG_FILE), 'utf-8') + } catch { + return { rows: [], unreadable: false, malformed: 0 } + } + const rows: JournalRow[] = [] + let unreadable = false + let malformed = 0 + for (const line of raw.split('\n')) { + if (!line.trim()) { + continue + } + const parsed = parseJournalRow(line) + if (parsed.ok) { + rows.push(parsed.row) + continue + } + if (parsed.unreadable) { + unreadable = true + break + } + malformed += 1 + } + return { rows, unreadable, malformed } +} + +/** + * Append rows and fsync before returning. The caller treats a resolved promise + * as "this row survives a power loss" — the write-ahead submission row depends + * on exactly that, so this must never be relaxed to a buffered write. + */ +export async function appendJournalRows( + journalDir: string, + rows: readonly JournalRow[] +): Promise { + if (rows.length === 0) { + return + } + const path = join(journalDir, JOURNAL_LOG_FILE) + const payload = `${rows.map(serializeJournalRow).join('\n')}\n` + await appendFile(path, payload, 'utf-8') + const handle = await open(path, 'r+') + try { + await handle.sync() + } finally { + await handle.close() + } +} + +/** Replace the log with exactly the retained tail. Runs only after the snapshot + * carrying that tail is durable, so a crash here loses nothing. */ +export async function rewriteJournalLog( + journalDir: string, + rows: readonly JournalRow[] +): Promise { + const target = join(journalDir, JOURNAL_LOG_FILE) + const payload = rows.length ? `${rows.map(serializeJournalRow).join('\n')}\n` : '' + await writeFileDurable(durableWriteTempPath(target), target, payload) +} diff --git a/src/main/native-chat/agent-session-journal/journal-open.ts b/src/main/native-chat/agent-session-journal/journal-open.ts new file mode 100644 index 00000000000..9fbd730fb01 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-open.ts @@ -0,0 +1,154 @@ +// Loading a journal from disk: snapshot + log → folded state. +// +// The snapshot is authoritative for the current epoch. Log rows belonging to a +// superseded epoch are dropped rather than merged — a crash between publishing +// a rollover snapshot and rewriting the log is the ordinary way that happens. +// A gap in the surviving sequence is corruption, and the caller rolls the epoch +// rather than rendering a partial timeline. + +import type { AgentJournalSubmission } from '../../../shared/agent-session-journal-types' +import { findSequenceGap } from './journal-cursor' +import { + readJournalLog, + readJournalSnapshotFile, + type JournalSnapshotFile +} from './journal-log-file' +import { + applyJournalRow, + createJournalReducerState, + type JournalReducerState +} from './journal-reducer' +import { journalRowByteLength, type JournalRow } from './journal-row-schema' + +export type JournalLoad = { + state: JournalReducerState + /** Rows still individually replayable, oldest first. */ + tailRows: JournalRow[] + /** Highest sequence folded into the snapshot; the tail starts after it. */ + compactedThrough: number + /** A future schema version was met: no writes, no compaction, no deletion. */ + readOnly: boolean + /** Set when the surviving prefix is unusable and the caller must roll the epoch. */ + corrupt: boolean + sizeBytes: number +} + +/** Returns null when no journal exists yet for this session. */ +export async function loadJournal( + journalDir: string, + sessionId: string +): Promise { + const snapshot = await readJournalSnapshotFile(journalDir) + const log = await readJournalLog(journalDir) + const epoch = resolveEpoch(snapshot, log.rows) + if (!epoch) { + return log.unreadable ? emptyReadOnlyLoad(sessionId) : null + } + + const compactedThrough = snapshot?.epoch === epoch ? snapshot.compactedThrough : 0 + const state = seedState(sessionId, epoch, snapshot?.epoch === epoch ? snapshot : null) + const liveRows = log.rows.filter((row) => row.epoch === epoch) + const tailRows = unionBySequence(snapshot?.epoch === epoch ? snapshot.tail : [], liveRows, epoch) + + const oldest = tailRows[0]?.seq ?? compactedThrough + 1 + const gap = findSequenceGap( + tailRows.map((row) => row.seq), + oldest + ) + // A hole below the snapshot boundary is unrecoverable too: the snapshot only + // covers `compactedThrough`, so a tail that starts above it lost rows. + const corrupt = Boolean(gap) || oldest > compactedThrough + 1 + + for (const row of liveRows) { + if (row.seq > compactedThrough) { + applyJournalRow(state, row) + } + } + state.oldestSequence = oldest + state.lastSequence = Math.max(state.lastSequence, compactedThrough) + + return { + state, + tailRows, + compactedThrough, + readOnly: log.unreadable, + corrupt, + sizeBytes: tailRows.reduce((total, row) => total + journalRowByteLength(row), 0) + } +} + +function emptyReadOnlyLoad(sessionId: string): JournalLoad { + const state = createJournalReducerState(sessionId, '') + return { + state, + tailRows: [], + compactedThrough: 0, + readOnly: true, + corrupt: false, + sizeBytes: 0 + } +} + +/** The snapshot names the live epoch; without one, the newest epoch row does. */ +function resolveEpoch(snapshot: JournalSnapshotFile | null, rows: JournalRow[]): string | null { + if (snapshot?.epoch) { + return snapshot.epoch + } + for (let index = rows.length - 1; index >= 0; index -= 1) { + const row = rows[index] + if (row?.kind === 'epoch') { + return row.epoch + } + } + return null +} + +function seedState( + sessionId: string, + epoch: string, + snapshot: JournalSnapshotFile | null +): JournalReducerState { + const state = createJournalReducerState(sessionId, epoch) + if (!snapshot) { + return state + } + for (const item of snapshot.items) { + state.items.set(item.itemId, item) + } + for (const submission of snapshot.submissions) { + state.submissions.set(submission.clientMessageId, { ...submission } as AgentJournalSubmission) + } + for (const receipt of snapshot.receipts) { + state.receipts.set(receipt.clientMessageId, { + clientMessageId: receipt.clientMessageId, + providerItemId: receipt.providerItemId, + cursor: { epoch: receipt.epoch, sequence: receipt.sequence }, + acceptedAt: receipt.acceptedAt + }) + } + for (const alias of snapshot.aliases) { + state.aliases.set(alias.providerItemId, alias.itemId) + } + state.lastSequence = snapshot.compactedThrough + state.oldestSequence = snapshot.compactedThrough + 1 + return state +} + +/** Merge the snapshot's retained tail with the live log, preferring the log's + * copy of any sequence both hold, and dropping rows from a superseded epoch. */ +function unionBySequence( + retained: readonly JournalRow[], + live: readonly JournalRow[], + epoch: string +): JournalRow[] { + const bySequence = new Map() + for (const row of retained) { + if (row.epoch === epoch) { + bySequence.set(row.seq, row) + } + } + for (const row of live) { + bySequence.set(row.seq, row) + } + return [...bySequence.values()].sort((a, b) => a.seq - b.seq) +} diff --git a/src/main/native-chat/agent-session-journal/journal-paths.ts b/src/main/native-chat/agent-session-journal/journal-paths.ts new file mode 100644 index 00000000000..7019b960ecc --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-paths.ts @@ -0,0 +1,43 @@ +// Where a session journal lives. +// +// Host-side per-workspace state, keyed by workspace id — never inside the +// user's working tree. A journal in the tree would show up in `git status`, +// vanish with `git worktree remove`, and have no defined home in a folder +// workspace that is not a repository at all. Keying by id rather than by path +// makes a worktree, a folder workspace, a WSL distro, and an SSH host identical. +// +// `app` is imported lazily so the store stays usable from tests and from the +// headless/SSH runtime, which have no Electron app object. + +import { createHash } from 'node:crypto' +import { join } from 'node:path' +import type { AgentSessionJournalIdentity } from '../../../shared/agent-session-journal-types' + +const JOURNAL_DIR_NAME = 'agent-session-journal' + +/** Filesystem-safe, collision-resistant segment for an arbitrary id. Ids come + * from providers and workspaces and can contain path separators or characters + * Windows rejects, so they are hashed rather than sanitized. */ +export function journalPathSegment(value: string): string { + return createHash('sha256').update(value, 'utf8').digest('hex').slice(0, 32) +} + +/** `/agent-session-journal//`. */ +export function journalDirectoryFor( + root: string, + identity: Pick +): string { + return join( + root, + JOURNAL_DIR_NAME, + journalPathSegment(identity.workspaceId), + journalPathSegment(identity.sessionId) + ) +} + +/** Default host state root. Electron is required lazily so importing this + * module from a non-Electron runtime does not pull the whole app in. */ +export async function defaultJournalRoot(): Promise { + const { app } = await import('electron') + return app.getPath('userData') +} diff --git a/src/main/native-chat/agent-session-journal/journal-payload-bounds.ts b/src/main/native-chat/agent-session-journal/journal-payload-bounds.ts new file mode 100644 index 00000000000..61cb82894a4 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-payload-bounds.ts @@ -0,0 +1,86 @@ +// Payload bounds for tool output and diffs. +// +// A looping agent must not be able to fill the host disk, and a 40 MB tool +// result must not be inlined into a row that every reconnecting client +// replays. A bounded payload keeps a head plus the original byte length and +// digest; the remainder lives in the content-addressed blob store under the +// same retention as its epoch. Crossing a bound is always marked — never a +// silent drop. + +import { createHash } from 'node:crypto' +import type { AgentJournalBoundedPayload } from '../../../shared/agent-session-journal-types' + +export type JournalPayloadLimits = { + /** Bytes of the payload kept inline on the row. */ + inlineHeadBytes: number + /** Total bytes of journal rows one session may hold before appends are refused. */ + maxSessionBytes: number + /** Appends allowed inside `appendWindowMs`, bounding a runaway agent's rate. */ + maxAppendsPerWindow: number + appendWindowMs: number +} + +export const DEFAULT_JOURNAL_PAYLOAD_LIMITS: JournalPayloadLimits = { + inlineHeadBytes: 16 * 1024, + maxSessionBytes: 256 * 1024 * 1024, + maxAppendsPerWindow: 5000, + appendWindowMs: 60_000 +} + +/** Marker appended to a clipped inline string so the UI never presents a + * truncated body as complete. Kept in the text itself because block-level + * payloads (tool-result output) have nowhere else to carry the flag. */ +export function journalTruncationMarker(byteLength: number, digest: string): string { + return `\n[Orca: output truncated — ${byteLength} bytes total, digest ${digest.slice(0, 12)}]` +} + +export function digestPayload(payload: string): string { + return createHash('sha256').update(payload, 'utf8').digest('hex') +} + +/** + * Clip `payload` to the inline head. `truncated` means the remainder must be + * written to the blob store under `digest` before the row is appended. + */ +export function boundPayload( + payload: string, + limits: JournalPayloadLimits +): AgentJournalBoundedPayload { + const buffer = Buffer.from(payload, 'utf8') + const digest = digestPayload(payload) + if (buffer.byteLength <= limits.inlineHeadBytes) { + return { head: payload, byteLength: buffer.byteLength, digest, truncated: false } + } + return { + head: clipUtf8(buffer, limits.inlineHeadBytes), + byteLength: buffer.byteLength, + digest, + truncated: true + } +} + +/** Bound a plain string that must stay a string (a tool-result block's output), + * keeping the explicit marker inline. Returns the blob payload to persist. */ +export function boundInlineText( + payload: string, + limits: JournalPayloadLimits +): { text: string; bounded: AgentJournalBoundedPayload } { + const bounded = boundPayload(payload, limits) + if (!bounded.truncated) { + return { text: payload, bounded } + } + return { + text: bounded.head + journalTruncationMarker(bounded.byteLength, bounded.digest), + bounded + } +} + +/** Slice at a byte budget without splitting a multi-byte character. */ +function clipUtf8(buffer: Buffer, maxBytes: number): string { + let end = maxBytes + // A UTF-8 continuation byte is 0b10xxxxxx; walk back off a split sequence. + while (end > 0 && (buffer[end] & 0b1100_0000) === 0b1000_0000) { + end -= 1 + } + return buffer.subarray(0, end).toString('utf8') +} diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.test.ts b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts new file mode 100644 index 00000000000..3ebb1dffc85 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-reducer.test.ts @@ -0,0 +1,295 @@ +import { describe, expect, it } from 'vitest' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { AgentJournalMessageItem } from '../../../shared/agent-session-journal-types' +import { + applyJournalRow, + createJournalReducerState, + referencedBlobDigests, + renderJournalState, + type JournalReducerState +} from './journal-reducer' +import type { JournalRow } from './journal-row-schema' + +const EPOCH = 'epoch-1' + +function text(value: string): AgentJournalMessageItem { + return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: value }] } +} + +function base(seq: number): { v: number; epoch: string; seq: number; fence: number; ts: number } { + return { v: 1, epoch: EPOCH, seq, fence: 1, ts: 1_000 + seq } +} + +function fold(rows: JournalRow[]): JournalReducerState { + const state = createJournalReducerState('session-1', EPOCH) + for (const row of rows) { + applyJournalRow(state, row) + } + return state +} + +describe('revisions and tombstones', () => { + it('takes the highest revision', () => { + const state = fold([ + { kind: 'item', itemId: 'a', revision: 1, body: text('first'), ...base(1) }, + { kind: 'item', itemId: 'a', revision: 2, body: text('second'), ...base(2) } + ]) + expect(renderJournalState(state).items[0]?.body).toEqual(text('second')) + }) + + it('drops a late lower revision instead of resurrecting stale content', () => { + const state = fold([ + { kind: 'item', itemId: 'a', revision: 2, body: text('second'), ...base(1) }, + { kind: 'item', itemId: 'a', revision: 1, body: text('first'), ...base(2) } + ]) + expect(renderJournalState(state).items[0]?.body).toEqual(text('second')) + }) + + it('removes an item on a tombstone', () => { + const state = fold([ + { kind: 'item', itemId: 'a', revision: 1, body: text('gone'), ...base(1) }, + { kind: 'tombstone', itemId: 'a', revision: 2, ...base(2) } + ]) + expect(renderJournalState(state).items).toHaveLength(0) + }) + + it('does not let a late lower revision resurrect a tombstoned item', () => { + const state = fold([ + { kind: 'item', itemId: 'a', revision: 1, body: text('gone'), ...base(1) }, + { kind: 'tombstone', itemId: 'a', revision: 3, ...base(2) }, + { kind: 'item', itemId: 'a', revision: 2, body: text('stale'), ...base(3) } + ]) + expect(renderJournalState(state).items).toHaveLength(0) + }) + + it('re-creates an item at a revision above the tombstone', () => { + const state = fold([ + { kind: 'item', itemId: 'a', revision: 1, body: text('gone'), ...base(1) }, + { kind: 'tombstone', itemId: 'a', revision: 2, ...base(2) }, + { kind: 'item', itemId: 'a', revision: 3, body: text('back'), ...base(3) } + ]) + expect(renderJournalState(state).items.map((item) => item.body)).toEqual([text('back')]) + }) +}) + +describe('ordering', () => { + it('orders by the sequence that created an item, not by a later revision', () => { + const state = fold([ + { kind: 'item', itemId: 'a', revision: 1, body: text('a'), ...base(1) }, + { kind: 'item', itemId: 'b', revision: 1, body: text('b'), ...base(2) }, + { kind: 'item', itemId: 'a', revision: 2, body: text('a2'), ...base(3) } + ]) + expect(renderJournalState(state).items.map((item) => item.itemId)).toEqual(['a', 'b']) + }) + + it('orders by sequence regardless of the order rows are applied in', () => { + // Live append and replay must render the same list, so the fold cannot lean + // on the order it happens to be handed rows in. + const state = fold([ + { kind: 'item', itemId: 'later', revision: 1, body: text('later'), ...base(9) }, + { kind: 'item', itemId: 'earlier', revision: 1, body: text('earlier'), ...base(3) } + ]) + expect(renderJournalState(state).items.map((item) => item.itemId)).toEqual(['earlier', 'later']) + }) + + it('orders by sequence even when the observed timestamp runs backwards', () => { + const state = fold([ + { kind: 'item', itemId: 'late', revision: 1, body: text('late'), ...base(1), ts: 9_000 }, + { + kind: 'item', + itemId: 'recovered', + revision: 1, + body: text('recovered'), + ...base(2), + ts: 10, + recovered: true + } + ]) + const items = renderJournalState(state).items + expect(items.map((item) => item.itemId)).toEqual(['late', 'recovered']) + expect(items[1]?.recovered).toBe(true) + }) + + it('never collapses two items that carry identical text', () => { + const state = fold([ + { kind: 'item', itemId: 'a', revision: 1, body: text('run the tests'), ...base(1) }, + { kind: 'item', itemId: 'b', revision: 1, body: text('run the tests'), ...base(2) } + ]) + expect(renderJournalState(state).items).toHaveLength(2) + }) +}) + +describe('submission and dispatch state machine', () => { + const submission: JournalRow = { + kind: 'submission', + clientMessageId: 'cm_1', + payloadFingerprint: 'fp_1', + providerHandle: { kind: 'codex', threadId: 'thread-1' }, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }, + ...base(1) + } + + it('seeds a pending submission and an optimistic bubble', () => { + const rendered = renderJournalState(fold([submission])) + expect(rendered.submissions[0]?.dispatchState).toBe('pending') + expect(rendered.items.map((item) => item.itemId)).toEqual([agentJournalSubmissionKey('cm_1')]) + }) + + it('mints a receipt and adopts the provider item id on accept', () => { + const state = fold([ + submission, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'accepted', + providerItemId: 'codex:thread-1:turn-1:0', + reason: null, + ...base(2) + } + ]) + expect(state.receipts.get('cm_1')?.cursor).toEqual({ epoch: EPOCH, sequence: 2 }) + expect(state.submissions.get('cm_1')?.providerItemId).toBe('codex:thread-1:turn-1:0') + }) + + it('folds the provider echo into the submission bubble instead of adding a second one', () => { + const state = fold([ + submission, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'accepted', + providerItemId: 'codex:thread-1:turn-1:0', + reason: null, + ...base(2) + }, + { + kind: 'item', + itemId: 'codex:thread-1:turn-1:0', + revision: 1, + body: { kind: 'message', role: 'user', blocks: [{ type: 'text', text: 'hi' }] }, + ...base(3) + } + ]) + const items = renderJournalState(state).items + expect(items).toHaveLength(1) + expect(items[0]?.itemId).toBe(agentJournalSubmissionKey('cm_1')) + // The echo updates content in place; the bubble keeps its original slot. + expect(items[0]?.sequence).toBe(1) + expect(items[0]?.revision).toBe(1) + }) + + it('treats rejected as terminal', () => { + const state = fold([ + submission, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'rejected', + providerItemId: null, + reason: 'not_delivered', + ...base(2) + }, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'unknown', + providerItemId: null, + reason: 'late', + ...base(3) + } + ]) + expect(state.submissions.get('cm_1')?.dispatchState).toBe('rejected') + expect(state.submissions.get('cm_1')?.reason).toBe('not_delivered') + }) + + it('lets an unknown submission settle later', () => { + const state = fold([ + submission, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'unknown', + providerItemId: null, + reason: 'host_restarted_before_acknowledgement', + ...base(2) + }, + { + kind: 'dispatch', + clientMessageId: 'cm_1', + state: 'accepted', + providerItemId: 'p1', + reason: null, + ...base(3) + } + ]) + expect(state.submissions.get('cm_1')?.dispatchState).toBe('accepted') + expect(state.receipts.get('cm_1')).toBeTruthy() + }) + + it('ignores a dispatch for a submission this epoch never saw', () => { + const state = fold([ + { + kind: 'dispatch', + clientMessageId: 'ghost', + state: 'accepted', + providerItemId: 'p', + reason: null, + ...base(1) + } + ]) + expect(state.submissions.size).toBe(0) + expect(state.receipts.size).toBe(0) + }) +}) + +describe('blob retention', () => { + it('reports the digests live rows still reference', () => { + const state = fold([ + { + kind: 'item', + itemId: 'tool', + revision: 1, + body: { + kind: 'tool-call', + name: 'bash', + input: {}, + state: 'completed', + output: { head: 'x', byteLength: 999, digest: 'digest-a', truncated: true } + }, + ...base(1) + }, + { + kind: 'item', + itemId: 'inline', + revision: 1, + body: { + kind: 'tool-call', + name: 'bash', + input: {}, + state: 'completed', + output: { head: 'y', byteLength: 1, digest: 'digest-b', truncated: false } + }, + ...base(2) + } + ]) + expect([...referencedBlobDigests(state)]).toEqual(['digest-a']) + }) + + it('stops referencing a digest once its item is tombstoned', () => { + const state = fold([ + { + kind: 'item', + itemId: 'tool', + revision: 1, + body: { + kind: 'diff', + path: 'a.ts', + patch: { head: 'x', byteLength: 999, digest: 'digest-a', truncated: true } + }, + ...base(1) + }, + { kind: 'tombstone', itemId: 'tool', revision: 2, ...base(2) } + ]) + expect(referencedBlobDigests(state).size).toBe(0) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-reducer.ts b/src/main/native-chat/agent-session-journal/journal-reducer.ts new file mode 100644 index 00000000000..31898d4c5e7 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-reducer.ts @@ -0,0 +1,199 @@ +// THE reducer. One implementation folds rows into the render model, and both +// the live append path and replay call it — a live-only shortcut is how a +// reconnect starts disagreeing with the screen it replaced. +// +// Rules: highest revision wins, a tombstone removes, a late lower revision is +// dropped rather than resurrecting stale content, and ordering is by the +// sequence of the row that CREATED an item (a later revision updates the body, +// it does not move the bubble). + +import type { + AgentJournalAcceptanceReceipt, + AgentJournalRenderItem, + AgentJournalSnapshot, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import { agentJournalSubmissionKey } from '../../../shared/agent-session-journal-item-key' +import type { JournalRow } from './journal-row-schema' + +export type JournalReducerState = { + sessionId: string + epoch: string + lastSequence: number + /** Lowest sequence still individually replayable; rows below it were compacted. */ + oldestSequence: number + highestFence: number + items: Map + /** Revision of a removed item, so a late lower revision cannot resurrect it. */ + tombstones: Map + submissions: Map + receipts: Map + /** Provider item id → the submission slot that adopted it. Stops an accepted + * echo from appending a second copy of the user's own message. */ + aliases: Map +} + +export function createJournalReducerState(sessionId: string, epoch: string): JournalReducerState { + return { + sessionId, + epoch, + lastSequence: 0, + oldestSequence: 1, + highestFence: 0, + items: new Map(), + tombstones: new Map(), + submissions: new Map(), + receipts: new Map(), + aliases: new Map() + } +} + +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 + } + if (row.kind === 'item') { + const itemId = resolveItemId(state, row.itemId) + upsertItem(state, itemId, row.revision, { + itemId, + revision: row.revision, + body: row.body, + sequence: row.seq, + observedAt: row.ts, + ...(row.recovered ? { recovered: row.recovered } : {}) + }) + return + } + if (row.kind === 'tombstone') { + removeItem(state, resolveItemId(state, row.itemId), row.revision) + return + } + if (row.kind === 'submission') { + applySubmission(state, row) + return + } + applyDispatch(state, row) +} + +function resolveItemId(state: JournalReducerState, itemId: string): string { + return state.aliases.get(itemId) ?? itemId +} + +function upsertItem( + state: JournalReducerState, + itemId: string, + revision: number, + next: AgentJournalRenderItem +): void { + const tombstoned = state.tombstones.get(itemId) + if (tombstoned !== undefined && revision <= tombstoned) { + return + } + const existing = state.items.get(itemId) + if (existing && revision <= existing.revision) { + return + } + if (!existing) { + state.items.set(itemId, next) + state.tombstones.delete(itemId) + return + } + // Creation sequence is the ordering key; a revision refreshes content only. + state.items.set(itemId, { ...next, sequence: existing.sequence }) + state.tombstones.delete(itemId) +} + +function removeItem(state: JournalReducerState, itemId: string, revision: number): void { + const existing = state.items.get(itemId) + if (existing && revision <= existing.revision) { + return + } + const tombstoned = state.tombstones.get(itemId) + if (tombstoned !== undefined && revision <= tombstoned) { + return + } + state.tombstones.set(itemId, revision) + state.items.delete(itemId) +} + +function applySubmission( + state: JournalReducerState, + row: Extract +): void { + state.submissions.set(row.clientMessageId, { + clientMessageId: row.clientMessageId, + fence: row.fence, + payloadFingerprint: row.payloadFingerprint, + dispatchState: 'pending', + providerItemId: null, + reason: null, + submittedAt: row.ts, + resolvedAt: null + }) + const itemId = agentJournalSubmissionKey(row.clientMessageId) + upsertItem(state, itemId, 0, { + itemId, + revision: 0, + body: row.body, + sequence: row.seq, + observedAt: row.ts + }) +} + +function applyDispatch( + state: JournalReducerState, + row: Extract +): void { + const submission = state.submissions.get(row.clientMessageId) + if (!submission) { + return + } + // `rejected` is terminal; a late `unknown` must not reopen a settled answer. + if (submission.dispatchState === 'rejected' || submission.dispatchState === 'accepted') { + return + } + submission.dispatchState = row.state + submission.providerItemId = row.providerItemId + submission.reason = row.reason + submission.resolvedAt = row.ts + if (row.state !== 'accepted' || !row.providerItemId) { + return + } + state.aliases.set(row.providerItemId, agentJournalSubmissionKey(row.clientMessageId)) + state.receipts.set(row.clientMessageId, { + clientMessageId: row.clientMessageId, + providerItemId: row.providerItemId, + cursor: { epoch: row.epoch, sequence: row.seq }, + acceptedAt: row.ts + }) +} + +/** Project the folded state into the client-facing snapshot. */ +export function renderJournalState(state: JournalReducerState): AgentJournalSnapshot { + // Sequence is the sole ordering key; map insertion order is not, because a + // re-created item re-enters the map after the items that followed it. + const items = [...state.items.values()].sort((a, b) => a.sequence - b.sequence) + return { + sessionId: state.sessionId, + cursor: { epoch: state.epoch, sequence: state.lastSequence }, + items, + submissions: [...state.submissions.values()].sort((a, b) => a.submittedAt - b.submittedAt) + } +} + +/** Digests referenced by live rows, so compaction knows which blobs to keep. */ +export function referencedBlobDigests(state: JournalReducerState): Set { + const digests = new Set() + for (const item of state.items.values()) { + const body = item.body + if (body.kind === 'tool-call' && body.output?.truncated) { + digests.add(body.output.digest) + } + if (body.kind === 'diff' && body.patch.truncated) { + digests.add(body.patch.digest) + } + } + return digests +} 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 new file mode 100644 index 00000000000..6f89d9e6395 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -0,0 +1,163 @@ +// Persisted journal row shapes plus read-time upcasting. +// +// The journal is append-only, so migration is upcasting on read and never an +// 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. + +import { + AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + type AgentJournalDispatchState, + type AgentJournalItemBody, + type AgentJournalMessageItem, + type AgentSessionProviderHandle +} from '../../../shared/agent-session-journal-types' + +type JournalRowBase = { + /** Schema version of THIS row. */ + v: number + epoch: string + seq: number + /** Runtime fence held by the writer that appended the row. */ + fence: number + /** Observed (provider or host) timestamp. Ordering is by `seq`, not by this. */ + ts: number + /** Set when crash reconciliation appended the row after the fact. */ + recovered?: true +} + +/** First row of every epoch: binds the epoch to a provider handle and records why it opened. */ +export type JournalEpochRow = JournalRowBase & { + kind: 'epoch' + reason: AgentJournalEpochReason + providerHandle: AgentSessionProviderHandle +} + +export const AGENT_JOURNAL_EPOCH_REASONS = [ + 'session_created', + 'legacy_import', + 'corruption', + 'unreconcilable_prefix', + 'handle_forked', + 'schema_unreadable' +] as const +export type AgentJournalEpochReason = (typeof AGENT_JOURNAL_EPOCH_REASONS)[number] + +export type JournalItemRow = JournalRowBase & { + kind: 'item' + itemId: string + revision: number + body: AgentJournalItemBody +} + +export type JournalTombstoneRow = JournalRowBase & { + kind: 'tombstone' + itemId: string + revision: number +} + +/** The write-ahead row. Durable BEFORE the adapter dispatches anything; it + * doubles as the optimistic user bubble so an accepted echo has a slot to + * reconcile into instead of appending a second copy. */ +export type JournalSubmissionRow = JournalRowBase & { + kind: 'submission' + clientMessageId: string + payloadFingerprint: string + providerHandle: AgentSessionProviderHandle + body: AgentJournalMessageItem +} + +export type JournalDispatchRow = JournalRowBase & { + kind: 'dispatch' + clientMessageId: string + state: Exclude + /** Provider item identity adopted on accept. */ + providerItemId: string | null + reason: string | null +} + +export type JournalRow = + | JournalEpochRow + | JournalItemRow + | JournalTombstoneRow + | JournalSubmissionRow + | JournalDispatchRow + +export type JournalRowParse = + | { ok: true; row: JournalRow } + /** Malformed JSON or a shape this build rejects outright. */ + | { ok: false; unreadable: false } + /** A future schema version. The host must not write or compact this journal. */ + | { ok: false; unreadable: true } + +const ROW_KINDS = new Set(['epoch', 'item', 'tombstone', 'submission', 'dispatch']) + +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. + */ +export function parseJournalRow(line: string): JournalRowParse { + let parsed: unknown + try { + parsed = JSON.parse(line) + } catch { + return { ok: false, unreadable: false } + } + if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) { + return { ok: false, unreadable: false } + } + const record = parsed as Record + const version = typeof record.v === 'number' ? record.v : null + if (version === null || !Number.isInteger(version) || version < 1) { + return { ok: false, unreadable: false } + } + if (version > AGENT_SESSION_JOURNAL_SCHEMA_VERSION) { + return { ok: false, unreadable: true } + } + const upcast = upcastRow(record, version) + return isJournalRow(upcast) ? { ok: true, row: upcast } : { ok: false, unreadable: false } +} + +/** Read-time upcast chain. Each step raises a row exactly one version. */ +function upcastRow(record: Record, version: number): Record { + let current = record + let at = version + while (at < AGENT_SESSION_JOURNAL_SCHEMA_VERSION) { + // No upcasters yet — v1 is the first shipped schema. New cases go here. + current = { ...current, v: at + 1 } + at += 1 + } + return current +} + +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' + ) { + return false + } + if (record.kind === 'item' || record.kind === 'tombstone') { + return typeof record.itemId === 'string' && Number.isInteger(record.revision) + } + if (record.kind === 'submission' || record.kind === 'dispatch') { + return typeof record.clientMessageId === 'string' && record.clientMessageId.length > 0 + } + return typeof record.reason === 'string' +} + +/** Approximate on-disk cost of a row, used for the per-session size bound. */ +export function journalRowByteLength(row: JournalRow): number { + return Buffer.byteLength(serializeJournalRow(row), 'utf8') + 1 +} diff --git a/src/main/native-chat/agent-session-journal/journal-store.test.ts b/src/main/native-chat/agent-session-journal/journal-store.test.ts new file mode 100644 index 00000000000..0d13061b264 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-store.test.ts @@ -0,0 +1,357 @@ +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' +import { afterEach, beforeEach, describe, expect, it } from 'vitest' +import type { + AgentJournalItemBody, + AgentJournalItemIdentity, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { readJournalBlob } from './journal-blob-store' +import { JOURNAL_LOG_FILE, JOURNAL_SNAPSHOT_FILE } from './journal-log-file' +import { + boundInlineText, + boundPayload, + DEFAULT_JOURNAL_PAYLOAD_LIMITS +} from './journal-payload-bounds' +import { journalDirectoryFor, journalPathSegment } from './journal-paths' +import { + AgentSessionJournalError, + openAgentSessionJournal, + type AgentSessionJournal +} from './journal-store' + +const IDENTITY: AgentSessionJournalIdentity = { + sessionId: 'session-1', + workspaceId: 'ws-1', + hostId: 'host-1', + agent: 'codex', + providerHandle: { kind: 'codex', threadId: 'thread-1' } +} + +let root: string +let clock = 1_000 + +function tick(): number { + clock += 1 + return clock +} + +function item(ordinal: number): AgentJournalItemIdentity { + return { provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal } +} + +function body(value: string): AgentJournalItemBody { + return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text: value }] } +} + +async function open(overrides: Partial[0]> = {}) { + return openAgentSessionJournal({ + identity: IDENTITY, + journalDir: root, + now: tick, + mintEpoch: () => `epoch-${clock}`, + ...overrides + }) +} + +beforeEach(async () => { + root = await mkdtemp(join(tmpdir(), 'orca-journal-')) + clock = 1_000 +}) + +afterEach(async () => { + await rm(root, { recursive: true, force: true }) +}) + +describe('sequences', () => { + it('assigns a contiguous sequence with no gaps or reuse under concurrent appends', async () => { + const journal = await open() + const results = await Promise.all( + Array.from({ length: 25 }, (_unused, index) => + journal.appendItem(item(index), body(`m${index}`), { fence: 1 }) + ) + ) + const sequences = results.map((result) => result.cursor.sequence) + expect(new Set(sequences).size).toBe(25) + expect(sequences.slice().sort((a, b) => a - b)).toEqual( + Array.from({ length: 25 }, (_unused, index) => index + 2) + ) + }) + + it('serializes revisions of one item so the last write wins deterministically', async () => { + const journal = await open() + const results = await Promise.all([ + journal.appendItem(item(0), body('a'), { fence: 1 }), + journal.appendItem(item(0), body('b'), { fence: 1 }), + journal.appendItem(item(0), body('c'), { fence: 1 }) + ]) + expect(results.map((result) => result.revision)).toEqual([1, 2, 3]) + expect(journal.snapshot().items).toHaveLength(1) + expect(journal.snapshot().items[0]?.revision).toBe(3) + }) +}) + +describe('fences', () => { + it('rejects an append from a writer behind the journal', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 7 }) + await expect(journal.appendItem(item(1), body('b'), { fence: 6 })).rejects.toBeInstanceOf( + AgentSessionJournalError + ) + }) + + it('keeps accepting appends after a rejected one', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 7 }) + await journal.appendItem(item(1), body('b'), { fence: 6 }).catch(() => undefined) + await journal.appendItem(item(2), body('c'), { fence: 7 }) + expect(journal.snapshot().items.map((entry) => entry.body)).toEqual([body('a'), body('c')]) + }) +}) + +describe('replay', () => { + it('reopens to the same render model the live writer held', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + await journal.appendItem(item(1), body('b'), { fence: 1 }) + await journal.appendItem(item(0), body('a2'), { fence: 1 }) + await journal.appendTombstone(item(1), { fence: 1 }) + const live = journal.snapshot() + + const reopened = await open() + expect(reopened.snapshot()).toEqual(live) + }) + + it('serves a resume from a cursor and refuses one from a stale epoch', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + const cursor = journal.cursor() + await journal.appendItem(item(1), body('b'), { fence: 1 }) + + const resumed = journal.readSince(cursor) + expect(resumed.ok && resumed.rows).toHaveLength(1) + + await journal.rollEpoch('handle_forked', 2) + expect(journal.readSince(cursor)).toEqual({ ok: false, reset: 'epoch_changed' }) + }) + + it('rebuilds from a clean epoch after a rollover', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + await journal.rollEpoch('unreconcilable_prefix', 2) + expect(journal.snapshot().items).toHaveLength(0) + + const reopened = await open() + expect(reopened.epoch).toBe(journal.epoch) + expect(reopened.snapshot().items).toHaveLength(0) + }) + + it('rolls the epoch when the log lost a row', async () => { + const journal = await open() + for (let index = 0; index < 4; index += 1) { + await journal.appendItem(item(index), body(`m${index}`), { fence: 1 }) + } + const before = journal.epoch + const logPath = join(root, JOURNAL_LOG_FILE) + const lines = (await readFile(logPath, 'utf-8')).split('\n').filter(Boolean) + await writeFile(logPath, `${[...lines.slice(0, 2), ...lines.slice(3)].join('\n')}\n`, 'utf-8') + + const reopened = await open() + expect(reopened.epoch).not.toBe(before) + expect(reopened.snapshot().items).toHaveLength(0) + }) +}) + +describe('compaction and retention', () => { + it('folds the prefix into the snapshot and keeps serving the retained tail', async () => { + const journal = await open({ compaction: { minTailRows: 2, retainTailMs: 0 } }) + for (let index = 0; index < 6; index += 1) { + await journal.appendItem(item(index), body(`m${index}`), { fence: 1 }) + } + const rendered = journal.snapshot() + const tip = journal.cursor() + await journal.compact() + + expect(journal.snapshot()).toEqual(rendered) + expect(journal.readSince({ epoch: tip.epoch, sequence: 1 })).toEqual({ + ok: false, + reset: 'cursor_compacted' + }) + const nearTip = journal.readSince({ epoch: tip.epoch, sequence: tip.sequence - 1 }) + expect(nearTip.ok && nearTip.rows).toHaveLength(1) + + const reopened = await open({ compaction: { minTailRows: 2, retainTailMs: 0 } }) + expect(reopened.snapshot()).toEqual(rendered) + expect(reopened.compactionBoundary).toBe(tip.sequence) + }) + + it('publishes the snapshot and its tail as one write, so a crash before the log rewrite loses nothing', async () => { + const journal = await open({ compaction: { minTailRows: 2, retainTailMs: 0 } }) + for (let index = 0; index < 5; index += 1) { + await journal.appendItem(item(index), body(`m${index}`), { fence: 1 }) + } + const rendered = journal.snapshot() + const logBefore = await readFile(join(root, JOURNAL_LOG_FILE), 'utf-8') + await journal.compact() + // Simulate the crash: the snapshot landed, the truncation did not. + await writeFile(join(root, JOURNAL_LOG_FILE), logBefore, 'utf-8') + + const reopened = await open() + expect(reopened.snapshot()).toEqual(rendered) + expect(reopened.snapshot().items).toHaveLength(5) + }) + + it('prunes blobs no live row references and keeps the ones that survive', async () => { + const journal = await open({ compaction: { minTailRows: 1, retainTailMs: 0 } }) + const kept = boundPayload('k'.repeat(64), { + ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, + inlineHeadBytes: 8 + }) + const dropped = boundPayload('d'.repeat(64), { + ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, + inlineHeadBytes: 8 + }) + const { putJournalBlob } = await import('./journal-blob-store') + await putJournalBlob(root, kept.digest, 'k'.repeat(64)) + await putJournalBlob(root, dropped.digest, 'd'.repeat(64)) + await journal.appendItem( + item(0), + { kind: 'tool-call', name: 'bash', input: {}, state: 'completed', output: kept }, + { fence: 1 } + ) + await journal.compact() + + expect(await readJournalBlob(root, kept.digest)).toBe('k'.repeat(64)) + expect(await readJournalBlob(root, dropped.digest)).toBeNull() + }) +}) + +describe('bounds', () => { + it('marks a clipped payload instead of dropping bytes silently', () => { + const limits = { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, inlineHeadBytes: 16 } + const bounded = boundPayload('x'.repeat(4_096), limits) + expect(bounded.truncated).toBe(true) + expect(bounded.head).toHaveLength(16) + expect(bounded.byteLength).toBe(4_096) + expect(boundInlineText('x'.repeat(4_096), limits).text).toContain('output truncated') + }) + + it('never splits a multi-byte character across the bound', () => { + const limits = { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, inlineHeadBytes: 4 } + // Each character is three bytes, so a naive slice would land mid-sequence. + const bounded = boundPayload('日本語テスト', limits) + expect(bounded.head).toBe('日') + expect(Buffer.byteLength(bounded.head, 'utf8')).toBeLessThanOrEqual(4) + }) + + it('leaves a payload inside the bound untouched', () => { + const bounded = boundPayload('small', DEFAULT_JOURNAL_PAYLOAD_LIMITS) + expect(bounded.truncated).toBe(false) + expect(bounded.head).toBe('small') + expect(boundInlineText('small', DEFAULT_JOURNAL_PAYLOAD_LIMITS).text).toBe('small') + }) + + it('refuses an append past the per-session size bound', async () => { + const journal = await open({ + limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxSessionBytes: 400 } + }) + await expect( + (async () => { + for (let index = 0; index < 50; index += 1) { + await journal.appendItem(item(index), body('x'.repeat(64)), { fence: 1 }) + } + })() + ).rejects.toMatchObject({ code: 'journal_bound_exceeded' }) + }) + + it('refuses an append past the per-window rate bound', async () => { + const journal = await open({ + limits: { ...DEFAULT_JOURNAL_PAYLOAD_LIMITS, maxAppendsPerWindow: 3, appendWindowMs: 60_000 } + }) + await expect( + (async () => { + for (let index = 0; index < 10; index += 1) { + await journal.appendItem(item(index), body('x'), { fence: 1 }) + } + })() + ).rejects.toMatchObject({ code: 'journal_rate_exceeded' }) + }) +}) + +describe('schema', () => { + it('degrades to read-only on a row from a newer build, without skipping or deleting it', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + const logPath = join(root, JOURNAL_LOG_FILE) + const future = JSON.stringify({ + v: 99, + kind: 'item', + epoch: journal.epoch, + seq: 99, + fence: 1, + ts: 1, + itemId: 'future', + revision: 1, + body: { kind: 'status', text: 'from a newer host' } + }) + const before = await readFile(logPath, 'utf-8') + await writeFile(logPath, `${before}${future}\n`, 'utf-8') + + const reopened = await open() + expect(reopened.isReadOnly).toBe(true) + await expect(reopened.appendItem(item(1), body('b'), { fence: 1 })).rejects.toMatchObject({ + code: 'journal_read_only' + }) + await expect(reopened.compact()).rejects.toMatchObject({ code: 'journal_read_only' }) + expect(reopened.readSince({ epoch: reopened.epoch, sequence: 0 })).toEqual({ + ok: false, + reset: 'schema_unreadable' + }) + // The unreadable row is still on disk, and nothing was compacted past it. + expect(await readFile(logPath, 'utf-8')).toContain('"v":99') + }) + + it('skips a malformed line without giving up the journal', async () => { + const journal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + const logPath = join(root, JOURNAL_LOG_FILE) + await writeFile(logPath, `${await readFile(logPath, 'utf-8')}{not json\n`, 'utf-8') + + const reopened = await open() + expect(reopened.isReadOnly).toBe(false) + expect(reopened.snapshot().items).toHaveLength(1) + }) +}) + +describe('journal location', () => { + it('keys by workspace and session id rather than by a path in the working tree', () => { + const dir = journalDirectoryFor('/state', { workspaceId: 'ws/1', sessionId: 'sess:2' }) + expect(dir).toBe( + join( + '/state', + 'agent-session-journal', + journalPathSegment('ws/1'), + journalPathSegment('sess:2') + ) + ) + expect(dir).not.toContain('ws/1') + }) + + it('separates two sessions in one workspace', () => { + const a = journalDirectoryFor('/state', { workspaceId: 'ws', sessionId: 'a' }) + const b = journalDirectoryFor('/state', { workspaceId: 'ws', sessionId: 'b' }) + expect(a).not.toBe(b) + }) +}) + +describe('on-disk layout', () => { + it('writes the log and snapshot beside each other', async () => { + const journal: AgentSessionJournal = await open() + await journal.appendItem(item(0), body('a'), { fence: 1 }) + await expect(readFile(join(root, JOURNAL_LOG_FILE), 'utf-8')).resolves.toContain( + '"kind":"item"' + ) + await expect(readFile(join(root, JOURNAL_SNAPSHOT_FILE), 'utf-8')).resolves.toContain('"epoch"') + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts new file mode 100644 index 00000000000..d4cb6358454 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -0,0 +1,368 @@ +// The append-only journal store for one agent session. +// +// Sequence numbers are assigned inside the same serialized append that makes +// the row durable, so an `(epoch, sequence)` space has exactly one writer and +// no gaps or reuse. Every mutating call carries the runtime fence; a stale +// fence is rejected outright rather than merged or queued for the new owner. + +import { randomUUID } from 'node:crypto' +import type { + AgentJournalAcceptanceReceipt, + AgentJournalCursor, + AgentJournalItemBody, + AgentJournalItemIdentity, + AgentJournalMessageItem, + AgentJournalResetReason, + AgentJournalSnapshot, + AgentJournalSubmission, + AgentSessionJournalIdentity +} from '../../../shared/agent-session-journal-types' +import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' +import { compactJournal, type JournalCompactionPolicy } from './journal-compaction' +import { resolveJournalResume, type JournalResume } from './journal-cursor' +import { publishNewEpoch } from './journal-epoch-rollover' +import { appendJournalRows, ensureJournalDir } from './journal-log-file' +import { loadJournal } from './journal-open' +import { DEFAULT_JOURNAL_PAYLOAD_LIMITS, type JournalPayloadLimits } from './journal-payload-bounds' +import { + applyJournalRow, + createJournalReducerState, + renderJournalState, + type JournalReducerState +} from './journal-reducer' +import { + journalRowByteLength, + type AgentJournalEpochReason, + type JournalRow +} from './journal-row-schema' +import { + assertJournalFence, + assertJournalWritable, + JournalAppendBudget +} from './journal-write-guards' + +export { AgentSessionJournalError } from './journal-write-guards' + +export type AgentSessionJournalOptions = { + identity: AgentSessionJournalIdentity + journalDir: string + limits?: JournalPayloadLimits + compaction?: JournalCompactionPolicy + now?: () => number + mintEpoch?: () => string +} + +export type JournalReadSince = + | { ok: true; rows: JournalRow[]; cursor: AgentJournalCursor } + | { ok: false; reset: AgentJournalResetReason } + +export type ResolveDispatchInput = { + clientMessageId: string + fence: number + /** Set when crash reconciliation, not the live dispatch, settled this. */ + recovered?: true +} & ( + | { state: 'accepted'; providerIdentity: AgentJournalItemIdentity } + | { state: 'rejected' | 'unknown'; reason?: string | null } +) + +export type JournalAppendResult = { + cursor: AgentJournalCursor + itemId: string + revision: number +} + +export async function openAgentSessionJournal( + options: AgentSessionJournalOptions +): Promise { + const journal = new AgentSessionJournal(options) + await journal.open() + return journal +} + +export class AgentSessionJournal { + private readonly identity: AgentSessionJournalIdentity + private readonly journalDir: string + private readonly budget: JournalAppendBudget + private readonly compaction: JournalCompactionPolicy | undefined + private readonly now: () => number + private readonly mintEpoch: () => string + + private state: JournalReducerState + private tailRows: JournalRow[] = [] + private compactedThrough = 0 + private sizeBytes = 0 + private readOnly = false + /** Serializes sequence assignment with the durable write behind it. */ + private writes: Promise = Promise.resolve() + + constructor(options: AgentSessionJournalOptions) { + this.identity = options.identity + this.journalDir = options.journalDir + this.budget = new JournalAppendBudget( + options.identity.sessionId, + options.limits ?? DEFAULT_JOURNAL_PAYLOAD_LIMITS + ) + this.compaction = options.compaction + this.now = options.now ?? (() => Date.now()) + this.mintEpoch = options.mintEpoch ?? randomUUID + this.state = createJournalReducerState(options.identity.sessionId, '') + } + + get isReadOnly(): boolean { + return this.readOnly + } + + get epoch(): string { + return this.state.epoch + } + + get directory(): string { + return this.journalDir + } + + /** Highest sequence folded into the snapshot; rows at or below it are no + * longer individually replayable. */ + get compactionBoundary(): number { + return this.compactedThrough + } + + async open(): Promise { + await ensureJournalDir(this.journalDir) + const loaded = await loadJournal(this.journalDir, this.identity.sessionId) + if (!loaded) { + await this.startEpoch('session_created', 0) + return + } + this.state = loaded.state + this.tailRows = loaded.tailRows + this.compactedThrough = loaded.compactedThrough + this.sizeBytes = loaded.sizeBytes + this.readOnly = loaded.readOnly + if (loaded.corrupt && !loaded.readOnly) { + // A reader that observes a gap rolls the epoch rather than rendering a + // partial timeline; clients take the snapshot reload they already handle. + await this.rollEpoch('corruption', this.state.highestFence) + } + } + + cursor(): AgentJournalCursor { + return { epoch: this.state.epoch, sequence: this.state.lastSequence } + } + + snapshot(): AgentJournalSnapshot { + return renderJournalState(this.state) + } + + submissions(): AgentJournalSubmission[] { + return [...this.state.submissions.values()] + } + + pendingSubmissions(): AgentJournalSubmission[] { + return this.submissions().filter((entry) => entry.dispatchState === 'pending') + } + + /** The durable answer to "did my send land?" — a reconnecting client asking + * again gets this instead of re-sending. */ + receiptFor(clientMessageId: string): AgentJournalAcceptanceReceipt | null { + return this.state.receipts.get(clientMessageId) ?? null + } + + readSince(cursor: AgentJournalCursor): JournalReadSince { + const resume = this.resume(cursor) + if (!resume.ok) { + return { ok: false, reset: resume.reset } + } + return { + ok: true, + rows: this.tailRows.filter((row) => row.seq > resume.afterSequence), + cursor: this.cursor() + } + } + + private resume(cursor: AgentJournalCursor): JournalResume { + if (this.readOnly) { + return { ok: false, reset: 'schema_unreadable' } + } + return resolveJournalResume( + { + epoch: this.state.epoch, + lastSequence: this.state.lastSequence, + oldestSequence: this.state.oldestSequence + }, + cursor + ) + } + + /** Upsert by stable identity. The revision is assigned here so a caller + * cannot accidentally publish a revision the reducer will drop. */ + appendItem( + identity: AgentJournalItemIdentity, + body: AgentJournalItemBody, + options: { fence: number; observedAt?: number; recovered?: true } = { fence: 0 } + ): Promise { + const itemId = agentJournalItemKey(identity) + return this.enqueue((seq, ts) => { + const resolved = this.state.aliases.get(itemId) ?? itemId + const revision = (this.state.items.get(resolved)?.revision ?? 0) + 1 + return { + kind: 'item', + itemId, + revision, + body, + ...this.rowBase(seq, options.fence, options.observedAt ?? ts), + ...(options.recovered ? { recovered: options.recovered } : {}) + } + }).then((row) => ({ + cursor: { epoch: row.epoch, sequence: row.seq }, + itemId, + revision: (row as Extract).revision + })) + } + + appendTombstone( + identity: AgentJournalItemIdentity, + options: { fence: number } + ): Promise { + const itemId = agentJournalItemKey(identity) + return this.enqueue((seq, ts) => { + const resolved = this.state.aliases.get(itemId) ?? itemId + const revision = (this.state.items.get(resolved)?.revision ?? 0) + 1 + return { kind: 'tombstone', itemId, revision, ...this.rowBase(seq, options.fence, ts) } + }).then((row) => ({ epoch: row.epoch, sequence: row.seq })) + } + + /** + * Write-ahead submission row. It is durable before the caller dispatches + * anything, and it doubles as the optimistic user bubble so an accepted echo + * reconciles into an existing slot instead of appending a second copy. + */ + appendSubmission(input: { + clientMessageId: string + payloadFingerprint: string + body: AgentJournalMessageItem + fence: number + }): Promise { + return this.enqueue((seq, ts) => ({ + kind: 'submission', + clientMessageId: input.clientMessageId, + payloadFingerprint: input.payloadFingerprint, + providerHandle: this.identity.providerHandle, + body: input.body, + ...this.rowBase(seq, input.fence, ts) + })).then((row) => ({ epoch: row.epoch, sequence: row.seq })) + } + + /** + * Advance a submission to exactly one of accepted / rejected / unknown. + * + * Accepting REQUIRES the provider identity rather than a free-form id: the + * adopted key is what the provider's echo will upsert into, so a mismatched + * string here would silently give the user a second copy of their own message. + */ + resolveDispatch(input: ResolveDispatchInput): Promise { + const providerItemId = + input.state === 'accepted' ? agentJournalItemKey(input.providerIdentity) : null + return this.enqueue((seq, ts) => ({ + kind: 'dispatch', + clientMessageId: input.clientMessageId, + state: input.state, + providerItemId, + reason: input.state === 'accepted' ? null : (input.reason ?? null), + ...this.rowBase(seq, input.fence, ts), + ...(input.recovered ? { recovered: input.recovered } : {}) + })).then((row) => ({ epoch: row.epoch, sequence: row.seq })) + } + + /** On restart every `pending` submission becomes `unknown` before the session + * accepts a writer. Orca never re-sends on the user's behalf. */ + async markPendingSubmissionsUnknown(fence: number): Promise { + const pending = this.pendingSubmissions().map((entry) => entry.clientMessageId) + for (const clientMessageId of pending) { + await this.resolveDispatch({ + clientMessageId, + state: 'unknown', + reason: 'host_restarted_before_acknowledgement', + fence, + recovered: true + }) + } + return pending + } + + async compact(): Promise { + assertJournalWritable(this.readOnly, this.identity.sessionId) + const result = await compactJournal({ + journalDir: this.journalDir, + state: this.state, + tailRows: this.tailRows, + policy: this.compaction, + now: this.now() + }) + this.tailRows = result.tailRows + this.compactedThrough = result.compactedThrough + this.state.oldestSequence = result.oldestSequence + this.sizeBytes = this.tailRows.reduce((total, row) => total + journalRowByteLength(row), 0) + } + + /** The escape hatch for corruption, an unreconcilable prefix, a forked handle, + * and an unreadable schema. It invalidates every cursor; clients reload. */ + async rollEpoch(reason: AgentJournalEpochReason, fence: number): Promise { + assertJournalWritable(this.readOnly, this.identity.sessionId) + await this.startEpoch(reason, fence) + return this.cursor() + } + + private async startEpoch(reason: AgentJournalEpochReason, fence: number): Promise { + const published = await publishNewEpoch({ + journalDir: this.journalDir, + sessionId: this.identity.sessionId, + providerHandle: this.identity.providerHandle, + epoch: this.mintEpoch(), + reason, + fence, + now: this.now() + }) + this.state = published.state + this.tailRows = [published.row] + this.compactedThrough = 0 + this.sizeBytes = journalRowByteLength(published.row) + } + + private rowBase( + seq: number, + fence: number, + ts: number + ): { v: number; epoch: string; seq: number; fence: number; ts: number } { + return { + v: AGENT_SESSION_JOURNAL_SCHEMA_VERSION, + epoch: this.state.epoch, + seq, + fence, + ts + } + } + + /** + * Assign the next sequence, make the row durable, and fold it through the + * SAME reducer replay uses — all inside one serialized step, so concurrent + * callers cannot interleave and mint the same sequence. + */ + private enqueue(build: (seq: number, ts: number) => JournalRow): Promise { + const run = this.writes.then(async () => { + assertJournalWritable(this.readOnly, this.identity.sessionId) + const ts = this.now() + const row = build(this.state.lastSequence + 1, ts) + assertJournalFence(row.fence, this.state.highestFence) + this.budget.assert(row, ts, this.sizeBytes) + await appendJournalRows(this.journalDir, [row]) + applyJournalRow(this.state, row) + this.tailRows.push(row) + this.sizeBytes += journalRowByteLength(row) + return row + }) + this.writes = run.catch(() => undefined) + return run + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts new file mode 100644 index 00000000000..50a47d02bd1 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-submission-reconciler.ts @@ -0,0 +1,191 @@ +// Restart reconciliation for the crash boundary. +// +// A submission row is durable before dispatch, so after a crash the host knows +// what it TRIED to send but not whether the provider took it. Every surviving +// `pending` becomes `unknown` and is then matched against provider history. +// +// Matching is by identity only — an echoed client message id, else a provider +// item id the journal already adopted, else the payload fingerprint when it +// picks out exactly one unclaimed item. Never by text equality: the same +// question asked twice is two messages, and collapsing them silently loses one. +// +// Orca never re-sends on the user's behalf. An unresolved submission stays +// `unknown` — a displayed state meaning "delivery unconfirmed", neither sent nor +// failed — and the user chooses to resend or discard. + +import type { + AgentJournalItemIdentity, + AgentJournalSubmission +} from '../../../shared/agent-session-journal-types' +import { agentJournalItemKey } from '../../../shared/agent-session-journal-item-key' + +export type ProviderHistoryItem = { + /** The provider's own id for this item. Used to claim it at most once; the + * journal key comes from `identity`, because a provider id is not stable. */ + providerItemId: string + /** The client message id when the provider echoes one (Codex carries it on + * user messages); null for providers that drop it. */ + clientMessageId: string | null + /** Fingerprint of the submitted payload, when the caller can compute one from + * provider content. Used only to break an otherwise unique tie. */ + payloadFingerprint: string | null + identity: AgentJournalItemIdentity +} + +export type ProviderHistoryWindow = { + /** Provider items observed at or after the journal's last committed item. */ + items: readonly ProviderHistoryItem[] + /** + * The history read actually started at the journal's last committed item. A + * fork, a compacted provider log, or a truncated read makes absence + * meaningless, so a missing submission cannot be called "not delivered". + */ + boundaryConsistent: boolean + /** The provider reports a turn still running: absence proves nothing yet. */ + turnInFlight: boolean +} + +export type SubmissionReconciliation = + | { + clientMessageId: string + outcome: 'accepted' + providerItemId: string + identity: AgentJournalItemIdentity + } + | { clientMessageId: string; outcome: 'rejected'; reason: SubmissionRejectionReason } + | { clientMessageId: string; outcome: 'unknown'; reason: SubmissionUnknownReason } + +export type SubmissionRejectionReason = 'not_delivered' + +export type SubmissionUnknownReason = + | 'history_boundary_inconsistent' + | 'turn_in_flight' + | 'ambiguous_match' + +/** + * Resolve every unsettled submission against provider history. + * + * Passes run strongest-first across ALL submissions before the next begins, so + * a weak fingerprint tie can never claim an item that an echoed client message + * id would have matched exactly. Each history item is claimable once. + */ +export function reconcileSubmissions(input: { + submissions: readonly AgentJournalSubmission[] + history: ProviderHistoryWindow +}): SubmissionReconciliation[] { + const unsettled = input.submissions.filter( + (submission) => submission.dispatchState === 'pending' || submission.dispatchState === 'unknown' + ) + const claimed = new Set() + const matched = new Map() + const ambiguous = new Set() + + claimBy( + unsettled, + input.history.items, + claimed, + matched, + (submission, item) => + // A submission that already adopted a key re-matches on that key, not on the + // provider's raw id — the raw id renumbers, the identity-derived key does not. + Boolean(submission.providerItemId) && + submission.providerItemId === agentJournalItemKey(item.identity) + ) + claimBy( + unsettled, + input.history.items, + claimed, + matched, + (submission, item) => + Boolean(item.clientMessageId) && item.clientMessageId === submission.clientMessageId + ) + + for (const submission of unsettled) { + if (matched.has(submission.clientMessageId)) { + continue + } + const candidates = input.history.items.filter( + (item) => + !claimed.has(item.providerItemId) && + Boolean(item.payloadFingerprint) && + item.payloadFingerprint === submission.payloadFingerprint + ) + const only = candidates.length === 1 ? candidates[0] : undefined + if (only) { + claimed.add(only.providerItemId) + matched.set(submission.clientMessageId, only) + } else if (candidates.length > 1) { + // Two identical payloads and no id to tell them apart: guessing would + // either duplicate the user's message or drop one of them. + ambiguous.add(submission.clientMessageId) + } + } + + return unsettled.map((submission) => resolveOne(submission, matched, ambiguous, input.history)) +} + +function claimBy( + submissions: readonly AgentJournalSubmission[], + items: readonly ProviderHistoryItem[], + claimed: Set, + matched: Map, + matches: (submission: AgentJournalSubmission, item: ProviderHistoryItem) => boolean +): void { + for (const submission of submissions) { + if (matched.has(submission.clientMessageId)) { + continue + } + const item = items.find((candidate) => { + return !claimed.has(candidate.providerItemId) && matches(submission, candidate) + }) + if (item) { + claimed.add(item.providerItemId) + matched.set(submission.clientMessageId, item) + } + } +} + +function resolveOne( + submission: AgentJournalSubmission, + matched: Map, + ambiguous: Set, + history: ProviderHistoryWindow +): SubmissionReconciliation { + const item = matched.get(submission.clientMessageId) + if (item) { + return { + clientMessageId: submission.clientMessageId, + outcome: 'accepted', + providerItemId: item.providerItemId, + identity: item.identity + } + } + if (ambiguous.has(submission.clientMessageId)) { + return { + clientMessageId: submission.clientMessageId, + outcome: 'unknown', + reason: 'ambiguous_match' + } + } + if (!history.boundaryConsistent) { + return { + clientMessageId: submission.clientMessageId, + outcome: 'unknown', + reason: 'history_boundary_inconsistent' + } + } + if (history.turnInFlight) { + return { + clientMessageId: submission.clientMessageId, + outcome: 'unknown', + reason: 'turn_in_flight' + } + } + // Absent from a history we can trust the boundary of, with nothing running: + // the provider never took it. + return { + clientMessageId: submission.clientMessageId, + outcome: 'rejected', + reason: 'not_delivered' + } +} diff --git a/src/main/native-chat/agent-session-journal/journal-write-guards.ts b/src/main/native-chat/agent-session-journal/journal-write-guards.ts new file mode 100644 index 00000000000..eabdc9d750c --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-write-guards.ts @@ -0,0 +1,74 @@ +// Guards an append clears before it becomes durable. +// +// All four refuse loudly rather than degrade: a silent drop here is a message +// missing from the transcript with nothing to explain it. + +import type { JournalPayloadLimits } from './journal-payload-bounds' +import { journalRowByteLength, type JournalRow } from './journal-row-schema' + +export class AgentSessionJournalError extends Error { + constructor( + readonly code: + | 'journal_read_only' + | 'journal_stale_fence' + | 'journal_bound_exceeded' + | 'journal_rate_exceeded', + message: string + ) { + super(message) + this.name = 'AgentSessionJournalError' + } +} + +/** A journal written by a newer schema is readable but never writable: this + * host cannot represent rows it does not understand. */ +export function assertJournalWritable(readOnly: boolean, sessionId: string): void { + if (readOnly) { + throw new AgentSessionJournalError( + 'journal_read_only', + `agent-session journal for ${sessionId} uses a newer schema; this host is read-only` + ) + } +} + +/** A write from a superseded owner is rejected outright — merging it would let + * two writers share one sequence space. */ +export function assertJournalFence(fence: number, highestFence: number): void { + if (fence < highestFence) { + throw new AgentSessionJournalError( + 'journal_stale_fence', + `fence ${fence} is behind the journal's ${highestFence}` + ) + } +} + +/** Total size and append rate for one session, bounding a runaway agent. */ +export class JournalAppendBudget { + private windowStart = 0 + private appendsInWindow = 0 + + constructor( + private readonly sessionId: string, + private readonly limits: JournalPayloadLimits + ) {} + + assert(row: JournalRow, ts: number, sizeBytes: number): void { + if (sizeBytes + journalRowByteLength(row) > this.limits.maxSessionBytes) { + throw new AgentSessionJournalError( + 'journal_bound_exceeded', + `agent-session journal for ${this.sessionId} reached its ${this.limits.maxSessionBytes}-byte bound` + ) + } + if (ts - this.windowStart >= this.limits.appendWindowMs) { + this.windowStart = ts + this.appendsInWindow = 0 + } + this.appendsInWindow += 1 + if (this.appendsInWindow > this.limits.maxAppendsPerWindow) { + throw new AgentSessionJournalError( + 'journal_rate_exceeded', + `agent-session journal for ${this.sessionId} exceeded ${this.limits.maxAppendsPerWindow} appends per ${this.limits.appendWindowMs}ms` + ) + } + } +} diff --git a/src/shared/agent-session-journal-item-key.ts b/src/shared/agent-session-journal-item-key.ts new file mode 100644 index 00000000000..45931ddc4ed --- /dev/null +++ b/src/shared/agent-session-journal-item-key.ts @@ -0,0 +1,51 @@ +// Item-identity → stable journal key. Pure and shared: the host keys upserts +// with it and clients reconcile optimistic sends against the same string. +// +// Components are percent-encoded before joining so a value containing the +// delimiter cannot collide with a different identity. + +import type { AgentJournalItemIdentity } from './agent-session-journal-types' + +const KEY_DELIMITER = ':' + +function encodePart(value: string | number): string { + return encodeURIComponent(String(value)) +} + +/** + * Stable string key for an item identity. + * + * Codex renumbers `item-N` ids on every resume, so its key is the thread, the + * turn, and the item's ordinal WITHIN that turn — a position that survives + * renumbering because a completed turn's item list does not change. `thread/fork` + * copies turns keeping their original turn ids, so the thread id must stay in the + * key. Claude copies item uuids on `--fork-session`, so its key is the session id + * plus the uuid. Text never participates. + */ +export function agentJournalItemKey(identity: AgentJournalItemIdentity): string { + if (identity.provider === 'codex') { + return [ + 'codex', + encodePart(identity.threadId), + encodePart(identity.turnId), + encodePart(identity.ordinal) + ].join(KEY_DELIMITER) + } + if (identity.provider === 'claude') { + return ['claude', encodePart(identity.sessionId), encodePart(identity.uuid)].join(KEY_DELIMITER) + } + if (identity.provider === 'orca') { + return ['orca', encodePart(identity.clientMessageId)].join(KEY_DELIMITER) + } + return [ + 'legacy', + encodePart(identity.agent), + encodePart(identity.sessionId), + encodePart(identity.recordId) + ].join(KEY_DELIMITER) +} + +/** Key for the pre-dispatch submission placeholder, before any provider echo. */ +export function agentJournalSubmissionKey(clientMessageId: string): string { + return agentJournalItemKey({ provider: 'orca', clientMessageId }) +} diff --git a/src/shared/agent-session-journal-types.ts b/src/shared/agent-session-journal-types.ts new file mode 100644 index 00000000000..24bf7e57878 --- /dev/null +++ b/src/shared/agent-session-journal-types.ts @@ -0,0 +1,204 @@ +// ─── Canonical agent-session journal: cross-process wire shapes ───────────── +// The host-owned timeline for a structured agent session. Everything here must +// be plain JSON: rows are persisted verbatim and later republished to clients, +// 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. + +import type { AgentType } from './agent-status-types' +import type { NativeChatBlock, NativeChatRole } from './native-chat-types' + +export { type AgentType } + +/** Bump only alongside a read-time upcaster in `journal-row-schema.ts`. */ +export const AGENT_SESSION_JOURNAL_SCHEMA_VERSION = 1 + +/** Epoch-qualified position in one journal. `sequence` 0 means "before the first row". */ +export type AgentJournalCursor = { + epoch: string + sequence: number +} + +/** The durable provider session a journal is bound to. + * Codex is one thread id; Claude needs the leaf because concurrent resumes of + * one session id branch the same transcript. */ +export type AgentSessionProviderHandle = + | { kind: 'codex'; threadId: string } + | { kind: 'claude'; sessionId: string; leafUuid: string | null } + | { kind: 'opaque'; agent: AgentType; value: string } + +/** The narrow slice of the durable session record the journal needs. The full + * record (owner, lease, account home) belongs to the session store. */ +export type AgentSessionJournalIdentity = { + /** Orca agent-session id — the journal's primary key. */ + sessionId: string + /** Execution-host workspace key. Identical for a worktree, a folder + * workspace, a WSL distro, and an SSH host; never a path. */ + workspaceId: string + /** Execution host that owns the process, so a client restart adjudicates nothing. */ + hostId: string + agent: AgentType + providerHandle: AgentSessionProviderHandle +} + +// ─── Item identity ────────────────────────────────────────────────────────── +// Reconciliation keys, settled by the provider spikes. Codex renumbers items +// positionally on resume, so a persisted item id is never an identity. Claude +// copies the original uuids on fork, so the uuid is. + +export type AgentJournalItemIdentity = + | { provider: 'codex'; threadId: string; turnId: string; ordinal: number } + | { provider: 'claude'; sessionId: string; uuid: string } + /** A submission Orca minted before any provider echo existed. */ + | { provider: 'orca'; clientMessageId: string } + /** Bridge-era transcript record with no provider-stable identity. */ + | { provider: 'legacy'; agent: AgentType; sessionId: string; recordId: string } + +// ─── Bounded payloads ─────────────────────────────────────────────────────── + +/** A tool output or diff body clipped to a head plus a content-addressed + * remainder. Crossing a bound sets `truncated`; it never silently drops. */ +export type AgentJournalBoundedPayload = { + head: string + /** Byte length of the ORIGINAL payload, not of `head`. */ + byteLength: number + /** sha256 of the original payload, and the blob store key when `truncated`. */ + digest: string + truncated: boolean +} + +// ─── Render-model items ───────────────────────────────────────────────────── + +export type AgentJournalMessageItem = { + kind: 'message' + role: NativeChatRole + blocks: NativeChatBlock[] +} + +export type AgentJournalToolCallState = 'running' | 'completed' | 'failed' + +export type AgentJournalToolCallItem = { + kind: 'tool-call' + name: string + input: unknown + state: AgentJournalToolCallState + output?: AgentJournalBoundedPayload +} + +export type AgentJournalDiffItem = { + kind: 'diff' + path: string + patch: AgentJournalBoundedPayload +} + +export const AGENT_JOURNAL_RESOLUTION_STATES = ['pending', 'resolved', 'cancelled'] as const +export type AgentJournalResolutionState = (typeof AGENT_JOURNAL_RESOLUTION_STATES)[number] + +/** Approvals and questions are durable items with explicit resolution state, so + * a second client answering one prompt loses the compare-and-set instead of + * invoking the provider callback twice. */ +export type AgentJournalResolution = { + state: AgentJournalResolutionState + /** Option id the winner picked; null while pending or cancelled. */ + selectedOptionId: string | null + /** Opaque client identity of the resolver, for "answered on ". */ + resolvedBy: string | null + resolvedAt: number | null +} + +export type AgentJournalPromptOption = { + id: string + label: string +} + +export type AgentJournalApprovalItem = { + kind: 'approval' + title: string + detail: string | null + options: AgentJournalPromptOption[] + resolution: AgentJournalResolution +} + +export type AgentJournalQuestionItem = { + kind: 'question' + question: string + options: AgentJournalPromptOption[] + resolution: AgentJournalResolution +} + +export type AgentJournalStatusItem = { + kind: 'status' + text: string +} + +export type AgentJournalItemBody = + | AgentJournalMessageItem + | AgentJournalToolCallItem + | AgentJournalDiffItem + | AgentJournalApprovalItem + | AgentJournalQuestionItem + | AgentJournalStatusItem + +/** One reduced timeline entry. `sequence` orders the list; `observedAt` is the + * provider's own clock and may sort earlier than a later sequence when the row + * was recovered after a crash. */ +export type AgentJournalRenderItem = { + itemId: string + revision: number + body: AgentJournalItemBody + sequence: number + observedAt: number + /** Set when the row was appended by crash reconciliation rather than live. */ + recovered?: true +} + +// ─── Submissions ──────────────────────────────────────────────────────────── + +export const AGENT_JOURNAL_DISPATCH_STATES = ['pending', 'accepted', 'rejected', 'unknown'] as const +export type AgentJournalDispatchState = (typeof AGENT_JOURNAL_DISPATCH_STATES)[number] + +/** The write-ahead submission row, projected. `unknown` is a displayed state: + * the turn reads as delivery unconfirmed, never as sent and never as failed. */ +export type AgentJournalSubmission = { + clientMessageId: string + fence: number + payloadFingerprint: string + dispatchState: AgentJournalDispatchState + /** Provider item identity adopted on accept; null otherwise. */ + providerItemId: string | null + /** Terminal reason on `rejected`. */ + reason: string | null + submittedAt: number + resolvedAt: number | null +} + +/** Durable answer to "did my send land?", keyed by client message id. Only an + * `accepted` dispatch mints one, and it outlives the journal tail. */ +export type AgentJournalAcceptanceReceipt = { + clientMessageId: string + providerItemId: string + cursor: AgentJournalCursor + acceptedAt: number +} + +// ─── Snapshots and cursor resume ──────────────────────────────────────────── + +export type AgentJournalSnapshot = { + sessionId: string + cursor: AgentJournalCursor + items: AgentJournalRenderItem[] + submissions: AgentJournalSubmission[] +} + +/** Why a cursor could not be resumed. Every value forces a clean snapshot + * reload on the client. */ +export const AGENT_JOURNAL_RESET_REASONS = [ + 'epoch_changed', + 'cursor_ahead', + 'cursor_compacted', + 'journal_gap', + 'schema_unreadable' +] as const +export type AgentJournalResetReason = (typeof AGENT_JOURNAL_RESET_REASONS)[number]