Skip to content

fix(state): legacy sqlite adoption, correct migration semantics, atomic memory migration, surfaced close errors #201

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ScriptedAlchemy merged 1 commit into from
Sep 1, 2026
Merged

fix(state): legacy sqlite adoption, correct migration semantics, atomic memory migration, surfaced close errors #201

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
9 changes: 9 additions & 0 deletions .changeset/state-migration-compatibility.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
---
"@agent-bundle/runtime": patch
---

Preserve durable state across the sqlite filename transition, recover legacy
journal results before schema migrations rebase history, keep reset
idempotency inputs unchanged while migrating their committed results, make
in-memory migrations atomic, and surface sqlite close failures on otherwise
successful shutdown.
10 changes: 6 additions & 4 deletions packages/rsc-runtime/src/state/memory-driver.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,7 @@ interface MemoryStoreInternals<TState, TEvents extends AgentStateEventSchemas> {
definition: AgentStateDefinition<TState, TEvents>;
head: AgentStateSnapshot<TState>;
readonly journal: AgentStateJournalRecord[];
readonly keys: Map<string, CommittedResult<TState>>;
keys: Map<string, CommittedResult<TState>>;
}

interface MemoryStoreEntry<TState, TEvents extends AgentStateEventSchemas> {
Expand Down Expand Up @@ -319,17 +319,19 @@ const migrateOpenStore = <TState, TEvents extends AgentStateEventSchemas>(
state: migrated,
toVersion: definition.version,
};
internals.journal.push(record);
const keys = new Map<string, CommittedResult<TState>>();
// 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, {
keys.set(key, {
record: entry.record,
state: runStateMigrations(definition, fromVersion, entry.state),
});
}
internals.keys.set(record.idempotencyKey, { record, state: migrated });
keys.set(record.idempotencyKey, { record, state: migrated });
internals.keys = keys;
internals.journal.push(record);
internals.head = Object.freeze({ revision: record.revision, state: migrated });
internals.definition = definition;
};
Expand Down
129 changes: 100 additions & 29 deletions packages/rsc-runtime/src/state/sqlite.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,9 @@
import { createHash } from 'node:crypto';
import { mkdirSync } from 'node:fs';
import {
existsSync,
mkdirSync,
renameSync,
} from 'node:fs';
import { dirname, join, resolve } from 'node:path';
// node:sqlite emits an ExperimentalWarning on load (documented in the README):
// the module is Node's built-in SQLite binding, stable enough for Node >= 22.13
Expand Down Expand Up @@ -27,6 +31,7 @@ import type {
AgentStateDefinition,
AgentStateDispatchOptions,
AgentStateDriver,
AgentStateEvent,
AgentStateEventSchemas,
AgentStateJournalRecord,
AgentStateReadOptions,
Expand Down Expand Up @@ -197,6 +202,7 @@ interface JournalRow {
readonly kind: string;
readonly name: string | null;
readonly payload: string | null;
readonly result_state: string | null;
readonly revision: number;
readonly state: string | null;
readonly to_version: number | null;
Expand Down Expand Up @@ -230,6 +236,9 @@ const recordFromRow = (definitionId: string, row: JournalRow): AgentStateJournal
const sanitizedFileName = (definitionId: string): string =>
`${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${createHash('sha256').update(definitionId, 'utf8').digest('hex').slice(0, 16)}.sqlite`;

const legacySanitizedFileName = (definitionId: string): string =>
`${definitionId.replace(/[^a-zA-Z0-9._-]+/gu, '-')}-${Buffer.from(definitionId, 'utf8').toString('hex').slice(0, 12)}.sqlite`;

class SqliteConnection extends Context.Service<SqliteConnection, DatabaseSync>()(
'@agent-bundle/runtime/state/SqliteConnection',
) {}
Expand Down Expand Up @@ -341,11 +350,13 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
#committedByKey(
db: DatabaseSync,
key: string,
): { readonly record: AgentStateJournalRecord; readonly stateText: string | null } | undefined {
): { readonly record: AgentStateJournalRecord; readonly resultStateText: string | null } | undefined {
const row = db.prepare('SELECT * FROM agent_state_journal WHERE idempotency_key = ?').get(key) as
| JournalRow
| undefined;
return row === undefined ? undefined : { record: recordFromRow(this.#definition.id, row), stateText: row.state };
return row === undefined
? undefined
: { record: recordFromRow(this.#definition.id, row), resultStateText: row.result_state ?? row.state };
}

/**
Expand All @@ -356,11 +367,11 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
*/
#committedState(
db: DatabaseSync,
committed: { readonly record: AgentStateJournalRecord; readonly stateText: string | null },
committed: { readonly record: AgentStateJournalRecord; readonly resultStateText: string | null },
): TState {
const raw =
committed.stateText !== null
? parseStoredJson(this.#definition.id, 'state', committed.record.revision, committed.stateText)
committed.resultStateText !== null
? parseStoredJson(this.#definition.id, 'result state', committed.record.revision, committed.resultStateText)
: this.#replayTo(db, committed.record.revision);
const parsed = this.#definition.schema.safeParse(raw);
if (!parsed.success) {
Expand All @@ -387,7 +398,7 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
const stateText = canonicalJson(state);
db
.prepare(
'INSERT INTO agent_state_journal (revision, kind, name, payload, state, to_version, idempotency_key, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)',
'INSERT INTO agent_state_journal (revision, kind, name, payload, state, result_state, to_version, idempotency_key, committed_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)',
)
.run(
record.revision,
Expand All @@ -397,6 +408,7 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
// Event rows store their post-commit state too, so idempotent replay
// survives migrations without exact-revision replay.
stateText,
stateText,
record.kind === 'migrate' ? record.toVersion : null,
record.idempotencyKey,
record.committedAt,
Expand All @@ -417,7 +429,7 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
const committedByKey = (db: DatabaseSync, key: string) => this.#committedByKey(db, key);
const committedState = (
db: DatabaseSync,
committed: { readonly record: AgentStateJournalRecord; readonly stateText: string | null },
committed: { readonly record: AgentStateJournalRecord; readonly resultStateText: string | null },
) => this.#committedState(db, committed);
const headState = (db: DatabaseSync) => this.#headState(db, 'commit');
const now = this.#now;
Expand Down Expand Up @@ -601,6 +613,7 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
name TEXT,
payload TEXT,
state TEXT,
result_state TEXT,
to_version INTEGER,
idempotency_key TEXT NOT NULL UNIQUE,
committed_at TEXT NOT NULL
Expand All @@ -611,6 +624,12 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
state TEXT NOT NULL
);
`);
const journalColumns = transactionDb.prepare('PRAGMA table_info(agent_state_journal)').all() as unknown as {
readonly name: string;
}[];
if (!journalColumns.some((column) => column.name === 'result_state')) {
transactionDb.exec('ALTER TABLE agent_state_journal ADD COLUMN result_state TEXT');
}
const definition = this.#definition;
const meta = transactionDb
.prepare('SELECT definition_id, schema_version, kernel_format FROM agent_state_meta WHERE id = 1')
Expand Down Expand Up @@ -678,14 +697,43 @@ class SqliteStore<TState, TEvents extends AgentStateEventSchemas> implements Age
}
}
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);
// Journal records retain the original commit input for dedupe. Their
// committed results migrate separately, matching the memory driver's
// `{ record, state }` split. Legacy event rows without a result are
// replayed before the new migration baseline makes old revisions
// unavailable.
const updateResult = transactionDb.prepare(
'UPDATE agent_state_journal SET result_state = ? WHERE revision = ?',
);
let replayState: unknown = definition.initial;
for (const [index, row] of rows.entries()) {
const record = records[index] as AgentStateJournalRecord;
const storedResultText = row.result_state ?? row.state;
let migratedResult: TState;
if (storedResultText !== null) {
replayState = parseStoredJson(definition.id, 'result state', row.revision, storedResultText);
migratedResult = runStateMigrations(definition, meta.schema_version, replayState);
} else if (record.kind === 'event') {
try {
replayState = definition.reduce(
replayState as TState,
{ name: record.name, payload: record.payload } as AgentStateEvent<TEvents>,
Comment on lines +718 to +720

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P1 Badge Do not replay legacy rows with the current reducer

When a migration changes the state shape, reducer behavior, or event schema, these legacy rows were produced by the old reducer but are reconstructed with definition.reduce from the new definition. For example, a v1 { count } state migrated to v2 { value } will either fail while opening or produce a result that never existed, so retrying an old idempotency key returns incorrect state. Legacy post-commit results cannot safely be inferred with current-version reducer semantics; they need version-appropriate recovery or a fail-closed compatibility strategy.

Useful? React with 👍 / 👎.

@ScriptedAlchemy ScriptedAlchemy Sep 1, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in #213 (merged as 941aa08). Legacy event rows with no stored committed result are no longer replayed through the current-version reducer during a schema migration: the journal-head row's result is recovered from the authoritative materialized head state, and earlier unrecoverable rows now fail closed with a typed migration-failure error instructing the operator to materialize results under the old definition (or restore a backup) before migrating. Regression tests cover both the fail-closed path and a shape-changing v1→v2 migration replaying an old idempotency key from preserved state.

);
} catch (error) {
throw new AgentStateError(
'migration-failure',
`State '${definition.id}' could not recover legacy result at revision ${String(record.revision)}`,
{ cause: error },
);
}
migratedResult = runStateMigrations(definition, meta.schema_version, replayState);
} else {
throw new AgentStateError(
'corrupt',
`State '${definition.id}' journal row at revision ${String(record.revision)} has no committed result`,
);
}
updateResult.run(canonicalJson(migratedResult), row.revision);
}
const record: AgentStateJournalRecord = {
committedAt: this.#now().toISOString(),
Expand Down Expand Up @@ -763,25 +811,48 @@ export const createSqliteStateDriver = (options: SqliteStateDriverOptions): Agen
`State '${definition.id}' declares lifetime '${definition.lifetime}' but this driver provides 'workspace-durable'`,
);
}
return resolve(
options.file !== undefined ? options.file : join(options.root as string, sanitizedFileName(definition.id)),
);
}),
if (options.file !== undefined) return resolve(options.file);
const root = options.root as string;
const currentFile = resolve(join(root, sanitizedFileName(definition.id)));
const legacyFile = resolve(join(root, legacySanitizedFileName(definition.id)));
mkdirSync(dirname(currentFile), { recursive: true });
if (!existsSync(currentFile) && existsSync(legacyFile)) {
for (const suffix of ['-wal', '-shm']) {
const legacySidecar = `${legacyFile}${suffix}`;
if (!existsSync(legacySidecar)) continue;
try {
renameSync(legacySidecar, `${currentFile}${suffix}`);
} catch (error) {
// A concurrent adopter may have moved this sidecar after
// the existence check. Other failures must remain visible.
if ((error as SqliteErrorShape).code !== 'ENOENT') throw error;
}
}
try {
renameSync(legacyFile, currentFile);
} catch (error) {
// Another opener may have atomically adopted the same
// legacy file after both observed it. The winner's current
// path is authoritative; otherwise preserve the failure.
if (!existsSync(currentFile)) throw error;
}
}
return currentFile;
}, true),
);
const connection = Effect.acquireRelease(
sqliteEffect(definition.id, 'open database', () => {
mkdirSync(dirname(file), { recursive: true });
return new DatabaseSync(file);
}, true),
(db) =>
Effect.sync(() => {
try {
db.close();
} catch {
// Closing an already-broken connection must not mask the
// caller's path (the original failure carries the cause).
}
}),
(db, exit) => {
const close = sqliteEffect(definition.id, 'close database', () => {
db.close();
}, true);
return Exit.isFailure(exit)
? close.pipe(Effect.catch(() => Effect.void))
: close;
Comment on lines +852 to +854

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Preserve initialization errors when close also fails

If initialize() fails and db.close() also throws, the exit received here is the managed layer scope's disposal exit, not the earlier initialization exit, because initialization runs separately through runtime.run(). Consequently runtime.close() at line 868 takes the unsuppressed branch and its close error replaces the original corruption or migration error, contrary to the intended “do not mask the caller's failure” behavior.

Useful? React with 👍 / 👎.

@ScriptedAlchemy ScriptedAlchemy Sep 1, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in #213 (merged as 941aa08). The open() initialization-failure path now suppresses a runtime.close() failure and rethrows the original initialization error, so a corruption/migration error is no longer masked by a subsequent close failure. The #208 finalizer contract (infallible via orDie on the success path) is unchanged. Regression test: preserves the initialization error when database close also fails.

@chatgpt-codex-connector chatgpt-codex-connector Bot Sep 1, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

P2 Badge Close every store before propagating a close failure

Now that this finalizer can reject, closing a root-backed driver with multiple open stores stops at the first failing store.close() in the driver loop at line 790. Later stores are never closed, remain operable with live database connections, and cannot be retried because the driver caches the rejected closing promise. The driver should attempt all store closures and propagate the collected/first failure only afterward.

Useful? React with 👍 / 👎.

@ScriptedAlchemy ScriptedAlchemy Sep 1, 2026

Copy link
Copy Markdown
Owner Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Fixed in #213 (merged as 941aa08). Driver close() now attempts every open store's closure, collecting failures, and only then propagates the first failure — no store is left with a live connection after a partial-close. Regression test: attempts every store close before propagating the first close failure (asserts both stores closed and both read as store-closed afterwards).

},
);
const runtime = makeScopedEffectRuntime(
Layer.effect(SqliteConnection, connection),
Expand Down
37 changes: 37 additions & 0 deletions packages/rsc-runtime/tests/state-kernel.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -361,6 +361,43 @@ describe('explicit migrations', () => {
});
});

it('leaves the process store unchanged when a historical result migration throws', async () => {
const driver = createMemoryStateDriver();
const storeV1 = await driver.open(v1());
await storeV1.dispatch('incremented', { by: 1 }, { idempotencyKey: 'i1' });
await storeV1.dispatch('incremented', { by: 2 }, { idempotencyKey: 'i2' });
await storeV1.dispatch('incremented', { by: 3 }, { idempotencyKey: 'i3' });

await expect(
driver.open(
v2((persisted) => {
const state = persisted as CounterState;
if (state.count === 3) throw new Error('cannot migrate historical result');
return { ...state, unit: 'edits' };
}),
),
).rejects.toMatchObject({ code: 'migration-failure' });

expect(await storeV1.read()).toEqual({ revision: 3, state: { count: 6 } });
expect((await storeV1.changes({ afterRevision: 0 })).changes.map((change) => change.kind)).toEqual([
'event',
'event',
'event',
]);
await expect(
storeV1.dispatch('incremented', { by: 1 }, { idempotencyKey: 'i1' }),
).resolves.toEqual({ replayed: true, revision: 1, state: { count: 1 } });

const storeV2 = await driver.open(v2());
expect(await storeV2.read()).toEqual({ revision: 4, state: { count: 6, unit: 'edits' } });
expect((await storeV2.changes({ afterRevision: 0 })).changes.map((change) => change.kind)).toEqual([
'event',
'event',
'event',
'migrate',
]);
});

it('rejects opening a persisted-newer store with an older definition', async () => {
const driver = createMemoryStateDriver();
await driver.open(v2());
Expand Down
Loading
Loading