feat(native-chat): canonical agent-session journal with crash-safe receipts and epoch cursors (phase 1c)

Adds the host-side append-only journal that a native chat session will read and
write: one durable log per agent session, one reducer shared by live events and
replay, and epoch-qualified cursors so a reconnecting client either resumes
exactly or takes a clean snapshot reload.

Nothing is wired up yet. No RPC, no adapters, no UI: the journal is dormant
until a later phase feeds it, so behavior for existing sessions is unchanged.

- Append-only JSONL log plus an atomic snapshot that carries its own retained
  tail, so compaction is a single durable step and a crash mid-compaction leaves
  the log a recoverable superset.
- Sequences are assigned inside the same serialized append that makes the row
  durable, giving the (epoch, sequence) space exactly one writer with no gaps
  and no reuse. Every write carries the runtime fence; a stale fence is refused.
- A write-ahead submission row is durable before dispatch and doubles as the
  optimistic user bubble, so an accepted provider echo upserts into the bubble
  the user already sees instead of adding a second copy of their own message.
- Dispatch is a state machine: pending, accepted, rejected, or unknown. Unknown
  is a displayed state meaning "delivery unconfirmed"; Orca never re-sends on
  the user's behalf and never silently drops the message.
- Items upsert by stable provider identity with revisions and tombstones.
  Identity keying never uses text equality, so asking the same question twice
  stays two messages.
- Payload bounds keep a head plus the byte length and digest inline and push the
  remainder to a content-addressed blob store; crossing a bound is always
  marked, never a silent truncation.
- A row written by a newer schema puts the host into read-only mode rather than
  being skipped, so a newer client's rows are never dropped or compacted away.
- Legacy import hydrates a journal by running the existing transcript decoders
  unchanged and reading identity off the same raw lines they consume.
This commit is contained in:
Brennan Benson
2026-08-07 13:46:21 -07:00
parent 6382b8a053
commit d4727e1c4a
23 changed files with 3945 additions and 0 deletions
@@ -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<string> {
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<string | null> {
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<string>
): Promise<number> {
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<boolean> {
try {
await stat(path)
return true
} catch {
return false
}
}
@@ -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<JournalCompactionResult> {
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)
}
@@ -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> = {}
): 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([])
})
})
@@ -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 })
})
})
@@ -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
}
@@ -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 }
}
@@ -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' })
)
})
})
@@ -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<string, unknown> | null {
try {
const parsed: unknown = JSON.parse(line)
return parsed && typeof parsed === 'object' ? (parsed as Record<string, unknown>) : null
} catch {
return null
}
}
function stringField(source: unknown, key: string): string | null {
if (!source || typeof source !== 'object') {
return null
}
const value = (source as Record<string, unknown>)[key]
return typeof value === 'string' && value ? value : null
}
@@ -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<AgentSessionJournal> {
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<string> {
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 })
})
})
@@ -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<LegacyImportResult> {
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
}
@@ -0,0 +1,139 @@
// On-disk layout for one session's journal.
//
// <journal dir>/log.jsonl append-only rows, fsynced before the caller is told the write landed
// <journal dir>/snapshot.json folded state at a compaction boundary PLUS the retained tail
// <journal dir>/blobs/<sha256> 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<void> {
await mkdir(journalDir, { recursive: true })
}
export async function readJournalSnapshotFile(
journalDir: string
): Promise<JournalSnapshotFile | null> {
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<void> {
const target = join(journalDir, JOURNAL_SNAPSHOT_FILE)
await writeFileDurable(durableWriteTempPath(target), target, JSON.stringify(snapshot))
}
export async function readJournalLog(journalDir: string): Promise<JournalReadResult> {
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<void> {
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<void> {
const target = join(journalDir, JOURNAL_LOG_FILE)
const payload = rows.length ? `${rows.map(serializeJournalRow).join('\n')}\n` : ''
await writeFileDurable(durableWriteTempPath(target), target, payload)
}
@@ -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<JournalLoad | null> {
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<number, JournalRow>()
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)
}
@@ -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)
}
/** `<root>/agent-session-journal/<workspace>/<session>`. */
export function journalDirectoryFor(
root: string,
identity: Pick<AgentSessionJournalIdentity, 'workspaceId' | 'sessionId'>
): 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<string> {
const { app } = await import('electron')
return app.getPath('userData')
}
@@ -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')
}
@@ -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)
})
})
@@ -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<string, AgentJournalRenderItem>
/** Revision of a removed item, so a late lower revision cannot resurrect it. */
tombstones: Map<string, number>
submissions: Map<string, AgentJournalSubmission>
receipts: Map<string, AgentJournalAcceptanceReceipt>
/** 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<string, string>
}
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<JournalRow, { kind: 'submission' }>
): 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<JournalRow, { kind: 'dispatch' }>
): 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<string> {
const digests = new Set<string>()
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
}
@@ -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<AgentJournalDispatchState, 'pending'>
/** 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<string, unknown>
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<string, unknown>, version: number): Record<string, unknown> {
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<string, unknown>): 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
}
@@ -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<Parameters<typeof openAgentSessionJournal>[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"')
})
})
@@ -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<AgentSessionJournal> {
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<unknown> = 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<void> {
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<JournalAppendResult> {
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<JournalRow, { kind: 'item' }>).revision
}))
}
appendTombstone(
identity: AgentJournalItemIdentity,
options: { fence: number }
): Promise<AgentJournalCursor> {
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<AgentJournalCursor> {
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<AgentJournalCursor> {
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<string[]> {
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<void> {
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<AgentJournalCursor> {
assertJournalWritable(this.readOnly, this.identity.sessionId)
await this.startEpoch(reason, fence)
return this.cursor()
}
private async startEpoch(reason: AgentJournalEpochReason, fence: number): Promise<void> {
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<JournalRow> {
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
}
}
@@ -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<string>()
const matched = new Map<string, ProviderHistoryItem>()
const ambiguous = new Set<string>()
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<string>,
matched: Map<string, ProviderHistoryItem>,
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<string, ProviderHistoryItem>,
ambiguous: Set<string>,
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'
}
}
@@ -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`
)
}
}
}
@@ -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 })
}
+204
View File
@@ -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 <device>". */
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]