diff --git a/src/main/codex/codex-structured-item-translation.test.ts b/src/main/codex/codex-structured-item-translation.test.ts index 44266204405..8895facc11a 100644 --- a/src/main/codex/codex-structured-item-translation.test.ts +++ b/src/main/codex/codex-structured-item-translation.test.ts @@ -53,6 +53,23 @@ function keysFor(items: CodexThreadItem[]): string[] { ) } +describe('codex turn ordinals', () => { + it('releases a forgotten turn without ever reusing an ordinal it assigned', () => { + const ordinals = new CodexTurnOrdinals() + expect(ordinals.ordinalFor('thread-1', 'turn-1', 'item-1')).toBe(0) + expect(ordinals.ordinalFor('thread-1', 'turn-1', 'item-2')).toBe(1) + + ordinals.forgetTurn('thread-1', 'turn-1') + + // A straggler for the released turn — even a previously seen item id — gets + // a FRESH ordinal: reusing a released slot would upsert another item's row. + expect(ordinals.ordinalFor('thread-1', 'turn-1', 'item-1')).toBe(2) + expect(ordinals.ordinalFor('thread-1', 'turn-1', 'item-3')).toBe(3) + // Other turns are untouched. + expect(ordinals.ordinalFor('thread-1', 'turn-2', 'item-1')).toBe(0) + }) +}) + describe('codex item identity', () => { it('gives a resumed turn the same message keys as the live turn it renumbered', () => { expect(keysFor(LIVE_TURN)).toEqual(keysFor(RESUMED_TURN)) diff --git a/src/main/codex/codex-structured-item-translation.ts b/src/main/codex/codex-structured-item-translation.ts index 04f22c16b06..9269c952eb4 100644 --- a/src/main/codex/codex-structured-item-translation.ts +++ b/src/main/codex/codex-structured-item-translation.ts @@ -54,23 +54,33 @@ export function readCodexThreadItem(value: unknown): CodexThreadItem | null { * than no key, because it would look reconcilable and reconcile wrongly. */ export class CodexTurnOrdinals { - private readonly turns = new Map>() + private readonly turns = new Map; next: number }>() ordinalFor(threadId: string, turnId: string, codexItemId: string): number { const turnKey = `${encodeURIComponent(threadId)}:${encodeURIComponent(turnId)}` - const assigned = this.turns.get(turnKey) ?? new Map() - this.turns.set(turnKey, assigned) - const existing = assigned.get(codexItemId) + let turn = this.turns.get(turnKey) + if (!turn) { + turn = { assigned: new Map(), next: 0 } + this.turns.set(turnKey, turn) + } + const existing = turn.assigned.get(codexItemId) if (existing !== undefined) { return existing } - const ordinal = assigned.size - assigned.set(codexItemId, ordinal) + const ordinal = turn.next + turn.assigned.set(codexItemId, ordinal) + turn.next += 1 return ordinal } - forgetTurn(turnId: string): void { - this.turns.delete(turnId) + /** Releases a finished turn's per-item map while keeping its counter, so a + * straggler frame can never be assigned an ordinal the turn already used — + * a reused slot would upsert another item's journal row. */ + forgetTurn(threadId: string, turnId: string): void { + const turn = this.turns.get(`${encodeURIComponent(threadId)}:${encodeURIComponent(turnId)}`) + if (turn) { + turn.assigned = new Map() + } } } diff --git a/src/main/codex/codex-structured-journal-translation.test.ts b/src/main/codex/codex-structured-journal-translation.test.ts index cd0e626887a..0f1331fbcb4 100644 --- a/src/main/codex/codex-structured-journal-translation.test.ts +++ b/src/main/codex/codex-structured-journal-translation.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from 'vitest' +import { describe, expect, it, vi } from 'vitest' import type { AgentJournalItemBody, AgentJournalItemIdentity @@ -6,6 +6,7 @@ import type { import { agentJournalItemKey } from '../../shared/agent-session-journal-item-key' import { projectStructuredItemsToNativeChat } from '../../shared/structured-agent-session-projection' import type { StructuredAgentSessionEventSink } from '../native-chat/agent-session-wire/structured-agent-session-event-sink' +import { CodexTurnOrdinals } from './codex-structured-item-translation' import { createCodexJournalTranslator, MAX_CODEX_GENERIC_ROWS_PER_TURN @@ -462,6 +463,18 @@ describe('codex journal translation', () => { expect(tap.publishes()).toBe(1) }) + it('releases a turn ordinal map when the turn completes', () => { + const spy = vi.spyOn(CodexTurnOrdinals.prototype, 'forgetTurn') + try { + const { translator } = translatorWith() + translator.handle(TURN_STARTED) + translator.handle(notification('turn/completed', { turn: { id: TURN_ID } })) + expect(spy).toHaveBeenCalledWith(THREAD_ID, TURN_ID) + } finally { + spy.mockRestore() + } + }) + it('journals malformed item events but never malformed deltas', () => { const { translator, tap, window } = translatorWith() diff --git a/src/main/codex/codex-structured-journal-translation.ts b/src/main/codex/codex-structured-journal-translation.ts index 4de04769fdb..5b5c99aa11e 100644 --- a/src/main/codex/codex-structured-journal-translation.ts +++ b/src/main/codex/codex-structured-journal-translation.ts @@ -270,6 +270,7 @@ export function createCodexJournalTranslator( const turnId = readCodexTurnId(event.params) ?? currentTurnIds.get(event.threadId) if (turnId) { publishTurnLifecycle(event.sessionId, event.threadId, turnId, 'completed') + ordinals.forgetTurn(event.threadId, turnId) } // A later item with no turn of its own belongs to no turn, not to the // one that just ended. diff --git a/src/main/codex/codex-tui-rollout-proof.test.ts b/src/main/codex/codex-tui-rollout-proof.test.ts index 0ec6943d4db..843c6701df4 100644 --- a/src/main/codex/codex-tui-rollout-proof.test.ts +++ b/src/main/codex/codex-tui-rollout-proof.test.ts @@ -1,3 +1,6 @@ +import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' import { describe, expect, it, vi } from 'vitest' import { codexTuiStatusProbeInput, @@ -40,6 +43,32 @@ describe('Codex TUI rollout proof', () => { expect(readSessionMetaId).toHaveBeenCalledTimes(1) }) + it('skips a rollout file that vanishes mid-scan instead of aborting the proof', async () => { + const root = await mkdtemp(join(tmpdir(), 'orca-rollout-proof-')) + try { + const day = join(root, 'sessions', '2026', '08', '11') + await mkdir(day, { recursive: true }) + const real = join(day, `rollout-now-${THREAD}.jsonl`) + await writeFile( + real, + `${JSON.stringify({ type: 'session_meta', payload: { id: THREAD } })}\n` + ) + // Listed but already deleted by the time the scan reads it — Codex prunes + // and rewrites rollout files while the scan runs. + const vanished = join(day, `rollout-gone-${THREAD}.jsonl`) + const files = async function* (): AsyncGenerator { + yield vanished + yield real + } + + await expect( + resolvePinnedCodexRolloutProof(root, THREAD, { listFiles: files }) + ).resolves.toBe(real) + } finally { + await rm(root, { recursive: true, force: true }) + } + }) + it('rejects a rollout whose session_meta names another thread', async () => { const files = async function* (): AsyncGenerator { yield `/pinned/sessions/2026/08/11/rollout-now-${THREAD}.jsonl` diff --git a/src/main/codex/codex-tui-rollout-proof.ts b/src/main/codex/codex-tui-rollout-proof.ts index 190d33a2ebe..96bfd6b7695 100644 --- a/src/main/codex/codex-tui-rollout-proof.ts +++ b/src/main/codex/codex-tui-rollout-proof.ts @@ -118,7 +118,14 @@ export async function proveCodexTuiRollout(input: { } async function readCodexRolloutSessionMetaId(filePath: string): Promise { - const file = await open(filePath, 'r') + // A listed rollout may vanish before it is read — Codex prunes and rewrites + // these files. One missing file must not abort the whole scan. + let file: Awaited> + try { + file = await open(filePath, 'r') + } catch { + return null + } try { const buffer = Buffer.alloc(ROLLOUT_READ_LIMIT) const { bytesRead } = await file.read(buffer, 0, buffer.length, 0) diff --git a/src/main/native-chat/agent-session-journal/journal-compaction.ts b/src/main/native-chat/agent-session-journal/journal-compaction.ts index 2d535e684cc..9bc5dd21941 100644 --- a/src/main/native-chat/agent-session-journal/journal-compaction.ts +++ b/src/main/native-chat/agent-session-journal/journal-compaction.ts @@ -113,3 +113,17 @@ function retainTail( const start = byAge === -1 ? byCount : Math.min(byAge, byCount) return rows.slice(start) } + +/** Only when the retention window would actually drop rows: inside it, + * compaction rewrites an identical log, and doing that per append is a full + * state serialization on the hot path. */ +export function journalTailIsReadyToCompact( + tailRows: readonly JournalRow[], + policy: JournalCompactionPolicy, + now: number +): boolean { + if (tailRows.length <= policy.minTailRows * 2) { + return false + } + return (tailRows[0]?.ts ?? now) < now - policy.retainTailMs +} diff --git a/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts b/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts index 3d9c0c6528e..4b4e7f64968 100644 --- a/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts +++ b/src/main/native-chat/agent-session-journal/journal-corruption-quarantine.ts @@ -38,3 +38,20 @@ export async function quarantineUnreadableSchema(journalDir: string): Promise AGENT_SESSION_JOURNAL_SCHEMA_VERSION, corrupt, + malformedRows: log.malformed, sizeBytes: tailRows.reduce((total, row) => total + journalRowByteLength(row), 0), quarantineRemainder } @@ -101,6 +105,7 @@ function emptyReadOnlyLoad(sessionId: string): JournalLoad { compactedThrough: 0, readOnly: true, corrupt: false, + malformedRows: 0, sizeBytes: 0 } } diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts new file mode 100644 index 00000000000..b2a3468bc90 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.test.ts @@ -0,0 +1,122 @@ +import { describe, expect, it } from 'vitest' +import { parseJournalRow } from './journal-row-schema' + +const BASE = { v: 1, epoch: 'epoch-1', seq: 1, fence: 1, ts: 1 } + +function parse(row: Record): boolean { + return parseJournalRow(JSON.stringify(row)).ok +} + +describe('journal row validation', () => { + it('accepts every fully-formed row shape this build writes', () => { + expect( + parse({ + ...BASE, + kind: 'epoch', + reason: 'session_created', + providerHandle: { kind: 'codex', threadId: 't' } + }) + ).toBe(true) + expect( + parse({ + ...BASE, + kind: 'item', + itemId: 'i-1', + revision: 1, + body: { kind: 'status', text: 'x' } + }) + ).toBe(true) + expect(parse({ ...BASE, kind: 'tombstone', itemId: 'i-1', revision: 2 })).toBe(true) + expect( + parse({ + ...BASE, + kind: 'submission', + clientMessageId: 'm-1', + payloadFingerprint: 'a'.repeat(64), + providerHandle: { kind: 'codex', threadId: 't' }, + body: { kind: 'message', role: 'user', blocks: [] } + }) + ).toBe(true) + expect( + parse({ + ...BASE, + kind: 'dispatch', + clientMessageId: 'm-1', + state: 'accepted', + providerItemId: 'codex:t:turn:0', + reason: null + }) + ).toBe(true) + }) + + it('rejects a dispatch row missing its state or with mistyped fields', () => { + expect(parse({ ...BASE, kind: 'dispatch', clientMessageId: 'm-1' })).toBe(false) + expect( + parse({ + ...BASE, + kind: 'dispatch', + clientMessageId: 'm-1', + state: 7, + providerItemId: null, + reason: null + }) + ).toBe(false) + expect( + parse({ + ...BASE, + kind: 'dispatch', + clientMessageId: 'm-1', + state: 'accepted', + providerItemId: 7, + reason: null + }) + ).toBe(false) + expect( + parse({ + ...BASE, + kind: 'dispatch', + clientMessageId: 'm-1', + state: 'rejected', + providerItemId: null, + reason: 7 + }) + ).toBe(false) + }) + + it('rejects a submission row without its fingerprint, handle, or message body', () => { + const submission = { + ...BASE, + kind: 'submission', + clientMessageId: 'm-1', + payloadFingerprint: 'a'.repeat(64), + providerHandle: { kind: 'codex', threadId: 't' }, + body: { kind: 'message', role: 'user', blocks: [] } + } + expect(parse({ ...submission, payloadFingerprint: undefined as never })).toBe(false) + expect(parse({ ...submission, providerHandle: 'codex' })).toBe(false) + expect(parse({ ...submission, body: 'hi' })).toBe(false) + }) + + it('rejects an item row whose body is not a kinded object', () => { + expect(parse({ ...BASE, kind: 'item', itemId: 'i-1', revision: 1 })).toBe(false) + expect(parse({ ...BASE, kind: 'item', itemId: 'i-1', revision: 1, body: 'text' })).toBe(false) + expect(parse({ ...BASE, kind: 'item', itemId: 'i-1', revision: 1, body: {} })).toBe(false) + }) + + it('rejects an epoch row without a provider handle', () => { + expect(parse({ ...BASE, kind: 'epoch', reason: 'session_created' })).toBe(false) + }) + + it('keeps forward compatibility for new dispatch states without a version bump', () => { + expect( + parse({ + ...BASE, + kind: 'dispatch', + clientMessageId: 'm-1', + state: 'some-future-state', + providerItemId: null, + reason: null + }) + ).toBe(true) + }) +}) diff --git a/src/main/native-chat/agent-session-journal/journal-row-schema.ts b/src/main/native-chat/agent-session-journal/journal-row-schema.ts index 6f89d9e6395..a90c32d18cc 100644 --- a/src/main/native-chat/agent-session-journal/journal-row-schema.ts +++ b/src/main/native-chat/agent-session-journal/journal-row-schema.ts @@ -134,6 +134,18 @@ function upcastRow(record: Record, version: number): Record { + return typeof value === 'object' && value !== null && !Array.isArray(value) +} + +/** Every render body is a kinded object; anything else would poison the reducer. */ +function isKindedBody(value: unknown): boolean { + return isPlainObject(value) && typeof value.kind === 'string' && value.kind.length > 0 +} + +/** Field values are type-checked, never enum-checked: a future build adding a + * dispatch state or handle kind must bump the row version, but this build + * should not misread a same-version row as malformed over a wider enum. */ function isJournalRow(record: Record): record is JournalRow { if (typeof record.kind !== 'string' || !ROW_KINDS.has(record.kind)) { return false @@ -148,13 +160,36 @@ function isJournalRow(record: Record): record is JournalRow { ) { return false } - if (record.kind === 'item' || record.kind === 'tombstone') { + if (record.kind === 'item') { + return ( + typeof record.itemId === 'string' && + Number.isInteger(record.revision) && + isKindedBody(record.body) + ) + } + if (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 + if (record.kind === 'submission') { + return ( + typeof record.clientMessageId === 'string' && + record.clientMessageId.length > 0 && + typeof record.payloadFingerprint === 'string' && + isPlainObject(record.providerHandle) && + isKindedBody(record.body) + ) } - return typeof record.reason === 'string' + if (record.kind === 'dispatch') { + return ( + typeof record.clientMessageId === 'string' && + record.clientMessageId.length > 0 && + typeof record.state === 'string' && + record.state.length > 0 && + (record.providerItemId === null || typeof record.providerItemId === 'string') && + (record.reason === null || typeof record.reason === 'string') + ) + } + return typeof record.reason === 'string' && isPlainObject(record.providerHandle) } /** Approximate on-disk cost of a row, used for the per-session size bound. */ diff --git a/src/main/native-chat/agent-session-journal/journal-store.test.ts b/src/main/native-chat/agent-session-journal/journal-store.test.ts index 516cc986635..988820bd36d 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.test.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.test.ts @@ -384,7 +384,7 @@ describe('schema', () => { expect(await readFile(logPath, 'utf-8')).toContain('"v":99') }) - it('skips a malformed line without giving up the journal', async () => { + it('skips a malformed line without giving up the journal, and discloses the skip', async () => { const journal = await open() await journal.appendItem(item(0), body('a'), { fence: 1 }) const logPath = join(root, JOURNAL_LOG_FILE) @@ -392,7 +392,32 @@ describe('schema', () => { const reopened = await open() expect(reopened.isReadOnly).toBe(false) - expect(reopened.snapshot().items).toHaveLength(1) + const items = reopened.snapshot().items + // The surviving row is untouched… + expect(items.some((entry) => entry.body.kind === 'message')).toBe(true) + // …and the skip is visible in the timeline instead of silently swallowed. + expect( + items.some( + (entry) => entry.body.kind === 'status' && entry.body.text.includes('could not be read') + ) + ).toBe(true) + }) + + it('keeps one disclosure row across reopens instead of stacking duplicates', 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') + + await open() + const reopened = await open() + expect( + reopened + .snapshot() + .items.filter( + (entry) => entry.body.kind === 'status' && entry.body.text.includes('could not be read') + ) + ).toHaveLength(1) }) it('repairs a torn tail before acknowledging the next append', async () => { diff --git a/src/main/native-chat/agent-session-journal/journal-store.ts b/src/main/native-chat/agent-session-journal/journal-store.ts index d18e8efe2cf..b3d735485f5 100644 --- a/src/main/native-chat/agent-session-journal/journal-store.ts +++ b/src/main/native-chat/agent-session-journal/journal-store.ts @@ -21,12 +21,14 @@ import { agentJournalItemKey } from '../../../shared/agent-session-journal-item- import { compactJournal, DEFAULT_JOURNAL_COMPACTION_POLICY, + journalTailIsReadyToCompact, type JournalCompactionPolicy } from './journal-compaction' import { readJournalSince } from './journal-cursor' import { publishNewEpoch } from './journal-epoch-rollover' import { appendJournalRows, ensureJournalDir } from './journal-log-file' import { + malformedRowsDisclosure, quarantineCorruptSuffix, quarantineUnreadableSchema } from './journal-corruption-quarantine' @@ -81,6 +83,7 @@ export class AgentSessionJournal { private compactedThrough = 0 private sizeBytes = 0 private readOnly = false + private malformedRows = 0 /** Serializes sequence assignment with the durable write behind it. */ private writes: Promise = Promise.resolve() @@ -128,10 +131,17 @@ export class AgentSessionJournal { this.compactedThrough = loaded.compactedThrough this.sizeBytes = loaded.sizeBytes this.readOnly = loaded.readOnly + this.malformedRows = loaded.malformedRows if (loaded.corrupt && !loaded.readOnly) { // The epoch stays put: no intact history is discarded to recover. await quarantineCorruptSuffix(this.journalDir, this.tailRows, loaded.quarantineRemainder) } + if (this.malformedRows > 0 && !this.readOnly) { + const disclosure = malformedRowsDisclosure(this.malformedRows) + await this.appendItem(disclosure.identity, disclosure.body, { + fence: this.state.highestFence + }) + } } cursor(): AgentJournalCursor { @@ -268,16 +278,6 @@ export class AgentSessionJournal { return pending } - /** Only when the retention window would actually drop rows: inside it, - * compaction rewrites an identical log, and doing that per append is a full - * state serialization on the hot path. */ - private shouldAutoCompact(now: number): boolean { - if (!this.autoCompact || this.tailRows.length <= this.compaction.minTailRows * 2) { - return false - } - return (this.tailRows[0]?.ts ?? now) < now - this.compaction.retainTailMs - } - async compact(now = this.now()): Promise { assertJournalWritable(this.readOnly, this.identity.sessionId) const result = await compactJournal({ @@ -354,7 +354,7 @@ export class AgentSessionJournal { this.sizeBytes += journalRowByteLength(row) // Nothing else calls compact(), so without this the log only ever grows — // until the size bound refuses every append for the rest of the session. - if (this.shouldAutoCompact(ts)) { + if (this.autoCompact && journalTailIsReadyToCompact(this.tailRows, this.compaction, ts)) { await this.compact(ts) } return row