diff --git a/.changeset/state-review-followups.md b/.changeset/state-review-followups.md new file mode 100644 index 000000000..4bbbefa1d --- /dev/null +++ b/.changeset/state-review-followups.md @@ -0,0 +1,17 @@ +--- +"@agent-bundle/runtime": patch +--- + +State kernel review follow-ups from #142/#149. Both drivers now consult the +idempotency key before running the reducer, so a committed key replays its +stored result even when the reducer would fail against the current head; the +committed result is stored per key (event journal rows now persist their +post-commit state) and rides the migration chain, so replay survives schema +migrations instead of failing `revision-unavailable`. The sqlite driver +verifies storage on open — journal continuity (a hand-deleted intermediate +row fails closed) and the materialized head against journal replay (a +schema-valid but hand-edited head fails closed) — and derives database file +names from a sha-256 hash of the complete definition id, so ids that share a +sanitized prefix no longer collide onto one file. Sparse arrays are rejected +at the JSON boundary instead of silently canonicalizing like dense ones. The +shared conformance suite pins the corrected semantics for every driver. diff --git a/packages/rsc-runtime/src/state/conformance.ts b/packages/rsc-runtime/src/state/conformance.ts index 550e75f87..1907cd7e2 100644 --- a/packages/rsc-runtime/src/state/conformance.ts +++ b/packages/rsc-runtime/src/state/conformance.ts @@ -223,6 +223,22 @@ export const stateDriverConformanceCases: readonly StateConformanceCase[] = Obje assert.equal((await store.read()).revision, 2); }, }, + { + name: 'a committed key replays without re-running the reducer', + run: async (context) => { + const store = await context.open(taskDefinition(context.lifetime)); + await addTask(store, 'a'); + const removed = await store.dispatch('taskRemoved', { id: 'a' }, { idempotencyKey: 'remove:a' }); + assert.equal(removed.revision, 2); + // Task 'a' is gone, so the reducer would now throw; the key must be + // consulted before the reducer runs for the retry to replay. + const replayed = await store.dispatch('taskRemoved', { id: 'a' }, { idempotencyKey: 'remove:a' }); + assert.equal(replayed.replayed, true); + assert.equal(replayed.revision, removed.revision); + assert.deepEqual(replayed.state, removed.state); + assert.equal((await store.read()).revision, 2); + }, + }, { name: 'reusing an idempotency key with a different payload is an idempotency-conflict', run: async (context) => { @@ -433,6 +449,36 @@ export const stateDriverConformanceCases: readonly StateConformanceCase[] = Obje await assert.rejects(context.reopen(taskDefinition(context.lifetime)), rejectsWith('migration-missing')); }, }, + { + name: 'replaying a key committed before a migration returns its committed result', + run: async (context) => { + const storeV1 = await context.open(taskDefinition(context.lifetime)); + const original = await addTask(storeV1, 'a'); + await addTask(storeV1, 'b'); + const storeV2 = await context.reopen(taskDefinitionV2(context.lifetime)); + assert.equal((await storeV2.read()).revision, 3); + // The migration rebases exact-revision history, but a pre-deployment + // retry must still replay: the committed result rides the migration + // chain instead of depending on exact-revision replay. + const replayed = await storeV2.dispatch('taskAdded', { id: 'a', title: 'Task a' }, { idempotencyKey: 'add:a' }); + assert.equal(replayed.replayed, true); + assert.equal(replayed.revision, original.revision); + assert.deepEqual(replayed.state, { labels: [], tasks: [{ id: 'a', title: 'Task a' }], total: 1 }); + assert.equal((await storeV2.read()).revision, 3); + }, + }, + { + name: 'definition ids sharing a sanitized prefix stay isolated', + run: async (context) => { + // These two ids sanitize identically and share their leading bytes, so + // storage naming must derive from the complete id, never a truncation. + const first = await context.open(taskDefinition(context.lifetime, 'abcdef/a')); + const second = await context.open(taskDefinition(context.lifetime, 'abcdef-a')); + await addTask(first, 'a'); + assert.equal((await second.read()).revision, 0); + assert.deepEqual((await second.read()).state, { tasks: [], total: 0 }); + }, + }, { name: 'closed stores fail typed', run: async (context) => { diff --git a/packages/rsc-runtime/src/state/contract.ts b/packages/rsc-runtime/src/state/contract.ts index b632b6979..1097becf7 100644 --- a/packages/rsc-runtime/src/state/contract.ts +++ b/packages/rsc-runtime/src/state/contract.ts @@ -105,8 +105,12 @@ export type AgentStateEvent = { * persisted at definition version `n - 1` to version `n`; a definition of * version `v > 1` must supply every step from `2` through `v`. Steps receive * the raw persisted value and their final output must satisfy the current - * schema. Migrating rebases history: exact-revision reads below the recorded - * migration become `revision-unavailable`. + * schema; steps must accept any valid version `n - 1` state, because + * committed results stored for idempotent replay migrate through the same + * chain. Migrating rebases history: exact-revision reads below the recorded + * migration become `revision-unavailable`, but replaying a committed + * idempotency key still returns its committed result (migrated to the + * current version). */ export type AgentStateMigrations = Readonly unknown>>; diff --git a/packages/rsc-runtime/src/state/index.ts b/packages/rsc-runtime/src/state/index.ts index 4f0a04ad5..93f2210f7 100644 --- a/packages/rsc-runtime/src/state/index.ts +++ b/packages/rsc-runtime/src/state/index.ts @@ -46,6 +46,8 @@ export { changeFromJournalRecord, expectConsistentJournal, migrationIdempotencyKey, + parseEventPayload, + reduceStateEvent, replayJournal, resolveResetState, runStateMigrations, diff --git a/packages/rsc-runtime/src/state/journal.ts b/packages/rsc-runtime/src/state/journal.ts index ff0b07e11..8277f8440 100644 --- a/packages/rsc-runtime/src/state/journal.ts +++ b/packages/rsc-runtime/src/state/journal.ts @@ -66,16 +66,16 @@ export const migrationIdempotencyKey = (toVersion: number): string => `${AGENT_STATE_RESERVED_KEY_PREFIX}migrate:${String(toVersion)}`; /** - * Validates one event payload, runs the reducer, and validates its output. - * Throws typed `invalid-event`, `reducer-failure`, or `invalid-state` - * errors; never exposes payload or state contents in messages. + * Validates one event payload against its declared schema without running + * the reducer. Idempotency-key replay must be decided from the validated + * payload alone: a committed key retried after the state changed replays + * the committed result, so the reducer must not run first. */ -export const applyStateEvent = ( +export const parseEventPayload = ( definition: AgentStateDefinition, - state: TState, name: string, payload: unknown, -): { readonly payload: unknown; readonly state: TState } => { +): unknown => { const schema = definition.events[name]; if (schema === undefined) { throw new AgentStateError('invalid-event', `State '${definition.id}' has no event '${name}'`); @@ -90,9 +90,23 @@ export const applyStateEvent = ( if (!isJsonSafe(parsed.data)) { throw new AgentStateError('invalid-event', `State '${definition.id}' event '${name}' payload must be JSON-safe`); } + return deepFreezeJson(parsed.data); +}; + +/** + * Runs the reducer over an already-validated payload (see + * {@link parseEventPayload}) and validates its output. Throws typed + * `reducer-failure` or `invalid-state`; never exposes state contents. + */ +export const reduceStateEvent = ( + definition: AgentStateDefinition, + state: TState, + name: string, + payload: unknown, +): TState => { let next: TState; try { - next = definition.reduce(state, { name, payload: parsed.data } as AgentStateEvent); + next = definition.reduce(state, { name, payload } as AgentStateEvent); } catch (error) { throw new AgentStateError( 'reducer-failure', @@ -110,7 +124,22 @@ export const applyStateEvent = ( if (!isJsonSafe(validated.data)) { throw new AgentStateError('invalid-state', `State '${definition.id}' reducer output for event '${name}' must be JSON-safe`); } - return { payload: deepFreezeJson(parsed.data), state: deepFreezeJson(validated.data) }; + return deepFreezeJson(validated.data); +}; + +/** + * Validates one event payload, runs the reducer, and validates its output. + * Throws typed `invalid-event`, `reducer-failure`, or `invalid-state` + * errors; never exposes payload or state contents in messages. + */ +export const applyStateEvent = ( + definition: AgentStateDefinition, + state: TState, + name: string, + payload: unknown, +): { readonly payload: unknown; readonly state: TState } => { + const parsed = parseEventPayload(definition, name, payload); + return { payload: parsed, state: reduceStateEvent(definition, state, name, parsed) }; }; /** Validates a reset seed (or resolves the initial state) against the schema. */ diff --git a/packages/rsc-runtime/src/state/json.ts b/packages/rsc-runtime/src/state/json.ts index 3608a0452..8273f1439 100644 --- a/packages/rsc-runtime/src/state/json.ts +++ b/packages/rsc-runtime/src/state/json.ts @@ -16,7 +16,15 @@ const isPlainObject = (value: unknown): value is Readonly { if (value === null || typeof value === 'string' || typeof value === 'boolean') return true; if (typeof value === 'number') return Number.isFinite(value); - if (Array.isArray(value)) return value.every(isJsonSafe); + if (Array.isArray(value)) { + // Index-by-index so holes fail closed: `every` skips holes, which would + // let a sparse array canonicalize to the same text as a denser one and + // break both round-tripping and idempotency-key comparison. + for (let index = 0; index < value.length; index += 1) { + if (!(index in value) || !isJsonSafe(value[index])) return false; + } + return true; + } return isPlainObject(value) && Object.values(value).every(isJsonSafe); }; diff --git a/packages/rsc-runtime/src/state/memory-driver.ts b/packages/rsc-runtime/src/state/memory-driver.ts index 94cb0e308..ef11aa6e9 100644 --- a/packages/rsc-runtime/src/state/memory-driver.ts +++ b/packages/rsc-runtime/src/state/memory-driver.ts @@ -21,10 +21,11 @@ import type { import { AgentStateError, expectIdempotencyKey } from './contract.js'; import type { AgentStateJournalRecord } from './journal.js'; import { - applyStateEvent, canonicalCommitInput, changeFromJournalRecord, migrationIdempotencyKey, + parseEventPayload, + reduceStateEvent, replayJournal, resolveResetState, runStateMigrations, @@ -70,12 +71,18 @@ const expectVolatileLifetime = (lifetime: AgentStateLifetime): MemoryLifetime => } }; +interface CommittedResult { + readonly record: AgentStateJournalRecord; + /** Post-commit state, kept per key so replay survives history rebases. */ + readonly state: TState; +} + interface MemoryStoreInternals { closed: boolean; definition: AgentStateDefinition; head: AgentStateSnapshot; readonly journal: AgentStateJournalRecord[]; - readonly keys: Map; + readonly keys: Map>; } interface MemoryStoreEntry { @@ -133,31 +140,31 @@ const createMemoryStore = ( /** * Commits share one shape: validate inputs, then honor a committed - * idempotency key (replay/conflict), then compare-and-swap, then append. - * The interior is fully synchronous, so a commit is atomic per store - * within this process. + * idempotency key (replay/conflict), then compare-and-swap, then run the + * reducer, then append. The key check precedes the reducer because a + * committed key must replay its stored result even when the state that + * produced it has since changed — and that stored result is kept per key, + * so replay never depends on exact-revision history (which migrations + * rebase). The interior is fully synchronous, so a commit is atomic per + * store within this process. */ const commit = ( - record: Readonly<{ canonicalInput: string; key: string }> & + input: Readonly<{ canonicalInput: string; key: string }> & ( - | { readonly kind: 'event'; readonly name: string; readonly payload: unknown; readonly state: TState } + | { readonly kind: 'event'; readonly name: string; readonly payload: unknown } | { readonly kind: 'reset'; readonly state: TState } ), expectedRevision: number | undefined, ): AgentStateCommitResult => { - const committed = internals.keys.get(record.key); + const committed = internals.keys.get(input.key); if (committed !== undefined) { - if (canonicalCommitInput(committed) !== record.canonicalInput) { + if (canonicalCommitInput(committed.record) !== input.canonicalInput) { throw new AgentStateError( 'idempotency-conflict', `State '${internals.definition.id}' idempotency key was reused with a conflicting input`, ); } - return Object.freeze({ - replayed: true, - revision: committed.revision, - state: replayJournal(internals.definition, internals.journal, committed.revision), - }); + return Object.freeze({ replayed: true, revision: committed.record.revision, state: committed.state }); } if (expectedRevision !== undefined && expectedRevision !== internals.head.revision) { throw new AgentStateError( @@ -165,27 +172,31 @@ const createMemoryStore = ( `State '${internals.definition.id}' expected revision ${String(expectedRevision)} but the head is ${String(internals.head.revision)}`, ); } + const state = + input.kind === 'event' + ? reduceStateEvent(internals.definition, internals.head.state, input.name, input.payload) + : input.state; const journalRecord: AgentStateJournalRecord = - record.kind === 'event' + input.kind === 'event' ? { committedAt: now().toISOString(), - idempotencyKey: record.key, + idempotencyKey: input.key, kind: 'event', - name: record.name, - payload: record.payload, + name: input.name, + payload: input.payload, revision: internals.head.revision + 1, } : { committedAt: now().toISOString(), - idempotencyKey: record.key, + idempotencyKey: input.key, kind: 'reset', revision: internals.head.revision + 1, - state: record.state, + state: input.state, }; internals.journal.push(journalRecord); - internals.keys.set(journalRecord.idempotencyKey, journalRecord); - internals.head = Object.freeze({ revision: journalRecord.revision, state: record.state }); - return Object.freeze({ replayed: false, revision: journalRecord.revision, state: record.state }); + internals.keys.set(journalRecord.idempotencyKey, { record: journalRecord, state }); + internals.head = Object.freeze({ revision: journalRecord.revision, state }); + return Object.freeze({ replayed: false, revision: journalRecord.revision, state }); }; const store: AgentStateStore = { @@ -227,19 +238,18 @@ const createMemoryStore = ( expectOperable(internals.closed, internals.definition.id, options.signal); const key = expectIdempotencyKey(options.idempotencyKey); expectRevisionShape(options.expectedRevision, `State '${internals.definition.id}' expectedRevision`); - const applied = applyStateEvent(internals.definition, internals.head.state, name, payload); + // Payload validation only — the reducer runs inside `commit`, after + // the idempotency key has been consulted. + const parsed = parseEventPayload(internals.definition, name, payload); const canonicalInput = canonicalCommitInput({ committedAt: '', idempotencyKey: key, kind: 'event', name, - payload: applied.payload, + payload: parsed, revision: 0, }); - return commit( - { canonicalInput, key, kind: 'event', name, payload: applied.payload, state: applied.state }, - options.expectedRevision, - ); + return commit({ canonicalInput, key, kind: 'event', name, payload: parsed }, options.expectedRevision); }), ); }, @@ -298,7 +308,8 @@ const migrateOpenStore = ( definition: AgentStateDefinition, now: () => Date, ): void => { - const migrated = runStateMigrations(definition, internals.definition.version, internals.head.state); + const fromVersion = internals.definition.version; + const migrated = runStateMigrations(definition, fromVersion, internals.head.state); const record: AgentStateJournalRecord = { committedAt: now().toISOString(), idempotencyKey: migrationIdempotencyKey(definition.version), @@ -308,7 +319,16 @@ const migrateOpenStore = ( toVersion: definition.version, }; internals.journal.push(record); - internals.keys.set(record.idempotencyKey, record); + // Committed results replay across migrations: every stored result sits at + // `fromVersion` (this loop maintains that inductively), so each one rides + // the same migration chain as the head. + for (const [key, entry] of internals.keys) { + internals.keys.set(key, { + record: entry.record, + state: runStateMigrations(definition, fromVersion, entry.state), + }); + } + internals.keys.set(record.idempotencyKey, { record, state: migrated }); internals.head = Object.freeze({ revision: record.revision, state: migrated }); internals.definition = definition; }; diff --git a/packages/rsc-runtime/src/state/sqlite.ts b/packages/rsc-runtime/src/state/sqlite.ts index 490621634..ebad92a2e 100644 --- a/packages/rsc-runtime/src/state/sqlite.ts +++ b/packages/rsc-runtime/src/state/sqlite.ts @@ -1,3 +1,4 @@ +import { createHash } from 'node:crypto'; import { mkdirSync } from 'node:fs'; import { dirname, join, resolve } from 'node:path'; // node:sqlite emits an ExperimentalWarning on load (documented in the README): @@ -35,14 +36,16 @@ import type { } from './index.js'; import { AgentStateError, - applyStateEvent, canonicalCommitInput, canonicalJson, changeFromJournalRecord, deepFreezeJson, describeSchemaIssues, + expectConsistentJournal, expectIdempotencyKey, migrationIdempotencyKey, + parseEventPayload, + reduceStateEvent, replayJournal, resolveResetState, runStateMigrations, @@ -220,8 +223,11 @@ const recordFromRow = (definitionId: string, row: JournalRow): AgentStateJournal ); }; +// The readable prefix is lossy (distinct ids can sanitize identically), so a +// hash of the complete id disambiguates; truncating an encoding of only the +// leading bytes would collide for ids sharing a prefix. const sanitizedFileName = (definitionId: string): string => - `${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${Buffer.from(definitionId, 'utf8').toString('hex').slice(0, 12)}.sqlite`; + `${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${createHash('sha256').update(definitionId, 'utf8').digest('hex').slice(0, 16)}.sqlite`; class SqliteConnection extends Context.Service()( '@agent-bundle/runtime/state/SqliteConnection', @@ -331,11 +337,38 @@ class SqliteStore implements Age return row.revision; } - #committedByKey(db: DatabaseSync, key: string): AgentStateJournalRecord | undefined { + #committedByKey( + db: DatabaseSync, + key: string, + ): { readonly record: AgentStateJournalRecord; readonly stateText: string | null } | undefined { const row = db.prepare('SELECT * FROM agent_state_journal WHERE idempotency_key = ?').get(key) as | JournalRow | undefined; - return row === undefined ? undefined : recordFromRow(this.#definition.id, row); + return row === undefined ? undefined : { record: recordFromRow(this.#definition.id, row), stateText: row.state }; + } + + /** + * Recovers the state a committed record produced. Every record stores its + * post-commit state (migrated forward on schema migrations), so replay + * does not depend on exact-revision history; rows written before post- + * commit states were stored fall back to journal replay. + */ + #committedState( + db: DatabaseSync, + committed: { readonly record: AgentStateJournalRecord; readonly stateText: string | null }, + ): TState { + const raw = + committed.stateText !== null + ? parseStoredJson(this.#definition.id, 'state', committed.record.revision, committed.stateText) + : this.#replayTo(db, committed.record.revision); + const parsed = this.#definition.schema.safeParse(raw); + if (!parsed.success) { + throw new AgentStateError( + 'corrupt', + `State '${this.#definition.id}' committed result at revision ${String(committed.record.revision)} no longer satisfies the schema: ${describeSchemaIssues(parsed.error)}`, + ); + } + return deepFreezeJson(parsed.data); } #replayTo(db: DatabaseSync, revision: number): TState { @@ -360,7 +393,9 @@ class SqliteStore implements Age record.kind, record.kind === 'event' ? record.name : null, record.kind === 'event' ? canonicalJson(record.payload) : null, - record.kind === 'event' ? null : stateText, + // Event rows store their post-commit state too, so idempotent replay + // survives migrations without exact-revision replay. + stateText, record.kind === 'migrate' ? record.toVersion : null, record.idempotencyKey, record.committedAt, @@ -379,61 +414,97 @@ class SqliteStore implements Age const appendRecord = (db: DatabaseSync, record: AgentStateJournalRecord, state: TState) => this.#appendRecord(db, record, state); const committedByKey = (db: DatabaseSync, key: string) => this.#committedByKey(db, key); + const committedState = ( + db: DatabaseSync, + committed: { readonly record: AgentStateJournalRecord; readonly stateText: string | null }, + ) => this.#committedState(db, committed); const headState = (db: DatabaseSync) => this.#headState(db, 'commit'); const now = this.#now; - const replayTo = (db: DatabaseSync, revision: number) => this.#replayTo(db, revision); const transaction = this.#transaction.bind(this); - const validate = sqliteEffect(definition.id, 'validate commit', () => { + // Validation and canonicalization happen before the reducer and before + // any storage access: a committed key must replay its stored result even + // when the reducer would fail against the current head. + type PreparedCommit = + | { readonly canonicalInput: string; readonly kind: 'event'; readonly name: string; readonly payload: unknown } + | { readonly canonicalInput: string; readonly kind: 'reset'; readonly state: TState }; + const validate = sqliteEffect(definition.id, 'validate commit', (): { key: string; prepared: PreparedCommit } => { expectOperable(this.#closed, definition.id, options.signal); expectRevisionShape(options.expectedRevision, `State '${definition.id}' expectedRevision`); - return expectIdempotencyKey(options.idempotencyKey); + const key = expectIdempotencyKey(options.idempotencyKey); + if (input.kind === 'event') { + const payload = parseEventPayload(definition, input.name, input.rawPayload); + const canonicalInput = canonicalCommitInput({ + committedAt: '', + idempotencyKey: key, + kind: 'event', + name: input.name, + payload, + revision: 0, + }); + return { key, prepared: { canonicalInput, kind: 'event', name: input.name, payload } }; + } + const state = resolveResetState(definition, input.seed); + const canonicalInput = canonicalCommitInput({ + committedAt: '', + idempotencyKey: key, + kind: 'reset', + revision: 0, + state, + }); + return { key, prepared: { canonicalInput, kind: 'reset', state } }; }); return Effect.gen(function*() { - const key = yield* validate; + const { key, prepared } = yield* validate; return yield* transaction('write', input.kind === 'event' ? `dispatch '${input.name}'` : 'reset', (db) => { - const head = headState(db); - const prepared = - input.kind === 'event' - ? ((): { canonicalInput: string; record: AgentStateJournalRecord; state: TState } => { - const applied = applyStateEvent(definition, head.state, input.name, input.rawPayload); - const record: AgentStateJournalRecord = { - committedAt: now().toISOString(), - idempotencyKey: key, - kind: 'event', - name: input.name, - payload: applied.payload, - revision: head.revision + 1, - }; - return { canonicalInput: canonicalCommitInput(record), record, state: applied.state }; - })() - : ((): { canonicalInput: string; record: AgentStateJournalRecord; state: TState } => { - const state = resolveResetState(definition, input.seed); - const record: AgentStateJournalRecord = { - committedAt: now().toISOString(), - idempotencyKey: key, - kind: 'reset', - revision: head.revision + 1, - state, - }; - return { canonicalInput: canonicalCommitInput(record), record, state }; - })(); const committed = committedByKey(db, key); if (committed !== undefined) { - if (canonicalCommitInput(committed) !== prepared.canonicalInput) { + if (canonicalCommitInput(committed.record) !== prepared.canonicalInput) { throw new AgentStateError( 'idempotency-conflict', `State '${definition.id}' idempotency key was reused with a conflicting input`, ); } - return Object.freeze({ replayed: true, revision: committed.revision, state: replayTo(db, committed.revision) }); + return Object.freeze({ + replayed: true, + revision: committed.record.revision, + state: committedState(db, committed), + }); } + const head = headState(db); if (options.expectedRevision !== undefined && options.expectedRevision !== head.revision) { throw new AgentStateError( 'revision-conflict', `State '${definition.id}' expected revision ${String(options.expectedRevision)} but the head is ${String(head.revision)}`, ); } - return appendRecord(db, prepared.record, prepared.state); + const committedAt = now().toISOString(); + switch (prepared.kind) { + case 'event': { + const state = reduceStateEvent(definition, head.state, prepared.name, prepared.payload); + return appendRecord( + db, + { + committedAt, + idempotencyKey: key, + kind: 'event', + name: prepared.name, + payload: prepared.payload, + revision: head.revision + 1, + }, + state, + ); + } + case 'reset': + return appendRecord( + db, + { committedAt, idempotencyKey: key, kind: 'reset', revision: head.revision + 1, state: prepared.state }, + prepared.state, + ); + default: { + const unreachable: never = prepared; + throw new AgentStateError('invalid-input', `Unknown commit kind ${String(unreachable)}`); + } + } }); }); } @@ -565,20 +636,56 @@ class SqliteStore implements Age ); } const head = this.#headRow(transactionDb, 'open'); - const journalHead = ( - transactionDb.prepare('SELECT COALESCE(MAX(revision), 0) AS revision FROM agent_state_journal').get() as { - revision: number; - } - ).revision; + const rows = transactionDb + .prepare('SELECT * FROM agent_state_journal ORDER BY revision') + .all() as unknown as JournalRow[]; + const records = rows.map((row) => recordFromRow(definition.id, row)); + // Continuity first: a hand-deleted intermediate row must fail closed + // even when the final revision is still present. + expectConsistentJournal(definition.id, records); + const journalHead = records.length === 0 ? 0 : (records[records.length - 1] as AgentStateJournalRecord).revision; if (head.revision !== journalHead) { throw new AgentStateError( 'corrupt', `State '${definition.id}' head revision ${String(head.revision)} does not match the journal head ${String(journalHead)}`, ); } - if (meta.schema_version === definition.version) return; const rawHead = parseStoredJson(definition.id, 'head state', head.revision, head.state); + if (meta.schema_version === definition.version) { + // The materialized head must agree with journal replay: a corrupt or + // hand-edited head that still parses is otherwise served silently. + const replayed = replayJournal(definition, records, head.revision); + if (canonicalJson(replayed) !== canonicalJson(rawHead)) { + throw new AgentStateError( + 'corrupt', + `State '${definition.id}' head state at revision ${String(head.revision)} disagrees with journal replay`, + ); + } + return; + } + // A pending migration cannot replay records written under the older + // definition; verify the head against the last stored post-commit + // state instead (rows predating stored event states leave it null). + const lastStateText = rows.length === 0 ? null : (rows[rows.length - 1] as JournalRow).state; + if (lastStateText !== null) { + const lastState = parseStoredJson(definition.id, 'state', journalHead, lastStateText); + if (canonicalJson(lastState) !== canonicalJson(rawHead)) { + throw new AgentStateError( + 'corrupt', + `State '${definition.id}' head state at revision ${String(head.revision)} disagrees with the journal`, + ); + } + } const migrated = runStateMigrations(definition, meta.schema_version, rawHead); + // Stored post-commit states ride the same chain so committed + // idempotency keys keep replaying after the migration; every stored + // state sits at `meta.schema_version` (maintained inductively here). + const updateState = transactionDb.prepare('UPDATE agent_state_journal SET state = ? WHERE revision = ?'); + for (const row of rows) { + if (row.state === null) continue; + const rawState = parseStoredJson(definition.id, 'state', row.revision, row.state); + updateState.run(canonicalJson(runStateMigrations(definition, meta.schema_version, rawState)), row.revision); + } const record: AgentStateJournalRecord = { committedAt: this.#now().toISOString(), idempotencyKey: migrationIdempotencyKey(definition.version), diff --git a/packages/rsc-runtime/tests/state-kernel.test.ts b/packages/rsc-runtime/tests/state-kernel.test.ts index af0750884..691f16b74 100644 --- a/packages/rsc-runtime/tests/state-kernel.test.ts +++ b/packages/rsc-runtime/tests/state-kernel.test.ts @@ -6,9 +6,11 @@ import { AGENT_STATE_LIFETIMES, AgentStateError, agentStateLifetimeIsVolatile, + canonicalJson, createAgentStateHandle, createMemoryStateDriver, defineState, + isJsonSafe, type AgentStateDefinition, type AgentStateHandle, type AgentStateLifetime, @@ -35,6 +37,37 @@ const counterDefinition = ( schema: z.object({ count: z.number().int() }).strict(), }); +describe('JSON boundary', () => { + it('rejects sparse arrays as not JSON-safe', () => { + expect(isJsonSafe([1, 2, 3])).toBe(true); + expect(isJsonSafe([])).toBe(true); + // `every` skips holes, so a naive check would declare these safe and + // canonicalization would collapse `[<1 hole>]` to the same text as `[]`. + expect(isJsonSafe(new Array(1))).toBe(false); + expect(isJsonSafe(Object.assign([1, 3], { length: 3 }))).toBe(false); + expect(isJsonSafe({ nested: new Array(2) })).toBe(false); + expect(canonicalJson([])).toBe('[]'); + }); + + it('a sparse array payload is a typed invalid-event, never silently canonicalized', async () => { + const definition = defineState({ + events: { itemsSet: z.any() }, + id: 'state-kernel-test/sparse', + initial: { items: [] as readonly unknown[] }, + lifetime: 'process', + reduce: (_state, event) => ({ items: [event.payload] }), + schema: z.object({ items: z.array(z.unknown()) }).strict(), + }); + const driver = createMemoryStateDriver({ lifetime: 'process' }); + const store = await driver.open(definition); + await expect( + store.dispatch('itemsSet', new Array(1), { idempotencyKey: 'sparse:1' }), + ).rejects.toMatchObject({ code: 'invalid-event', name: 'AgentStateError' }); + expect((await store.read()).revision).toBe(0); + await driver.close(); + }); +}); + describe('defineState', () => { it('rejects invalid definitions with typed invalid-definition errors', () => { const base = { diff --git a/packages/rsc-runtime/tests/state-sqlite.test.ts b/packages/rsc-runtime/tests/state-sqlite.test.ts index f399c6a3d..09a0f4a2c 100644 --- a/packages/rsc-runtime/tests/state-sqlite.test.ts +++ b/packages/rsc-runtime/tests/state-sqlite.test.ts @@ -229,7 +229,7 @@ describe('sqlite driver storage behavior', () => { }); })); - it('fails closed when a persisted head no longer satisfies the schema', () => + it('fails closed on open when a persisted head no longer satisfies the schema', () => withRoot(async (root) => { const file = join(root, 'state.sqlite'); const store = await createSqliteStateDriver({ file }).open(counterDefinition()); @@ -238,12 +238,47 @@ describe('sqlite driver storage behavior', () => { const db = new DatabaseSync(file); db.exec(`UPDATE agent_state_head SET state = '{"wrong":true}'`); db.close(); - const reopened = await createSqliteStateDriver({ file }).open(counterDefinition()); - await expect(reopened.read()).rejects.toMatchObject({ code: 'corrupt' }); - await expect( - reopened.dispatch('bumped', { by: 1 }, { idempotencyKey: 'k2' }), - ).rejects.toMatchObject({ code: 'corrupt' }); - await reopened.close(); + await expect(createSqliteStateDriver({ file }).open(counterDefinition())).rejects.toMatchObject({ + code: 'corrupt', + name: 'AgentStateError', + }); + })); + + it('fails closed when a schema-valid head disagrees with journal replay', () => + withRoot(async (root) => { + const file = join(root, 'state.sqlite'); + const store = await createSqliteStateDriver({ file }).open(counterDefinition()); + await store.dispatch('bumped', { by: 1 }, { idempotencyKey: 'k1' }); + await store.dispatch('bumped', { by: 2 }, { idempotencyKey: 'k2' }); + await store.close(); + const db = new DatabaseSync(file); + // Schema-valid and revision-preserving, so only a replay comparison + // can tell this hand-edited head from the journal's truth. + db.exec(`UPDATE agent_state_head SET state = '{"count":999}'`); + db.close(); + await expect(createSqliteStateDriver({ file }).open(counterDefinition())).rejects.toMatchObject({ + code: 'corrupt', + name: 'AgentStateError', + }); + })); + + it('fails closed when an intermediate journal row is missing', () => + withRoot(async (root) => { + const file = join(root, 'state.sqlite'); + const store = await createSqliteStateDriver({ file }).open(counterDefinition()); + await store.dispatch('bumped', { by: 1 }, { idempotencyKey: 'k1' }); + await store.dispatch('bumped', { by: 2 }, { idempotencyKey: 'k2' }); + await store.dispatch('bumped', { by: 3 }, { idempotencyKey: 'k3' }); + await store.close(); + const db = new DatabaseSync(file); + // The final revision still matches the head; only journal continuity + // catches the deleted row. + db.exec('DELETE FROM agent_state_journal WHERE revision = 2'); + db.close(); + await expect(createSqliteStateDriver({ file }).open(counterDefinition())).rejects.toMatchObject({ + code: 'corrupt', + name: 'AgentStateError', + }); })); it('fails closed on a newer kernel storage format', () =>