fix(state): review follow-ups from #142/#149 — idempotency ordering, head verification, replay across migrations (#171)

Codex review follow-ups on the state kernel and drivers, re-verified
against the Effect-native internals from #154:

- Both drivers consult the idempotency key before running the reducer
  (#142/#149 P1): a committed key replays its stored result even when the
  reducer would now fail against the current head. Payload validation and
  canonicalization moved ahead of the key lookup; the reducer runs only on
  the append path.
- Committed results replay across migrations (#142 P1): every journal
  record stores its post-commit state (event rows too, in the sqlite
  driver's existing nullable column), and migrations run the same chain
  over the stored results, so replay no longer depends on exact-revision
  history that migrations rebase. Legacy event rows without a stored state
  fall back to journal replay.
- The sqlite driver verifies storage on open (#149 P1): journal continuity
  via expectConsistentJournal (a deleted intermediate row fails closed) and
  the materialized head against journal replay (a schema-valid hand-edited
  head fails closed); with a pending migration the head is checked against
  the last stored post-commit state instead.
- sanitizedFileName hashes the complete definition id (sha-256) instead of
  hex-encoding its first six bytes (#149 P2), so ids sharing a sanitized
  prefix get isolated database files.
- isJsonSafe rejects sparse arrays (#142 P2): holes no longer canonicalize
  like dense arrays under permissive schemas.

The conformance suite pins the corrected semantics for both drivers:
replay-before-reduce, replay across migrations, and prefix-colliding id
isolation; sqlite corruption tests cover the two new fail-closed opens.

Author: ScriptedAlchemy

.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.

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) => {

packages/rsc-runtime/src/state/contract.ts

@@ -105,8 +105,12 @@ export type AgentStateEvent<TEvents extends AgentStateEventSchemas> = {
  * 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<Record<number, (persisted: unknown) => unknown>>;
 

packages/rsc-runtime/src/state/index.ts

@@ -46,6 +46,8 @@ export {
   changeFromJournalRecord,
   expectConsistentJournal,
   migrationIdempotencyKey,
+  parseEventPayload,
+  reduceStateEvent,
   replayJournal,
   resolveResetState,
   runStateMigrations,

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 = <TState, TEvents extends AgentStateEventSchemas>(
+export const parseEventPayload = <TState, TEvents extends AgentStateEventSchemas>(
   definition: AgentStateDefinition<TState, TEvents>,
-  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 = <TState, TEvents extends AgentStateEventSchemas>(
   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 = <TState, TEvents extends AgentStateEventSchemas>(
+  definition: AgentStateDefinition<TState, TEvents>,
+  state: TState,
+  name: string,
+  payload: unknown,
+): TState => {
   let next: TState;
   try {
-    next = definition.reduce(state, { name, payload: parsed.data } as AgentStateEvent<TEvents>);
+    next = definition.reduce(state, { name, payload } as AgentStateEvent<TEvents>);
   } catch (error) {
     throw new AgentStateError(
       'reducer-failure',
@@ -110,7 +124,22 @@ export const applyStateEvent = <TState, TEvents extends AgentStateEventSchemas>(
   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 = <TState, TEvents extends AgentStateEventSchemas>(
+  definition: AgentStateDefinition<TState, TEvents>,
+  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. */

packages/rsc-runtime/src/state/json.ts

@@ -16,7 +16,15 @@ const isPlainObject = (value: unknown): value is Readonly<Record<string, unknown
 export const isJsonSafe = (value: unknown): boolean => {
   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);
 };
 

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<TState> {
+  readonly record: AgentStateJournalRecord;
+  /** Post-commit state, kept per key so replay survives history rebases. */
+  readonly state: TState;
+}
+
 interface MemoryStoreInternals<TState, TEvents extends AgentStateEventSchemas> {
   closed: boolean;
   definition: AgentStateDefinition<TState, TEvents>;
   head: AgentStateSnapshot<TState>;
   readonly journal: AgentStateJournalRecord[];
-  readonly keys: Map<string, AgentStateJournalRecord>;
+  readonly keys: Map<string, CommittedResult<TState>>;
 }
 
 interface MemoryStoreEntry<TState, TEvents extends AgentStateEventSchemas> {
@@ -133,59 +140,63 @@ const createMemoryStore = <TState, TEvents extends AgentStateEventSchemas>(
 
   /**
    * 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<TState> => {
-    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(
         'revision-conflict',
         `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<TState, TEvents> = {
@@ -227,19 +238,18 @@ const createMemoryStore = <TState, TEvents extends AgentStateEventSchemas>(
           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 = <TState, TEvents extends AgentStateEventSchemas>(
   definition: AgentStateDefinition<TState, TEvents>,
   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 = <TState, TEvents extends AgentStateEventSchemas>(
     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;
 };