fix(native-chat): close minor silent-drop seams in the journal path (#15144)

Grouped no-silent-drop minors from the structured-core review:

- A rollout file that vanishes mid-scan no longer aborts the whole TUI
  proof scan (open() was outside its try, so one ENOENT fed the
  indeterminate-owner path).
- CodexTurnOrdinals.forgetTurn was dead code deleting a bare turnId while
  keys are thread:turn. It now releases the finished turn's per-item map
  (called on turn/completed) while keeping the counter, so a straggler
  frame can never reuse an ordinal and upsert another item's row.
- Malformed journal lines were counted and discarded; the count now lands
  in the timeline as one stable disclosure row per journal, upserted on
  reopen. The malformed lines themselves stay on disk untouched.
- Dispatch and submission rows were validated only by clientMessageId;
  epoch and item rows skipped their payloads entirely. Rows are now
  type-checked field by field (never enum-checked, to stay forward
  compatible within a row version).
This commit is contained in:
Brennan Benson
2026-08-17 14:01:59 -07:00
committed by GitHub
parent abb1e05859
commit d05819bf10
13 changed files with 322 additions and 27 deletions
@@ -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))
@@ -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<string, Map<string, number>>()
private readonly turns = new Map<string, { assigned: Map<string, number>; next: number }>()
ordinalFor(threadId: string, turnId: string, codexItemId: string): number {
const turnKey = `${encodeURIComponent(threadId)}:${encodeURIComponent(turnId)}`
const assigned = this.turns.get(turnKey) ?? new Map<string, number>()
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()
}
}
}
@@ -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()
@@ -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.
@@ -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<string> {
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<string> {
yield `/pinned/sessions/2026/08/11/rollout-now-${THREAD}.jsonl`
+8 -1
View File
@@ -118,7 +118,14 @@ export async function proveCodexTuiRollout(input: {
}
async function readCodexRolloutSessionMetaId(filePath: string): Promise<string | null> {
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<ReturnType<typeof open>>
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)
@@ -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
}
@@ -38,3 +38,20 @@ export async function quarantineUnreadableSchema(journalDir: string): Promise<vo
await quarantineJournalRemainder(journalDir, preserved)
}
}
/** The disclosure row for lines that failed to parse. Skipped lines are lost
* rows; counting them silently is the drop this exists to prevent. */
export function malformedRowsDisclosure(count: number): {
identity: { provider: 'orca'; clientMessageId: string }
body: { kind: 'status'; text: string }
} {
const plural = count === 1 ? '' : 's'
return {
// One stable identity, so a reopen upserts the same row instead of adding one.
identity: { provider: 'orca', clientMessageId: 'journal-malformed-lines' },
body: {
kind: 'status',
text: `${count} journal line${plural} could not be read and ${count === 1 ? 'was' : 'were'} skipped`
}
}
}
@@ -31,6 +31,9 @@ export type JournalLoad = {
readOnly: boolean
/** Set when the surviving prefix is unusable and the caller must roll the epoch. */
corrupt: boolean
/** Log lines skipped because they failed to parse (schema-version rows are
* `readOnly`, never counted here). The store discloses these in the timeline. */
malformedRows: number
sizeBytes: number
/** Raw unreadable suffix retained for quarantine instead of deletion. */
quarantineRemainder?: string
@@ -88,6 +91,7 @@ export async function loadJournal(
compactedThrough,
readOnly: log.unreadable || (snapshot?.v ?? 0) > 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
}
}
@@ -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<string, unknown>): 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)
})
})
@@ -134,6 +134,18 @@ function upcastRow(record: Record<string, unknown>, version: number): Record<str
return current
}
function isPlainObject(value: unknown): value is Record<string, unknown> {
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<string, unknown>): record is JournalRow {
if (typeof record.kind !== 'string' || !ROW_KINDS.has(record.kind)) {
return false
@@ -148,13 +160,36 @@ function isJournalRow(record: Record<string, unknown>): 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. */
@@ -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 () => {
@@ -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<unknown> = 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<void> {
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