Add a stamped boot-time data-migration ledger for the libSQL apps

Before this, every boot-time data fix (auth-config placements, openapi envelope unwrap) re-scanned its tables on every startup to decide whether it had already run, inferring state from data shape. That accumulates — N migrations means N full-table scans per boot forever — and makes idempotence a per-migration proof obligation. Now the sdk owns a data_migration ledger (name -> completion time) and an Effect-native runner: apps compose an ordered registry of named migrations; pending ones run once and are stamped, stamped ones are skipped without touching data. Failures are typed (DataMigrationError / DuplicateDataMigrationError) and fail the boot without stamping. apps/local and apps/host-selfhost convert their two existing boot scans into the first registry entries. The openapi unwrap migration is rewritten Effect-native with the same envelope detection. Registry names are append-only and never renamed. Verified end-to-end against a real libSQL database: first boot applies both migrations and unwraps a seeded envelope row (payload rows untouched), second boot applies nothing, ledger holds both stamps. Also fixes a cloud test still asserting the retired envelope shape (missed in the payload-first change; the turbo test pipeline did not propagate that failure).

Rhys Sullivan committed 5 months ago

Commit bea88351 parent fed315110 files · +476 −130Browse files

Unified · Split

10 files

apps/host-selfhost/src/app.ts +7 −13
apps/host-selfhost/src/db/data-migrations.ts +23 −0
apps/local/src/db/data-migrations.ts +23 −0
apps/local/src/executor.ts +17 −22
packages/core/sdk/src/index.ts +10 −0
packages/core/sdk/src/sqlite-data-migrations.test.ts +130 −0
packages/core/sdk/src/sqlite-data-migrations.ts +126 −0
packages/plugins/openapi/src/sdk/index.ts +1 −1
packages/plugins/openapi/src/sdk/output-schema-migration.test.ts +71 −34
packages/plugins/openapi/src/sdk/output-schema-migration.ts +68 −60

apps/host-selfhost/src/app.ts+7−13

@@ -1,14 +1,13 @@
1	1	import { HttpApiSwagger } from "effect/unstable/httpapi";
2	2	import { HttpEffect, HttpRouter } from "effect/unstable/http";
3		−import { Layer } from "effect";
	3	+import { Effect, Layer } from "effect";
4	4
5	5	import { composePluginApi, ExecutorApp, textFailureStrategy } from "@executor-js/api/server";
6	6
7		−import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth";
8		−import { runSqliteOpenApiOutputSchemaMigration } from "@executor-js/plugin-openapi";
	7	+import { runSqliteDataMigrations } from "@executor-js/sdk";
9	8
10	9	import { resolveAuthProviders } from "./auth";
11		−import { authConfigTransforms } from "./db/auth-config-migration";
	10	+import { selfHostDataMigrations } from "./db/data-migrations";
12	11	import { makeSelfHostAdminApiLayer } from "./admin/handlers";
13	12	import { makeSelfHostSystemApiLayer } from "./system/handlers";
14	13	import { selfHostAccountMiddleware } from "./account";
@@ -58,15 +57,10 @@ export const makeSelfHostApp = async (options: MakeSelfHostAppOptions = {}) => { · 43 unmodified lines
58	57	version: SELF_HOST_SCHEMA_VERSION,
59	58	});
60	59
61		− // One-off data migration: rewrite pre-canonical integration auth configs
62		− // into the shared placements model. Idempotent — a no-op once every row is
63		− // canonical (same defensive-at-boot pattern as the column adds above).
64		− await runSqliteAuthConfigMigration(dbHandle.client, authConfigTransforms);
65		−
66		− // One-off data migration: unwrap the retired {status, headers, data}
67		− // transport envelope from persisted openapi tool output schemas (mirrors
68		− // cloud's drizzle 0002). Idempotent — payload-shaped rows don't match.
69		− await runSqliteOpenApiOutputSchemaMigration(dbHandle.client);
	60	+ // Boot-time data migrations: each registry entry runs once and is stamped
	61	+ // in the `data_migration` ledger; stamped entries are skipped without
	62	+ // touching the data.
	63	+ await Effect.runPromise(runSqliteDataMigrations(dbHandle.client, selfHostDataMigrations));
70	64
71	65	// ---- auth providers ---------------------------------------------------
72	66	// Better Auth: cookie/bearer/api-key identity + /api/auth handler + account

apps/host-selfhost/src/db/data-migrations.ts+23−0

@@ -0,0 +1,23 @@
	1	+// ---------------------------------------------------------------------------
	2	+// The ordered boot-time data-migration registry for the selfhost app.
	3	+// Entries run once and are stamped in the `data_migration` ledger (see
	4	+// @executor-js/sdk sqlite-data-migrations). Names are append-only and never
	5	+// renamed.
	6	+// ---------------------------------------------------------------------------
	7	+
	8	+import { sqliteDataMigration, type SqliteDataMigration } from "@executor-js/sdk";
	9	+import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth";
	10	+import { openApiOutputSchemaDataMigration } from "@executor-js/plugin-openapi";
	11	+
	12	+import { authConfigTransforms } from "./auth-config-migration";
	13	+
	14	+export const selfHostDataMigrations: readonly SqliteDataMigration[] = [
	15	+ // Rewrite pre-canonical integration auth configs into the shared
	16	+ // placements model.
	17	+ sqliteDataMigration("2026-06-05-auth-config-placements", (client) =>
	18	+ runSqliteAuthConfigMigration(client, authConfigTransforms),
	19	+ ),
	20	+ // Unwrap the retired {status, headers, data} transport envelope from
	21	+ // persisted openapi tool output schemas (mirrors cloud's drizzle 0002).
	22	+ openApiOutputSchemaDataMigration,
	23	+];

apps/local/src/db/data-migrations.ts+23−0

@@ -0,0 +1,23 @@
	1	+// ---------------------------------------------------------------------------
	2	+// The ordered boot-time data-migration registry for the local app. Entries
	3	+// run once and are stamped in the `data_migration` ledger (see
	4	+// @executor-js/sdk sqlite-data-migrations). Names are append-only and never
	5	+// renamed.
	6	+// ---------------------------------------------------------------------------
	7	+
	8	+import { sqliteDataMigration, type SqliteDataMigration } from "@executor-js/sdk";
	9	+import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth";
	10	+import { openApiOutputSchemaDataMigration } from "@executor-js/plugin-openapi";
	11	+
	12	+import { authConfigTransforms } from "./auth-config-migration";
	13	+
	14	+export const localDataMigrations: readonly SqliteDataMigration[] = [
	15	+ // Rewrite pre-canonical integration auth configs (incl. v1→v2 outputs)
	16	+ // into the shared placements model.
	17	+ sqliteDataMigration("2026-06-05-auth-config-placements", (client) =>
	18	+ runSqliteAuthConfigMigration(client, authConfigTransforms),
	19	+ ),
	20	+ // Unwrap the retired {status, headers, data} transport envelope from
	21	+ // persisted openapi tool output schemas (mirrors cloud's drizzle 0002).
	22	+ openApiOutputSchemaDataMigration,
	23	+];

apps/local/src/executor.ts+17−22

@@ -4,14 +4,19 @@ import { homedir } from "node:os"; · 3 unmodified lines
4	4	import { basename, join } from "node:path";
5	5	import { createHash } from "node:crypto";
6	6
7		−import { Subject, Tenant, createExecutor, type AnyPlugin, type Executor } from "@executor-js/sdk";
8		−import { runSqliteAuthConfigMigration } from "@executor-js/sdk/http-auth";
9		−import { runSqliteOpenApiOutputSchemaMigration } from "@executor-js/plugin-openapi";
	7	+import {
	8	+ Subject,
	9	+ Tenant,
	10	+ createExecutor,
	11	+ runSqliteDataMigrations,
	12	+ type AnyPlugin,
	13	+ type Executor,
	14	+} from "@executor-js/sdk";
10	15	import { collectTables } from "@executor-js/api/server";
11	16	import { loadPluginsFromJsonc } from "@executor-js/config";
12	17
13	18	import executorConfig from "../executor.config";
14		−import { authConfigTransforms } from "./db/auth-config-migration";
	19	+import { localDataMigrations } from "./db/data-migrations";
15	20	import { createSqliteFumaDb } from "./db/sqlite-fumadb";
16	21	import { migrateLocalV1ToV2IfNeeded } from "./db/v1-v2-migration";
17	22
@@ -170,24 +175,14 @@ const createLocalExecutorLayer = () => { · 152 unmodified lines
170	175	(db) => Effect.promise(() => db.close()).pipe(Effect.ignore),
171	176	);
172	177
173		− // One-off data migration: rewrite pre-canonical integration auth
174		− // configs (incl. v1→v2 outputs) into the shared placements model.
175		− // Idempotent — a no-op once every row is canonical.
176		− yield* Effect.tryPromise({
177		− try: () => runSqliteAuthConfigMigration(sqlite.client, authConfigTransforms),
178		− catch: (cause) =>
179		− new LocalExecutorCreateError({ message: CREATE_SQLITE_ERROR_MESSAGE, cause }),
180		− });
181		−
182		− // One-off data migration: unwrap the retired {status, headers, data}
183		− // transport envelope from persisted openapi tool output schemas
184		− // (mirrors cloud's drizzle 0002). Idempotent — payload-shaped rows
185		− // don't match the envelope signature.
186		− yield* Effect.tryPromise({
187		− try: () => runSqliteOpenApiOutputSchemaMigration(sqlite.client),
188		− catch: (cause) =>
189		− new LocalExecutorCreateError({ message: CREATE_SQLITE_ERROR_MESSAGE, cause }),
190		− });
	178	+ // Boot-time data migrations: each registry entry runs once and is
	179	+ // stamped in the `data_migration` ledger; stamped entries are skipped
	180	+ // without touching the data.
	181	+ yield* runSqliteDataMigrations(sqlite.client, localDataMigrations).pipe(
	182	+ Effect.mapError(
	183	+ (cause) => new LocalExecutorCreateError({ message: CREATE_SQLITE_ERROR_MESSAGE, cause }),
	184	+ ),
	185	+ );
191	186
192	187	// webBaseUrl is where the executor's web UI listens — same port as the
193	188	// daemon API since the daemon serves both. Mirrors serve.ts's port

packages/core/sdk/src/index.ts+10−0

@@ -324,6 +324,16 @@ export { InternalError } from "./api-errors"; · 323 unmodified lines
324	324
325	325	// ToolResult — typed value-based discriminated union for tool outcomes.
326	326	export { ToolResult, isToolResult, type ToolError, type ToolHttpMeta } from "./tool-result";
	327	+
	328	+// Stamped boot-time data-migration ledger for the libSQL-backed apps.
	329	+export {
	330	+ DataMigrationError,
	331	+ DuplicateDataMigrationError,
	332	+ runSqliteDataMigrations,
	333	+ sqliteDataMigration,
	334	+ type SqliteDataMigration,
	335	+ type SqliteDataMigrationClient,
	336	+} from "./sqlite-data-migrations";
327	337	export {
328	338	authToolFailure,
329	339	type AuthToolFailureCode,

packages/core/sdk/src/sqlite-data-migrations.test.ts+130−0

@@ -0,0 +1,130 @@
	1	+import { describe, expect, it } from "@effect/vitest";
	2	+import { Effect, Predicate } from "effect";
	3	+
	4	+import {
	5	+ DataMigrationError,
	6	+ runSqliteDataMigrations,
	7	+ sqliteDataMigration,
	8	+ type SqliteDataMigration,
	9	+ type SqliteDataMigrationClient,
	10	+} from "./sqlite-data-migrations";
	11	+
	12	+// A tiny scripted fake standing in for a libSQL client: tracks statements
	13	+// and simulates the ledger table.
	14	+const makeFakeClient = (stampedNames: string[]) => {
	15	+ const log: unknown[] = [];
	16	+ const stamps = [...stampedNames];
	17	+ const client: SqliteDataMigrationClient = {
	18	+ execute: (stmt) => {
	19	+ log.push(stmt);
	20	+ if (typeof stmt === "string" && stmt.startsWith("SELECT name FROM data_migration")) {
	21	+ return Promise.resolve({ rows: stamps.map((name) => ({ name })) });
	22	+ }
	23	+ if (typeof stmt === "object" && stmt.sql.startsWith("INSERT INTO data_migration")) {
	24	+ stamps.push(String(stmt.args[0]));
	25	+ }
	26	+ return Promise.resolve({ rows: [] });
	27	+ },
	28	+ };
	29	+ return { client, log, stamps };
	30	+};
	31	+
	32	+const migrationSpy = (name: string) => {
	33	+ const calls: number[] = [];
	34	+ const migration: SqliteDataMigration = {
	35	+ name,
	36	+ run: () => Effect.sync(() => void calls.push(calls.length)),
	37	+ };
	38	+ return { migration, calls };
	39	+};
	40	+
	41	+describe("runSqliteDataMigrations", () => {
	42	+ it.effect("runs pending migrations in order and stamps them", () =>
	43	+ Effect.gen(function* () {
	44	+ const { client, log, stamps } = makeFakeClient([]);
	45	+ const a = migrationSpy("2026-06-05-a");
	46	+ const b = migrationSpy("2026-06-11-b");
	47	+ const applied = yield* runSqliteDataMigrations(client, [a.migration, b.migration]);
	48	+ expect(applied).toEqual(["2026-06-05-a", "2026-06-11-b"]);
	49	+ expect(a.calls.length).toBe(1);
	50	+ expect(b.calls.length).toBe(1);
	51	+ expect(stamps).toEqual(["2026-06-05-a", "2026-06-11-b"]);
	52	+ expect(String(log[0])).toContain("CREATE TABLE IF NOT EXISTS data_migration");
	53	+ }),
	54	+ );
	55	+
	56	+ it.effect("skips stamped migrations without running them", () =>
	57	+ Effect.gen(function* () {
	58	+ const { client } = makeFakeClient(["2026-06-05-a"]);
	59	+ const a = migrationSpy("2026-06-05-a");
	60	+ const b = migrationSpy("2026-06-11-b");
	61	+ const applied = yield* runSqliteDataMigrations(client, [a.migration, b.migration]);
	62	+ expect(applied).toEqual(["2026-06-11-b"]);
	63	+ expect(a.calls.length).toBe(0);
	64	+ expect(b.calls.length).toBe(1);
	65	+ }),
	66	+ );
	67	+
	68	+ it.effect("is a no-op once everything is stamped", () =>
	69	+ Effect.gen(function* () {
	70	+ const { client } = makeFakeClient([]);
	71	+ const a = migrationSpy("2026-06-05-a");
	72	+ yield* runSqliteDataMigrations(client, [a.migration]);
	73	+ const applied = yield* runSqliteDataMigrations(client, [a.migration]);
	74	+ expect(applied).toEqual([]);
	75	+ expect(a.calls.length).toBe(1);
	76	+ }),
	77	+ );
	78	+
	79	+ it.effect("a failing migration leaves no stamp and surfaces the failure", () =>
	80	+ Effect.gen(function* () {
	81	+ const { client, stamps } = makeFakeClient([]);
	82	+ const boom: SqliteDataMigration = {
	83	+ name: "2026-06-11-boom",
	84	+ run: () =>
	85	+ Effect.fail(new DataMigrationError({ migration: "2026-06-11-boom", cause: "nope" })),
	86	+ };
	87	+ const after = migrationSpy("2026-06-12-after");
	88	+ const failure = yield* runSqliteDataMigrations(client, [boom, after.migration]).pipe(
	89	+ Effect.flip,
	90	+ );
	91	+ expect(Predicate.isTagged(failure, "DataMigrationError")).toBe(true);
	92	+ expect(stamps).toEqual([]);
	93	+ expect(after.calls.length).toBe(0);
	94	+ }),
	95	+ );
	96	+
	97	+ it.effect("rejects duplicate names before touching the database", () =>
	98	+ Effect.gen(function* () {
	99	+ const { client, log } = makeFakeClient([]);
	100	+ const a1 = migrationSpy("2026-06-05-a");
	101	+ const a2 = migrationSpy("2026-06-05-a");
	102	+ const failure = yield* runSqliteDataMigrations(client, [a1.migration, a2.migration]).pipe(
	103	+ Effect.flip,
	104	+ );
	105	+ expect(Predicate.isTagged(failure, "DuplicateDataMigrationError")).toBe(true);
	106	+ expect(log).toEqual([]);
	107	+ }),
	108	+ );
	109	+
	110	+ it.effect("sqliteDataMigration adapts promise-shaped bodies with typed failures", () =>
	111	+ Effect.gen(function* () {
	112	+ const { client, stamps } = makeFakeClient([]);
	113	+ let ran = 0;
	114	+ const ok = sqliteDataMigration("2026-06-05-promise", () => {
	115	+ ran++;
	116	+ return Promise.resolve(42);
	117	+ });
	118	+ yield* runSqliteDataMigrations(client, [ok]);
	119	+ expect(ran).toBe(1);
	120	+ expect(stamps).toEqual(["2026-06-05-promise"]);
	121	+
	122	+ // oxlint-disable-next-line executor/no-promise-reject -- simulates a raw driver rejection at the adapter boundary under test
	123	+ const bad = sqliteDataMigration("2026-06-12-bad", () => Promise.reject("disk full"));
	124	+ const failure = yield* runSqliteDataMigrations(client, [ok, bad]).pipe(Effect.flip);
	125	+ expect(Predicate.isTagged(failure, "DataMigrationError")).toBe(true);
	126	+ expect((failure as DataMigrationError).migration).toBe("2026-06-12-bad");
	127	+ expect(ran).toBe(1);
	128	+ }),
	129	+ );
	130	+});

packages/core/sdk/src/sqlite-data-migrations.ts+126−0

@@ -0,0 +1,126 @@
	1	+// ---------------------------------------------------------------------------
	2	+// Stamped data-migration ledger for the libSQL-backed apps (local boot,
	3	+// selfhost boot). Cloud runs schema + data migrations through its drizzle
	4	+// chain out-of-band; the local apps have no operator, so their migrations
	5	+// run at boot — and before this ledger existed, each one re-scanned its
	6	+// tables on every startup to decide "did I already run?" by data shape.
	7	+// That accumulates (N migrations = N full-table scans per boot, forever)
	8	+// and makes idempotence a per-migration proof obligation.
	9	+//
	10	+// This is the rail instead: a `data_migration` table (name → completion
	11	+// time), an ordered registry the app composes, and a runner that executes
	12	+// each pending migration once and stamps it. Stamped names are skipped
	13	+// without touching the data.
	14	+//
	15	+// Write migrations idempotently anyway (defense in depth — the support
	16	+// remedy for a half-applied state is deleting the stamp row and
	17	+// rebooting). One deliberate semantics change from the scan-every-boot
	18	+// era: after a migration is stamped, rows written later by an OLDER binary
	19	+// (downgrade, then re-upgrade) are NOT re-healed. That matches the cloud
	20	+// chain's semantics; the stamp row, not the data shape, is the source of
	21	+// truth.
	22	+// ---------------------------------------------------------------------------
	23	+
	24	+import { Data, Effect } from "effect";
	25	+
	26	+/** Structural client interface so this module stays dependency-free;
	27	+ * `@libsql/client` satisfies it. */
	28	+export interface SqliteDataMigrationClient {
	29	+ execute(
	30	+ stmt: string \| { readonly sql: string; readonly args: readonly unknown[] },
	31	+ ): Promise<{ readonly rows: readonly Record<string, unknown>[] }>;
	32	+}
	33	+
	34	+export class DataMigrationError extends Data.TaggedError("DataMigrationError")<{
	35	+ /** The migration that failed, or null when the ledger itself did. */
	36	+ readonly migration: string \| null;
	37	+ readonly cause: unknown;
	38	+}> {}
	39	+
	40	+export class DuplicateDataMigrationError extends Data.TaggedError("DuplicateDataMigrationError")<{
	41	+ readonly name: string;
	42	+}> {}
	43	+
	44	+export interface SqliteDataMigration {
	45	+ /** Stable unique id, date-prefixed so the registry reads in order
	46	+ * (e.g. "2026-06-05-auth-config-placements"). Renaming an applied
	47	+ * migration re-runs it — never rename. */
	48	+ readonly name: string;
	49	+ readonly run: (client: SqliteDataMigrationClient) => Effect.Effect<void, DataMigrationError>;
	50	+}
	51	+
	52	+const LEDGER_TABLE = "data_migration";
	53	+
	54	+const execute = (
	55	+ client: SqliteDataMigrationClient,
	56	+ stmt: string \| { readonly sql: string; readonly args: readonly unknown[] },
	57	+ migration: string \| null,
	58	+) =>
	59	+ Effect.tryPromise({
	60	+ try: () => client.execute(stmt),
	61	+ catch: (cause) => new DataMigrationError({ migration, cause }),
	62	+ });
	63	+
	64	+/** Wrap a promise-shaped migration body as a registry entry. */
	65	+export const sqliteDataMigration = (
	66	+ name: string,
	67	+ run: (client: SqliteDataMigrationClient) => Promise<unknown>,
	68	+): SqliteDataMigration => ({
	69	+ name,
	70	+ run: (client) =>
	71	+ Effect.tryPromise({
	72	+ try: () => run(client),
	73	+ catch: (cause) => new DataMigrationError({ migration: name, cause }),
	74	+ }).pipe(Effect.asVoid),
	75	+});
	76	+
	77	+/**
	78	+ * Run every registry entry whose name has no stamp row, in registry order,
	79	+ * stamping each on success. Returns the applied names.
	80	+ *
	81	+ * Atomicity is the migration's own job (the existing migrations run their
	82	+ * rewrites inside BEGIN…COMMIT), so the runner does not wrap them — SQLite
	83	+ * has no nested transactions. The stamp is written after the migration
	84	+ * succeeds; a crash between the two re-runs the (idempotent) migration on
	85	+ * the next boot, which is a no-op. A failed migration leaves no stamp and
	86	+ * fails the boot.
	87	+ */
	88	+export const runSqliteDataMigrations = (
	89	+ client: SqliteDataMigrationClient,
	90	+ migrations: readonly SqliteDataMigration[],
	91	+): Effect.Effect<readonly string[], DataMigrationError \| DuplicateDataMigrationError> =>
	92	+ Effect.gen(function* () {
	93	+ const names = new Set<string>();
	94	+ for (const migration of migrations) {
	95	+ if (names.has(migration.name)) {
	96	+ return yield* new DuplicateDataMigrationError({ name: migration.name });
	97	+ }
	98	+ names.add(migration.name);
	99	+ }
	100	+
	101	+ yield* execute(
	102	+ client,
	103	+ `CREATE TABLE IF NOT EXISTS ${LEDGER_TABLE} (name TEXT PRIMARY KEY, time_completed INTEGER NOT NULL)`,
	104	+ null,
	105	+ );
	106	+ const stamped = yield* execute(client, `SELECT name FROM ${LEDGER_TABLE}`, null);
	107	+ const completed = new Set(
	108	+ stamped.rows.map((row) => row.name).filter((name) => typeof name === "string"),
	109	+ );
	110	+
	111	+ const applied: string[] = [];
	112	+ for (const migration of migrations) {
	113	+ if (completed.has(migration.name)) continue;
	114	+ yield* migration.run(client);
	115	+ yield* execute(
	116	+ client,
	117	+ {
	118	+ sql: `INSERT INTO ${LEDGER_TABLE} (name, time_completed) VALUES (?, ?)`,
	119	+ args: [migration.name, Date.now()],
	120	+ },
	121	+ migration.name,
	122	+ );
	123	+ applied.push(migration.name);
	124	+ }
	125	+ return applied;
	126	+ });

packages/plugins/openapi/src/sdk/index.ts+1−1

@@ -79,7 +79,7 @@ export { variable, type ApiKeyAuthTemplate } from "@executor-js/sdk/http-auth"; · 78 unmodified lines
79	79	export { migrateOpenApiAuthConfig } from "./migrate-config";
80	80
81	81	export {
	82	+ openApiOutputSchemaDataMigration,
82	83	runSqliteOpenApiOutputSchemaMigration,
83	84	unwrapOpenApiTransportEnvelope,
84		− type SqliteToolSchemaClient,
85	85	} from "./output-schema-migration";

packages/plugins/openapi/src/sdk/output-schema-migration.test.ts+71−34

@@ -1,9 +1,10 @@
1	1	import { describe, expect, it } from "@effect/vitest";
	2	+import { Effect, Predicate } from "effect";
	3	+import type { SqliteDataMigrationClient } from "@executor-js/sdk/core";
2	4
3	5	import {
4	6	runSqliteOpenApiOutputSchemaMigration,
5	7	unwrapOpenApiTransportEnvelope,
6		− type SqliteToolSchemaClient,
7	8	} from "./output-schema-migration";
8	9
9	10	// The exact shape openApiTransportOutputSchema used to emit.
@@ -50,7 +51,7 @@ describe("unwrapOpenApiTransportEnvelope", () => { · 40 unmodified lines
50	51	// A tiny scripted fake standing in for a libSQL client.
51	52	const makeFakeClient = (rows: Record<string, unknown>[], options?: { noTable?: boolean }) => {
52	53	const log: unknown[] = [];
53		− const client: SqliteToolSchemaClient = {
	54	+ const client: SqliteDataMigrationClient = {
54	55	execute: (stmt) => {
55	56	log.push(stmt);
56	57	if (typeof stmt === "string" && stmt.includes("sqlite_master")) {
@@ -66,38 +67,74 @@ const makeFakeClient = (rows: Record<string, unknown>[], options?: { noTable?: b · 9 unmodified lines
66	67	};
67	68
68	69	describe("runSqliteOpenApiOutputSchemaMigration", () => {
69		− it("rewrites envelope rows in a transaction and reports the count", async () => {
70		− const payload = { type: "array", items: { type: "object" } };
71		− const { client, log } = makeFakeClient([
72		− { row_id: "a", output_schema: JSON.stringify(envelope(payload)) },
73		− { row_id: "b", output_schema: JSON.stringify(envelope({})) },
74		− { row_id: "c", output_schema: JSON.stringify({ $ref: "#/$defs/already_payload" }) },
75		− { row_id: "d", output_schema: "not json" },
76		− ]);
77		− const count = await runSqliteOpenApiOutputSchemaMigration(client);
78		− expect(count).toBe(2);
79		− expect(log).toContainEqual("BEGIN");
80		− expect(log).toContainEqual({
81		− sql: "UPDATE tool SET output_schema = ? WHERE row_id = ?",
82		− args: [JSON.stringify(payload), "a"],
83		− });
84		− expect(log).toContainEqual({
85		− sql: "UPDATE tool SET output_schema = ? WHERE row_id = ?",
86		− args: [null, "b"],
87		− });
88		− expect(log).toContainEqual("COMMIT");
89		− });
	70	+ it.effect("rewrites envelope rows in a transaction and reports the count", () =>
	71	+ Effect.gen(function* () {
	72	+ const payload = { type: "array", items: { type: "object" } };
	73	+ const { client, log } = makeFakeClient([
	74	+ { row_id: "a", output_schema: JSON.stringify(envelope(payload)) },
	75	+ { row_id: "b", output_schema: JSON.stringify(envelope({})) },
	76	+ { row_id: "c", output_schema: JSON.stringify({ $ref: "#/$defs/already_payload" }) },
	77	+ { row_id: "d", output_schema: "not json" },
	78	+ ]);
	79	+ const count = yield* runSqliteOpenApiOutputSchemaMigration(client);
	80	+ expect(count).toBe(2);
	81	+ expect(log).toContainEqual("BEGIN");
	82	+ expect(log).toContainEqual({
	83	+ sql: "UPDATE tool SET output_schema = ? WHERE row_id = ?",
	84	+ args: [JSON.stringify(payload), "a"],
	85	+ });
	86	+ expect(log).toContainEqual({
	87	+ sql: "UPDATE tool SET output_schema = ? WHERE row_id = ?",
	88	+ args: [null, "b"],
	89	+ });
	90	+ expect(log).toContainEqual("COMMIT");
	91	+ }),
	92	+ );
90	93
91		− it("no-ops when every row is already payload-shaped", async () => {
92		− const { client, log } = makeFakeClient([
93		− { row_id: "a", output_schema: JSON.stringify({ $ref: "#/$defs/already_payload" }) },
94		− ]);
95		− expect(await runSqliteOpenApiOutputSchemaMigration(client)).toBe(0);
96		− expect(log).not.toContainEqual("BEGIN");
97		− });
	94	+ it.effect("no-ops when every row is already payload-shaped", () =>
	95	+ Effect.gen(function* () {
	96	+ const { client, log } = makeFakeClient([
	97	+ { row_id: "a", output_schema: JSON.stringify({ $ref: "#/$defs/already_payload" }) },
	98	+ ]);
	99	+ const count = yield* runSqliteOpenApiOutputSchemaMigration(client);
	100	+ expect(count).toBe(0);
	101	+ expect(log).not.toContainEqual("BEGIN");
	102	+ }),
	103	+ );
98	104
99		− it("treats a missing tool table as nothing to migrate", async () => {
100		− const { client } = makeFakeClient([], { noTable: true });
101		− expect(await runSqliteOpenApiOutputSchemaMigration(client)).toBe(0);
102		− });
	105	+ it.effect("treats a missing tool table as nothing to migrate", () =>
	106	+ Effect.gen(function* () {
	107	+ const { client } = makeFakeClient([], { noTable: true });
	108	+ const count = yield* runSqliteOpenApiOutputSchemaMigration(client);
	109	+ expect(count).toBe(0);
	110	+ }),
	111	+ );
	112	+
	113	+ it.effect("rolls back and surfaces the failure when an update fails", () =>
	114	+ Effect.gen(function* () {
	115	+ const log: unknown[] = [];
	116	+ const client: SqliteDataMigrationClient = {
	117	+ execute: (stmt) => {
	118	+ log.push(stmt);
	119	+ if (typeof stmt === "string" && stmt.includes("sqlite_master")) {
	120	+ return Promise.resolve({ rows: [{ name: "tool" }] });
	121	+ }
	122	+ if (typeof stmt === "string" && stmt.startsWith("SELECT row_id")) {
	123	+ return Promise.resolve({
	124	+ rows: [{ row_id: "a", output_schema: JSON.stringify(envelope({})) }],
	125	+ });
	126	+ }
	127	+ if (typeof stmt === "object" && stmt.sql.startsWith("UPDATE tool")) {
	128	+ // oxlint-disable-next-line executor/no-promise-reject, executor/no-error-constructor -- simulates a raw driver failure at the adapter boundary under test
	129	+ return Promise.reject(new Error("disk full"));
	130	+ }
	131	+ return Promise.resolve({ rows: [] });
	132	+ },
	133	+ };
	134	+ const failure = yield* runSqliteOpenApiOutputSchemaMigration(client).pipe(Effect.flip);
	135	+ expect(Predicate.isTagged(failure, "DataMigrationError")).toBe(true);
	136	+ expect(log).toContainEqual("ROLLBACK");
	137	+ expect(log).not.toContainEqual("COMMIT");
	138	+ }),
	139	+ );
103	140	});

packages/plugins/openapi/src/sdk/output-schema-migration.ts+68−60

@@ -1,18 +1,22 @@
1		−/* oxlint-disable executor/no-try-catch-or-throw, executor/no-json-parse -- boundary: one-shot data migration drives a raw SQL client (JSON text columns, transaction + rollback) */
2	1	// ---------------------------------------------------------------------------
3		−// One-off boot migration: unwrap the retired {status, headers, data}
4		−// transport envelope from persisted OpenAPI tool output schemas. The runtime
5		−// returns the upstream payload as `data` (status/headers live in the
6		−// ToolResult `http` side channel), so persisted schemas must describe the
7		−// payload only — otherwise describe previews show an envelope invocations no
	2	+// Data migration: unwrap the retired {status, headers, data} transport
	3	+// envelope from persisted OpenAPI tool output schemas. The runtime returns
	4	+// the upstream payload as `data` (status/headers live in the ToolResult
	5	+// `http` side channel), so persisted schemas must describe the payload
	6	+// only — otherwise describe previews show an envelope invocations no
8	7	// longer return. Mirrors the cloud drizzle migration
9	8	// (apps/cloud/drizzle/0002_unwrap_openapi_output_envelope.sql) for the
10		−// libSQL-backed apps, which have no SQL migration chain at boot.
	9	+// libSQL-backed apps, where it runs once through the data-migration ledger.
11	10	//
12		−// Idempotent by construction: payload-shaped rows don't match the envelope
13		−// signature, so re-running plans zero updates.
	11	+// Idempotent: payload-shaped rows don't match the envelope signature, so
	12	+// re-running plans zero updates.
14	13	// ---------------------------------------------------------------------------
15	14
	15	+import { Effect, Option, Schema } from "effect";
	16	+import { DataMigrationError, type SqliteDataMigrationClient } from "@executor-js/sdk/core";
	17	+
	18	+const MIGRATION_NAME = "2026-06-11-openapi-output-envelope-unwrap";
	19	+
16	20	const isRecord = (value: unknown): value is Record<string, unknown> =>
17	21	typeof value === "object" && value !== null && !Array.isArray(value);
18	22
@@ -54,61 +58,65 @@ export const unwrapOpenApiTransportEnvelope = ( · 35 unmodified lines
54	58	return { outputSchema };
55	59	};
56	60
57		−// ---------------------------------------------------------------------------
58		−// SQLite runner — shared by the libSQL-backed apps (local boot, selfhost
59		−// boot). Structural client interface so the plugin doesn't take a driver
60		−// dependency; `@libsql/client` satisfies it.
61		−// ---------------------------------------------------------------------------
	61	+const decodeJsonOption = Schema.decodeUnknownOption(Schema.UnknownFromJsonString);
62	62
63		−export interface SqliteToolSchemaClient {
64		− execute(
65		− stmt: string \| { readonly sql: string; readonly args: readonly unknown[] },
66		− ): Promise<{ readonly rows: readonly Record<string, unknown>[] }>;
67		−}
	63	+const execute = (
	64	+ client: SqliteDataMigrationClient,
	65	+ stmt: string \| { readonly sql: string; readonly args: readonly unknown[] },
	66	+) =>
	67	+ Effect.tryPromise({
	68	+ try: () => client.execute(stmt),
	69	+ catch: (cause) => new DataMigrationError({ migration: MIGRATION_NAME, cause }),
	70	+ });
68	71
69	72	/** Unwrap envelope-shaped openapi tool output schemas in a SQLite database.
70		− * Idempotent; returns the number of rows rewritten. The `tool` table may
71		− * not exist yet on a fresh database — that counts as nothing to migrate. */
72		−export const runSqliteOpenApiOutputSchemaMigration = async (
73		− client: SqliteToolSchemaClient,
74		−): Promise<number> => {
75		− const exists = await client.execute(
76		− "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'tool'",
77		− );
78		− if (exists.rows.length === 0) return 0;
	73	+ * Returns the number of rows rewritten. The `tool` table may not exist yet
	74	+ * on a fresh database — that counts as nothing to migrate. */
	75	+export const runSqliteOpenApiOutputSchemaMigration = (
	76	+ client: SqliteDataMigrationClient,
	77	+): Effect.Effect<number, DataMigrationError> =>
	78	+ Effect.gen(function* () {
	79	+ const exists = yield* execute(
	80	+ client,
	81	+ "SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'tool'",
	82	+ );
	83	+ if (exists.rows.length === 0) return 0;
79	84
80		− const result = await client.execute(
81		− "SELECT row_id, output_schema FROM tool WHERE plugin_id = 'openapi' AND output_schema IS NOT NULL",
82		− );
83		− const updates: { readonly rowId: string; readonly outputSchema: unknown \| null }[] = [];
84		− for (const row of result.rows) {
85		− if (typeof row.row_id !== "string" \|\| typeof row.output_schema !== "string") continue;
86		− let schema: unknown;
87		− try {
88		− schema = JSON.parse(row.output_schema);
89		− } catch {
90		− continue;
	85	+ const result = yield* execute(
	86	+ client,
	87	+ "SELECT row_id, output_schema FROM tool WHERE plugin_id = 'openapi' AND output_schema IS NOT NULL",
	88	+ );
	89	+ const updates: { readonly rowId: string; readonly outputSchema: unknown \| null }[] = [];
	90	+ for (const row of result.rows) {
	91	+ if (typeof row.row_id !== "string" \|\| typeof row.output_schema !== "string") continue;
	92	+ const schema = decodeJsonOption(row.output_schema);
	93	+ if (Option.isNone(schema)) continue;
	94	+ const unwrapped = unwrapOpenApiTransportEnvelope(schema.value);
	95	+ if (unwrapped !== undefined) updates.push({ rowId: row.row_id, ...unwrapped });
91	96	}
92		− const unwrapped = unwrapOpenApiTransportEnvelope(schema);
93		− if (unwrapped !== undefined) updates.push({ rowId: row.row_id, ...unwrapped });
94		− }
95		− if (updates.length === 0) return 0;
	97	+ if (updates.length === 0) return 0;
96	98
97		− await client.execute("BEGIN");
98		− try {
99		− for (const update of updates) {
100		− await client.execute({
101		− sql: "UPDATE tool SET output_schema = ? WHERE row_id = ?",
102		− args: [
103		− update.outputSchema === null ? null : JSON.stringify(update.outputSchema),
104		− update.rowId,
105		− ],
106		− });
107		− }
108		− await client.execute("COMMIT");
109		− } catch (cause) {
110		− await client.execute("ROLLBACK");
111		− throw cause;
112		− }
113		− return updates.length;
	99	+ const applyAll = Effect.gen(function* () {
	100	+ for (const update of updates) {
	101	+ yield* execute(client, {
	102	+ sql: "UPDATE tool SET output_schema = ? WHERE row_id = ?",
	103	+ args: [
	104	+ update.outputSchema === null ? null : JSON.stringify(update.outputSchema),
	105	+ update.rowId,
	106	+ ],
	107	+ });
	108	+ }
	109	+ yield* execute(client, "COMMIT");
	110	+ });
	111	+
	112	+ yield* execute(client, "BEGIN");
	113	+ yield* applyAll.pipe(Effect.tapError(() => execute(client, "ROLLBACK").pipe(Effect.ignore)));
	114	+ return updates.length;
	115	+ });
	116	+
	117	+/** Registry entry for the boot-time data-migration ledger. */
	118	+export const openApiOutputSchemaDataMigration = {
	119	+ name: MIGRATION_NAME,
	120	+ run: (client: SqliteDataMigrationClient) =>
	121	+ runSqliteOpenApiOutputSchemaMigration(client).pipe(Effect.asVoid),
114	122	};