fix(native-chat): validate persisted journal state deeply

This commit is contained in:
Brennan Benson
2026-08-07 14:26:12 -07:00
parent 72b6843a4c
commit 1bbd2f01b2
4 changed files with 417 additions and 21 deletions
@@ -16,7 +16,22 @@ import type {
AgentJournalSubmission
} from '../../../shared/agent-session-journal-types'
import { AGENT_SESSION_JOURNAL_SCHEMA_VERSION } from '../../../shared/agent-session-journal-types'
import { parseJournalRow, serializeJournalRow, type JournalRow } from './journal-row-schema'
import {
isJournalAlias,
isJournalReceipt,
isJournalRenderItem,
isJournalSubmission,
isJournalTombstone,
isNonEmptyString,
isNonNegativeInteger,
isRecord
} from './journal-persisted-shapes'
import {
parseJournalRow,
parseJournalRowValue,
serializeJournalRow,
type JournalRow
} from './journal-row-schema'
export const JOURNAL_LOG_FILE = 'log.jsonl'
export const JOURNAL_SNAPSHOT_FILE = 'snapshot.json'
@@ -69,22 +84,57 @@ export async function readJournalSnapshotFile(
): Promise<JournalSnapshotReadResult> {
try {
const raw = await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8')
const parsed = JSON.parse(raw) as Partial<JournalSnapshotFile>
if (typeof parsed?.v !== 'number' || !Number.isInteger(parsed.v) || parsed.v < 1) {
return { snapshot: null, unreadable: false }
}
if (parsed.v > AGENT_SESSION_JOURNAL_SCHEMA_VERSION) {
return { snapshot: null, unreadable: true }
}
return {
snapshot: typeof parsed?.epoch === 'string' ? (parsed as JournalSnapshotFile) : null,
unreadable: false
}
return parseJournalSnapshot(JSON.parse(raw))
} catch {
return { snapshot: null, unreadable: false }
}
}
function parseJournalSnapshot(value: unknown): JournalSnapshotReadResult {
if (!isRecord(value) || !Number.isInteger(value.v) || (value.v as number) < 1) {
return { snapshot: null, unreadable: false }
}
if ((value.v as number) > AGENT_SESSION_JOURNAL_SCHEMA_VERSION) {
return { snapshot: null, unreadable: true }
}
if (
!isNonEmptyString(value.epoch) ||
!isNonNegativeInteger(value.compactedThrough) ||
!(value.highestFence === undefined || isNonNegativeInteger(value.highestFence)) ||
!Array.isArray(value.items) ||
!value.items.every(isJournalRenderItem) ||
!(value.tombstones === undefined || Array.isArray(value.tombstones)) ||
(Array.isArray(value.tombstones) && !value.tombstones.every(isJournalTombstone)) ||
!Array.isArray(value.submissions) ||
!value.submissions.every(isJournalSubmission) ||
!Array.isArray(value.receipts) ||
!value.receipts.every(isJournalReceipt) ||
!Array.isArray(value.aliases) ||
!value.aliases.every(isJournalAlias) ||
!Array.isArray(value.tail)
) {
return { snapshot: null, unreadable: false }
}
const tail: JournalRow[] = []
for (const candidate of value.tail) {
const parsed = parseJournalRowValue(candidate)
if (!parsed.ok) {
return { snapshot: null, unreadable: parsed.unreadable }
}
tail.push(parsed.row)
}
return {
snapshot: {
...value,
highestFence: value.highestFence ?? 0,
tombstones: value.tombstones ?? [],
tail
} as JournalSnapshotFile,
unreadable: false
}
}
export async function writeJournalSnapshotFile(
journalDir: string,
snapshot: JournalSnapshotFile
@@ -0,0 +1,215 @@
import type {
AgentJournalItemBody,
AgentJournalRenderItem,
AgentJournalSubmission,
AgentSessionProviderHandle
} from '../../../shared/agent-session-journal-types'
import type { NativeChatBlock } from '../../../shared/native-chat-types'
const ROLES = new Set(['user', 'assistant', 'tool', 'reasoning', 'system'])
const TOOL_STATES = new Set(['running', 'completed', 'failed'])
const RESOLUTION_STATES = new Set(['pending', 'resolved', 'cancelled'])
const DISPATCH_STATES = new Set(['pending', 'accepted', 'rejected', 'unknown'])
export function isRecord(value: unknown): value is Record<string, unknown> {
return Boolean(value) && typeof value === 'object' && !Array.isArray(value)
}
export function isNonEmptyString(value: unknown): value is string {
return typeof value === 'string' && value.length > 0
}
export function isFiniteNumber(value: unknown): value is number {
return typeof value === 'number' && Number.isFinite(value)
}
export function isNonNegativeInteger(value: unknown): value is number {
return Number.isInteger(value) && (value as number) >= 0
}
function isOptionalTrue(value: unknown): boolean {
return value === undefined || value === true
}
function isOptionalString(value: unknown): boolean {
return value === undefined || typeof value === 'string'
}
function isNullableString(value: unknown): boolean {
return value === null || typeof value === 'string'
}
function isNullableNumber(value: unknown): boolean {
return value === null || isFiniteNumber(value)
}
function hasOwn(value: Record<string, unknown>, key: string): boolean {
return Object.prototype.hasOwnProperty.call(value, key)
}
export function isAgentSessionProviderHandle(value: unknown): value is AgentSessionProviderHandle {
if (!isRecord(value)) {
return false
}
if (value.kind === 'codex') {
return isNonEmptyString(value.threadId)
}
if (value.kind === 'claude') {
return isNonEmptyString(value.sessionId) && isNullableString(value.leafUuid)
}
return value.kind === 'opaque' && isNonEmptyString(value.agent) && isNonEmptyString(value.value)
}
function isNativeChatBlock(value: unknown): value is NativeChatBlock {
if (!isRecord(value)) {
return false
}
if (value.type === 'text') {
return typeof value.text === 'string'
}
if (value.type === 'tool-call') {
return typeof value.name === 'string' && hasOwn(value, 'input')
}
if (value.type === 'tool-result') {
return (
typeof value.output === 'string' &&
(value.isError === undefined || typeof value.isError === 'boolean')
)
}
return (
value.type === 'image-ref' &&
isOptionalString(value.path) &&
isOptionalString(value.url) &&
isOptionalString(value.alt)
)
}
function isBoundedPayload(value: unknown): boolean {
return (
isRecord(value) &&
typeof value.head === 'string' &&
isNonNegativeInteger(value.byteLength) &&
typeof value.digest === 'string' &&
typeof value.truncated === 'boolean'
)
}
function isPromptOption(value: unknown): boolean {
return isRecord(value) && isNonEmptyString(value.id) && typeof value.label === 'string'
}
function isResolution(value: unknown): boolean {
return (
isRecord(value) &&
typeof value.state === 'string' &&
RESOLUTION_STATES.has(value.state) &&
isNullableString(value.selectedOptionId) &&
isNullableString(value.resolvedBy) &&
isNullableNumber(value.resolvedAt)
)
}
export function isAgentJournalItemBody(value: unknown): value is AgentJournalItemBody {
if (!isRecord(value)) {
return false
}
if (value.kind === 'message') {
return (
typeof value.role === 'string' &&
ROLES.has(value.role) &&
Array.isArray(value.blocks) &&
value.blocks.every(isNativeChatBlock)
)
}
if (value.kind === 'tool-call') {
return (
typeof value.name === 'string' &&
hasOwn(value, 'input') &&
typeof value.state === 'string' &&
TOOL_STATES.has(value.state) &&
(value.output === undefined || isBoundedPayload(value.output))
)
}
if (value.kind === 'diff') {
return typeof value.path === 'string' && isBoundedPayload(value.patch)
}
if (value.kind === 'approval') {
return (
typeof value.title === 'string' &&
isNullableString(value.detail) &&
Array.isArray(value.options) &&
value.options.every(isPromptOption) &&
isResolution(value.resolution)
)
}
if (value.kind === 'question') {
return (
typeof value.question === 'string' &&
Array.isArray(value.options) &&
value.options.every(isPromptOption) &&
isResolution(value.resolution)
)
}
return value.kind === 'status' && typeof value.text === 'string'
}
export function isJournalRenderItem(value: unknown): value is AgentJournalRenderItem {
return (
isRecord(value) &&
isNonEmptyString(value.itemId) &&
isNonNegativeInteger(value.revision) &&
isAgentJournalItemBody(value.body) &&
Number.isInteger(value.sequence) &&
(value.sequence as number) >= 1 &&
isFiniteNumber(value.observedAt) &&
isOptionalTrue(value.recovered)
)
}
export function isJournalSubmission(value: unknown): value is AgentJournalSubmission {
if (
!isRecord(value) ||
!isNonEmptyString(value.clientMessageId) ||
!isNonNegativeInteger(value.fence) ||
typeof value.payloadFingerprint !== 'string' ||
typeof value.dispatchState !== 'string' ||
!DISPATCH_STATES.has(value.dispatchState) ||
!isNullableString(value.providerItemId) ||
!isNullableString(value.reason) ||
!isFiniteNumber(value.submittedAt) ||
!isNullableNumber(value.resolvedAt)
) {
return false
}
if (value.dispatchState === 'pending') {
return value.providerItemId === null && value.reason === null && value.resolvedAt === null
}
if (value.dispatchState === 'accepted') {
return (
isNonEmptyString(value.providerItemId) &&
value.reason === null &&
isFiniteNumber(value.resolvedAt)
)
}
return value.providerItemId === null && isFiniteNumber(value.resolvedAt)
}
export function isJournalTombstone(value: unknown): boolean {
return isRecord(value) && isNonEmptyString(value.itemId) && isNonNegativeInteger(value.revision)
}
export function isJournalReceipt(value: unknown): boolean {
return (
isRecord(value) &&
isNonEmptyString(value.clientMessageId) &&
isNonEmptyString(value.providerItemId) &&
isNonEmptyString(value.epoch) &&
Number.isInteger(value.sequence) &&
(value.sequence as number) >= 1 &&
isFiniteNumber(value.acceptedAt)
)
}
export function isJournalAlias(value: unknown): boolean {
return isRecord(value) && isNonEmptyString(value.providerItemId) && isNonEmptyString(value.itemId)
}
@@ -0,0 +1,88 @@
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,
AgentSessionJournalIdentity
} from '../../../shared/agent-session-journal-types'
import { JOURNAL_LOG_FILE, JOURNAL_SNAPSHOT_FILE } from './journal-log-file'
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' }
}
let root: string
let clock = 1_000
function body(text: string): AgentJournalItemBody {
return { kind: 'message', role: 'assistant', blocks: [{ type: 'text', text }] }
}
async function open() {
return openAgentSessionJournal({
identity: IDENTITY,
journalDir: root,
now: () => ++clock,
mintEpoch: () => `epoch-${clock}`
})
}
beforeEach(async () => {
root = await mkdtemp(join(tmpdir(), 'orca-journal-validation-'))
clock = 1_000
})
afterEach(async () => {
await rm(root, { recursive: true, force: true })
})
describe('persisted journal validation', () => {
it('does not admit a malformed item body into reducer state', async () => {
const journal = await open()
const malformed = JSON.stringify({
v: 1,
kind: 'item',
epoch: journal.epoch,
seq: 2,
fence: 1,
ts: 1_002,
itemId: 'malformed',
revision: 1,
body: null
})
const logPath = join(root, JOURNAL_LOG_FILE)
await writeFile(logPath, `${await readFile(logPath, 'utf-8')}${malformed}\n`, 'utf-8')
const reopened = await open()
expect(reopened.snapshot().items).toEqual([])
await expect(reopened.compact(1)).resolves.toBeUndefined()
})
it('rebuilds from the log when a snapshot contains a malformed render item', async () => {
const journal = await open()
await journal.appendItem(
{ provider: 'codex', threadId: 'thread-1', turnId: 'turn-1', ordinal: 0 },
body('valid'),
{ fence: 1 }
)
const expected = journal.snapshot()
await journal.compact(1)
const snapshotPath = join(root, JOURNAL_SNAPSHOT_FILE)
const snapshot = JSON.parse(await readFile(snapshotPath, 'utf-8')) as {
items: { body: unknown }[]
}
snapshot.items[0]!.body = null
await writeFile(snapshotPath, JSON.stringify(snapshot), 'utf-8')
const reopened = await open()
expect(reopened.snapshot()).toEqual(expected)
await expect(reopened.compact(1)).resolves.toBeUndefined()
})
})
@@ -12,6 +12,13 @@ import {
type AgentJournalMessageItem,
type AgentSessionProviderHandle
} from '../../../shared/agent-session-journal-types'
import {
isAgentJournalItemBody,
isAgentSessionProviderHandle,
isFiniteNumber,
isNonEmptyString,
isNonNegativeInteger
} from './journal-persisted-shapes'
export type JournalRowBase = {
/** Schema version of THIS row. */
@@ -91,6 +98,8 @@ export type JournalRowParse =
| { ok: false; unreadable: true }
const ROW_KINDS = new Set(['epoch', 'item', 'tombstone', 'submission', 'dispatch'])
const EPOCH_REASONS = new Set(AGENT_JOURNAL_EPOCH_REASONS)
const DISPATCH_STATES = new Set(['accepted', 'rejected', 'unknown'])
export function journalRowBase(
epoch: string,
@@ -116,6 +125,10 @@ export function parseJournalRow(line: string): JournalRowParse {
} catch {
return { ok: false, unreadable: false }
}
return parseJournalRowValue(parsed)
}
export function parseJournalRowValue(parsed: unknown): JournalRowParse {
if (!parsed || typeof parsed !== 'object' || Array.isArray(parsed)) {
return { ok: false, unreadable: false }
}
@@ -148,22 +161,52 @@ function isJournalRow(record: Record<string, unknown>): record is JournalRow {
return false
}
if (
typeof record.epoch !== 'string' ||
!record.epoch ||
!isNonEmptyString(record.epoch) ||
!Number.isInteger(record.seq) ||
(record.seq as number) < 1 ||
!Number.isInteger(record.fence) ||
typeof record.ts !== 'number'
!isNonNegativeInteger(record.fence) ||
!isFiniteNumber(record.ts) ||
(record.recovered !== undefined && record.recovered !== true)
) {
return false
}
if (record.kind === 'item' || record.kind === 'tombstone') {
return typeof record.itemId === 'string' && Number.isInteger(record.revision)
if (record.kind === 'epoch') {
return (
typeof record.reason === 'string' &&
EPOCH_REASONS.has(record.reason as AgentJournalEpochReason) &&
isAgentSessionProviderHandle(record.providerHandle)
)
}
if (record.kind === 'submission' || record.kind === 'dispatch') {
return typeof record.clientMessageId === 'string' && record.clientMessageId.length > 0
if (record.kind === 'item') {
return (
isNonEmptyString(record.itemId) &&
isNonNegativeInteger(record.revision) &&
isAgentJournalItemBody(record.body)
)
}
return typeof record.reason === 'string'
if (record.kind === 'tombstone') {
return isNonEmptyString(record.itemId) && isNonNegativeInteger(record.revision)
}
if (record.kind === 'submission') {
return (
isNonEmptyString(record.clientMessageId) &&
typeof record.payloadFingerprint === 'string' &&
isAgentSessionProviderHandle(record.providerHandle) &&
isAgentJournalItemBody(record.body) &&
record.body.kind === 'message'
)
}
if (
!isNonEmptyString(record.clientMessageId) ||
typeof record.state !== 'string' ||
!DISPATCH_STATES.has(record.state) ||
!(record.reason === null || typeof record.reason === 'string')
) {
return false
}
return record.state === 'accepted'
? isNonEmptyString(record.providerItemId) && record.reason === null
: record.providerItemId === null
}
/** Approximate on-disk cost of a row, used for the per-session size bound. */