diff --git a/src/main/native-chat/agent-session-journal/journal-log-file.ts b/src/main/native-chat/agent-session-journal/journal-log-file.ts index c2c58471322..f67c6529faf 100644 --- a/src/main/native-chat/agent-session-journal/journal-log-file.ts +++ b/src/main/native-chat/agent-session-journal/journal-log-file.ts @@ -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 { try { const raw = await readFile(join(journalDir, JOURNAL_SNAPSHOT_FILE), 'utf-8') - const parsed = JSON.parse(raw) as Partial - 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 diff --git a/src/main/native-chat/agent-session-journal/journal-persisted-shapes.ts b/src/main/native-chat/agent-session-journal/journal-persisted-shapes.ts new file mode 100644 index 00000000000..1e88be845f7 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-persisted-shapes.ts @@ -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 { + 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, 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) +} diff --git a/src/main/native-chat/agent-session-journal/journal-persisted-validation.test.ts b/src/main/native-chat/agent-session-journal/journal-persisted-validation.test.ts new file mode 100644 index 00000000000..7ae5e22fce6 --- /dev/null +++ b/src/main/native-chat/agent-session-journal/journal-persisted-validation.test.ts @@ -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() + }) +}) 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 f7c15afe1ee..3e741c16598 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 @@ -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): 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. */