Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 9 additions & 6 deletions src/main/native-chat/agent-session-journal/journal-open.ts
Original file line number Diff line number Diff line change
Expand Up @@ -26,12 +26,14 @@ const FIRST_JOURNAL_SEQUENCE = 1

export type JournalLoad = {
state: JournalReducerState
/** A row from a future schema was met: no writes, no deletion. */
/** A row from a future schema, or of a kind this build does not know, was met: no writes, no
* deletion. */
readOnly: boolean
/** Set when the surviving prefix is unusable and the caller must roll the epoch. */
corrupt: boolean
/** Rows skipped because their body failed to parse (future-version rows are
* `readOnly`, never counted here). The store discloses these in the timeline. */
/** Rows dropped because they failed to parse or name another sequence than their key (an
* unreadable row latches `readOnly`, never counted here). The store discloses these in the
* timeline. */
malformedRows: number
/** Directory-internal: the first sequence of an unusable suffix. The store
* deletes from here before it accepts a write; a probe leaves it alone. */
Expand Down Expand Up @@ -92,10 +94,11 @@ export function startJournalRowFold(input: JournalRowFoldInput): {

const add = (entry: { seq: number; rowJson: string }): boolean => {
const parsed = parseJournalRow(entry.rowJson)
if (!parsed.ok) {
// A body naming another sequence than its key is malformed there: writes number past the key.
if (!parsed.ok || parsed.row.seq !== entry.seq) {
truncateFrom = entry.seq
latched = parsed.unreadable
malformedRows = parsed.unreadable ? 0 : 1
latched = !parsed.ok && parsed.unreadable
malformedRows = latched ? 0 : 1
return false
}
const row = parsed.row
Expand Down
144 changes: 73 additions & 71 deletions src/main/native-chat/agent-session-journal/journal-row-schema.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,14 @@
// 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.
//
// A row whose KIND this build does not know is UNREADABLE the same way, and kept
// on disk, when it has the envelope every row keeps: a non-empty `epoch`, an
// integer `seq` >= 1, an integer `fence` and a numeric `ts`. So a new kind keeps
// that envelope; changing the envelope is a `v` bump. Builds from before this
// rule delete the journal from an unknown kind, so a new kind either ships its
// reader first and is written only once no supported build lacks that reader,
// or is written at a bumped `v`.

import type { AgentSessionFailureFact } from '../../../shared/agent-session-failure'
import {
Expand Down Expand Up @@ -79,10 +87,10 @@ export type JournalTombstoneRow = JournalRowBase & {
queueResume?: true
}

/** One Stop that took effect. Temporary carrier: a tombstone's extra key, because a released host
* deletes the journal from the first row kind it does not know (`journal-open.ts` then
* `journal-store-open.ts`) but ignores an unknown key; a row kind of its own once released hosts
* skip unknown kinds instead. */
/** One Stop that took effect. Temporary carrier: a tombstone's extra key, which every host ignores,
* where a host from before the header's rule deletes the journal from a kind it does not know. A
* kind of its own ships its reader first and is written once no supported build lacks that
* reader, or is written at a bumped `v`: an older host must never fold past a person's Stop. */
export type JournalStopEvent = {
/** Persisted: never rename an arm. Only `user-stop` pauses the queue. */
reason: StructuredAgentSessionStopCause
Expand Down Expand Up @@ -180,25 +188,17 @@ 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. */
/** A future schema version, or a kind this build does not know. The host must not write or
* compact this journal. */
| { ok: false; unreadable: true }

const ROW_KINDS = new Set([
'epoch',
'item',
'tombstone',
'submission',
'dispatch',
'lifecycle-batch'
])

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.
* Parse one persisted line. Older versions are upcast; newer versions and newer
* kinds are reported as unreadable so the caller fails closed.
*/
export function parseJournalRow(line: string): JournalRowParse {
let parsed: unknown
Expand Down Expand Up @@ -230,7 +230,13 @@ export function parseJournalRow(line: string): JournalRowParse {
}
}
dropUnusableContextUsage(upcast)
return isJournalRow(upcast) ? { ok: true, row: upcast } : { ok: false, unreadable: false }
if (isJournalRow(upcast)) {
return { ok: true, row: upcast }
}
// A newer build's kind is placed by the envelope every row keeps; one without it is damage.
const { kind } = upcast
const unknownKind = typeof kind === 'string' && kind !== '' && !KNOWN_ROW_KINDS.has(kind)
return { ok: false, unreadable: unknownKind && hasJournalRowEnvelope(upcast) }
}

/** Linkage ids this build cannot trust, removed from a row it still keeps.
Expand Down Expand Up @@ -317,60 +323,56 @@ function isPlainObject(value: unknown): value is Record<string, unknown> {
* schema — their nested shapes are dereferenced unguarded all the way to the
* rendered surface, so a JSON-valid corruption must fail here, not there. */
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') {
return (
typeof record.itemId === 'string' &&
Number.isInteger(record.revision) &&
isAdmissibleAgentJournalItemBody(record.body)
)
}
if (record.kind === 'tombstone') {
return typeof record.itemId === 'string' && Number.isInteger(record.revision)
}
if (record.kind === 'submission') {
return (
typeof record.clientMessageId === 'string' &&
record.clientMessageId.length > 0 &&
typeof record.payloadFingerprint === 'string' &&
isPlainObject(record.providerHandle) &&
isAdmissibleAgentJournalMessageBody(record.body)
)
}
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')
)
}
if (record.kind === 'lifecycle-batch') {
return (
typeof record.settlementId === 'string' &&
record.settlementId.length > 0 &&
Array.isArray(record.mutations) &&
record.mutations.length > 0 &&
record.mutations.length <= MAX_JOURNAL_LIFECYCLE_BATCH_MUTATIONS &&
Buffer.byteLength(JSON.stringify(record), 'utf8') + 1 <= MAX_JOURNAL_LIFECYCLE_BATCH_BYTES &&
record.mutations.every(isLifecycleMutation)
)
}
return typeof record.reason === 'string' && isPlainObject(record.providerHandle)
const fieldCheck = typeof record.kind === 'string' ? KNOWN_ROW_KINDS.get(record.kind) : undefined
return fieldCheck !== undefined && hasJournalRowEnvelope(record) && fieldCheck(record)
}

/** Each kind's own fields, keyed by every kind the union holds: a kind without a check here fails
* to compile, never reads as a newer build's kind. */
const ROW_FIELD_CHECK_BY_KIND: Record<
JournalRow['kind'],
(record: Record<string, unknown>) => boolean
> = {
epoch: (record) => typeof record.reason === 'string' && isPlainObject(record.providerHandle),
item: (record) =>
typeof record.itemId === 'string' &&
Number.isInteger(record.revision) &&
isAdmissibleAgentJournalItemBody(record.body),
tombstone: (record) => typeof record.itemId === 'string' && Number.isInteger(record.revision),
submission: (record) =>
typeof record.clientMessageId === 'string' &&
record.clientMessageId.length > 0 &&
typeof record.payloadFingerprint === 'string' &&
isPlainObject(record.providerHandle) &&
isAdmissibleAgentJournalMessageBody(record.body),
dispatch: (record) =>
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'),
'lifecycle-batch': (record) =>
typeof record.settlementId === 'string' &&
record.settlementId.length > 0 &&
Array.isArray(record.mutations) &&
record.mutations.length > 0 &&
record.mutations.length <= MAX_JOURNAL_LIFECYCLE_BATCH_MUTATIONS &&
Buffer.byteLength(JSON.stringify(record), 'utf8') + 1 <= MAX_JOURNAL_LIFECYCLE_BATCH_BYTES &&
record.mutations.every(isLifecycleMutation)
}
const KNOWN_ROW_KINDS = new Map(Object.entries(ROW_FIELD_CHECK_BY_KIND))

/** The fields every row keeps whatever its kind: its epoch, its place in it, its writer, its time. */
function hasJournalRowEnvelope(record: Record<string, unknown>): boolean {
return (
typeof record.epoch === 'string' &&
record.epoch !== '' &&
Number.isInteger(record.seq) &&
Number(record.seq) >= 1 &&
Number.isInteger(record.fence) &&
typeof record.ts === 'number'
)
}

function isLifecycleMutation(value: unknown): value is JournalLifecycleMutation {
Expand Down
Loading
Loading