naytewilson · naytewilson · Aug 28, 2026
diff --git a/apps/server/src/persistence/CustomMigrationCompatibility.test.ts b/apps/server/src/persistence/CustomMigrationCompatibility.test.ts
@@ -0,0 +1,162 @@
+import { assert, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+
+import {
+  CUSTOM_WORK_LANES_MIGRATION_ID,
+  LEGACY_WORK_LANES_MIGRATION_NAME,
+  runCustomMigrationCompatibilityBridge,
+} from "./CustomMigrationCompatibility.ts";
+import { runMigrations } from "./Migrations.ts";
+import * as NodeSqliteClient from "./NodeSqliteClient.ts";
+
+const createLegacyPrivateTables = (sql: SqlClient.SqlClient) =>
+  Effect.gen(function* () {
+    yield* sql`CREATE TABLE projection_work_lanes (id TEXT PRIMARY KEY)`;
+    yield* sql`CREATE TABLE projection_source_truth_revisions (id TEXT PRIMARY KEY)`;
+    yield* sql`CREATE TABLE projection_lane_acceptance_criteria (id TEXT PRIMARY KEY)`;
+  });
+
+const seedLegacyMigration35 = (sql: SqlClient.SqlClient) =>
+  sql`
+    INSERT INTO effect_sql_migrations (migration_id, name)
+    VALUES (35, ${LEGACY_WORK_LANES_MIGRATION_NAME})
+  `;
+
+const assertIntegrity = Effect.fn("customMigrationCompatibility.assertIntegrity")(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  const rows = yield* sql<{ readonly integrity_check: string }>`PRAGMA integrity_check`;
+  assert.equal(rows[0]?.integrity_check, "ok");
+});
+
+const freshOfficialLayer = it.layer(NodeSqliteClient.layerMemory());
+freshOfficialLayer("custom migration compatibility: fresh official database", (it) => {
+  it.effect("preserves the official migration-35 path", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+
+      yield* runMigrations();
+
+      const migration35 = yield* sql<{ readonly name: string }>`
+        SELECT name FROM effect_sql_migrations WHERE migration_id = 35
+      `;
+      assert.equal(migration35[0]?.name, "ProjectionThreadTitleRegeneration");
+
+      const columns = yield* sql<{ readonly name: string }>`
+        PRAGMA table_info(projection_threads)
+      `;
+      const names = new Set(columns.map((column) => column.name));
+      assert.ok(names.has("title_regeneration_request_id"));
+      assert.ok(names.has("title_regeneration_started_at"));
+
+      const customRows = yield* sql<{ readonly count: number }>`
+        SELECT COUNT(*) AS count FROM t3_custom_schema_migrations
+      `;
+      assert.equal(customRows[0]?.count, 0);
+      yield* assertIntegrity();
+    }),
+  );
+});
+
+const legacyCustomLayer = it.layer(NodeSqliteClient.layerMemory());
+legacyCustomLayer("custom migration compatibility: legacy custom database", (it) => {
+  it.effect("keeps historical row 35, repairs upstream 35, and continues through 43", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+
+      yield* runMigrations({ toMigrationInclusive: 34 });
+      yield* createLegacyPrivateTables(sql);
+      yield* seedLegacyMigration35(sql);
+
+      yield* runMigrations();
+
+      const migration35 = yield* sql<{ readonly name: string }>`
+        SELECT name FROM effect_sql_migrations WHERE migration_id = 35
+      `;
+      assert.equal(migration35[0]?.name, LEGACY_WORK_LANES_MIGRATION_NAME);
+
+      const latestMigration = yield* sql<{ readonly migration_id: number; readonly name: string }>`
+        SELECT migration_id, name
+        FROM effect_sql_migrations
+        ORDER BY migration_id DESC
+        LIMIT 1
+      `;
+      assert.equal(latestMigration[0]?.migration_id, 43);
+      assert.equal(latestMigration[0]?.name, "ProjectionThreadsUnsettledAt");
+
+      const columns = yield* sql<{ readonly name: string }>`
+        PRAGMA table_info(projection_threads)
+      `;
+      const names = new Set(columns.map((column) => column.name));
+      assert.ok(names.has("title_regeneration_request_id"));
+      assert.ok(names.has("title_regeneration_started_at"));
+
+      const adopted = yield* sql<{ readonly name: string }>`
+        SELECT name
+        FROM t3_custom_schema_migrations
+        WHERE migration_id = ${CUSTOM_WORK_LANES_MIGRATION_ID}
+      `;
+      assert.equal(adopted[0]?.name, LEGACY_WORK_LANES_MIGRATION_NAME);
+      yield* assertIntegrity();
+    }),
+  );
+});
+
+const alreadyBridgedLayer = it.layer(NodeSqliteClient.layerMemory());
+alreadyBridgedLayer("custom migration compatibility: already bridged database", (it) => {
+  it.effect("is idempotent", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+
+      yield* runMigrations({ toMigrationInclusive: 34 });
+      yield* createLegacyPrivateTables(sql);
+      yield* seedLegacyMigration35(sql);
+
+      const first = yield* runCustomMigrationCompatibilityBridge();
+      const second = yield* runCustomMigrationCompatibilityBridge();
+
+      assert.deepEqual(first.addedTitleRegenerationColumns, [
+        "title_regeneration_request_id",
+        "title_regeneration_started_at",
+      ]);
+      assert.deepEqual(second.addedTitleRegenerationColumns, []);
+      assert.equal(second.legacyWorkLanesMigrationDetected, true);
+      assert.equal(second.adoptedLegacyWorkLanesMigration, true);
+
+      const columns = yield* sql<{ readonly name: string }>`
+        PRAGMA table_info(projection_threads)
+      `;
+      assert.equal(
+        columns.filter((column) => column.name === "title_regeneration_request_id").length,
+        1,
+      );
+      assert.equal(
+        columns.filter((column) => column.name === "title_regeneration_started_at").length,
+        1,
+      );
+
+      const customRows = yield* sql<{ readonly count: number }>`
+        SELECT COUNT(*) AS count
+        FROM t3_custom_schema_migrations
+        WHERE migration_id = ${CUSTOM_WORK_LANES_MIGRATION_ID}
+      `;
+      assert.equal(customRows[0]?.count, 1);
+      yield* assertIntegrity();
+    }),
+  );
+});
+
+const malformedLegacyLayer = it.layer(NodeSqliteClient.layerMemory());
+malformedLegacyLayer("custom migration compatibility: malformed legacy database", (it) => {
+  it.effect("fails closed when the historical ledger row exists without its private tables", () =>
+    Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+
+      yield* runMigrations({ toMigrationInclusive: 34 });
+      yield* seedLegacyMigration35(sql);
+
+      const error = yield* runCustomMigrationCompatibilityBridge().pipe(Effect.flip);
+      assert.match(String(error), /required private tables are missing/i);
+    }),
+  );
+});
diff --git a/apps/server/src/persistence/CustomMigrationCompatibility.ts b/apps/server/src/persistence/CustomMigrationCompatibility.ts
@@ -0,0 +1,159 @@
+import * as Effect from "effect/Effect";
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+
+export const CUSTOM_MIGRATIONS_TABLE = "t3_custom_schema_migrations";
+export const LEGACY_WORK_LANES_MIGRATION_ID = 35;
+export const LEGACY_WORK_LANES_MIGRATION_NAME = "WorkLanesAndSourceTruth";
+export const CUSTOM_WORK_LANES_MIGRATION_ID = 1;
+
+const REQUIRED_LEGACY_WORK_LANE_TABLES = [
+  "projection_work_lanes",
+  "projection_source_truth_revisions",
+  "projection_lane_acceptance_criteria",
+] as const;
+
+export interface CustomMigrationCompatibilityResult {
+  readonly legacyWorkLanesMigrationDetected: boolean;
+  readonly addedTitleRegenerationColumns: ReadonlyArray<string>;
+  readonly adoptedLegacyWorkLanesMigration: boolean;
+}
+
+const tableExists = Effect.fn("customMigrationCompatibility.tableExists")(function* (name: string) {
+  const sql = yield* SqlClient.SqlClient;
+  const rows = yield* sql<{ readonly name: string }>`
+    SELECT name
+    FROM sqlite_master
+    WHERE type = 'table' AND name = ${name}
+    LIMIT 1
+  `;
+  return rows.length > 0;
+});
+
+const ensureCustomMigrationTable = Effect.fn(
+  "customMigrationCompatibility.ensureCustomMigrationTable",
+)(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  yield* sql`
+    CREATE TABLE IF NOT EXISTS t3_custom_schema_migrations (
+      migration_id INTEGER PRIMARY KEY NOT NULL,
+      name VARCHAR(255) NOT NULL,
+      created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+    )
+  `;
+});
+
+/**
+ * Bridge the one known collision between the historical custom fork and upstream.
+ *
+ * The old custom fork recorded migration 35 as WorkLanesAndSourceTruth in
+ * effect_sql_migrations. Upstream v0.0.35 also uses id 35, for the title-regeneration
+ * columns. Effect's migrator skips ids <= the latest recorded id, so an old custom
+ * database would otherwise skip upstream 35 and later fail when code reads the
+ * missing columns.
+ *
+ * This bridge deliberately leaves effect_sql_migrations row 35 untouched. When that row
+ * is the legacy custom migration, it installs only the idempotent upstream-35 schema
+ * additions, verifies the private tables implied by the historical row, and adopts the
+ * private migration into a separate custom namespace for future fork-only migrations.
+ */
+export const runCustomMigrationCompatibilityBridge = Effect.fn(
+  "runCustomMigrationCompatibilityBridge",
+)(function* () {
+  const sql = yield* SqlClient.SqlClient;
+
+  yield* ensureCustomMigrationTable();
+
+  if (!(yield* tableExists("effect_sql_migrations"))) {
+    return {
+      legacyWorkLanesMigrationDetected: false,
+      addedTitleRegenerationColumns: [],
+      adoptedLegacyWorkLanesMigration: false,
+    } satisfies CustomMigrationCompatibilityResult;
+  }
+
+  const legacyRows = yield* sql<{ readonly name: string }>`
+    SELECT name
+    FROM effect_sql_migrations
+    WHERE migration_id = ${LEGACY_WORK_LANES_MIGRATION_ID}
+    LIMIT 1
+  `;
+  const legacyName = legacyRows[0]?.name;
+  if (legacyName !== LEGACY_WORK_LANES_MIGRATION_NAME) {
+    return {
+      legacyWorkLanesMigrationDetected: false,
+      addedTitleRegenerationColumns: [],
+      adoptedLegacyWorkLanesMigration: false,
+    } satisfies CustomMigrationCompatibilityResult;
+  }
+
+  if (!(yield* tableExists("projection_threads"))) {
+    return yield* Effect.fail(
+      new Error(
+        "Legacy migration 35_WorkLanesAndSourceTruth is recorded, but projection_threads is missing",
+      ),
+    );
+  }
+
+  const missingPrivateTables: Array<string> = [];
+  for (const table of REQUIRED_LEGACY_WORK_LANE_TABLES) {
+    if (!(yield* tableExists(table))) {
+      missingPrivateTables.push(table);
+    }
+  }
+  if (missingPrivateTables.length > 0) {
+    return yield* Effect.fail(
+      new Error(
+        `Legacy migration 35_WorkLanesAndSourceTruth is recorded, but required private tables are missing: ${missingPrivateTables.join(", ")}`,
+      ),
+    );
+  }
+
+  const columns = yield* sql<{ readonly name: string }>`
+    PRAGMA table_info(projection_threads)
+  `;
+  const columnNames = new Set(columns.map((column) => column.name));
+  const addedTitleRegenerationColumns: Array<string> = [];
+
+  if (!columnNames.has("title_regeneration_request_id")) {
+    yield* sql`
+      ALTER TABLE projection_threads
+      ADD COLUMN title_regeneration_request_id TEXT
+    `;
+    addedTitleRegenerationColumns.push("title_regeneration_request_id");
+  }
+
+  if (!columnNames.has("title_regeneration_started_at")) {
+    yield* sql`
+      ALTER TABLE projection_threads
+      ADD COLUMN title_regeneration_started_at TEXT
+    `;
+    addedTitleRegenerationColumns.push("title_regeneration_started_at");
+  }
+
+  yield* sql`
+    INSERT INTO t3_custom_schema_migrations (migration_id, name)
+    VALUES (${CUSTOM_WORK_LANES_MIGRATION_ID}, ${LEGACY_WORK_LANES_MIGRATION_NAME})
+    ON CONFLICT (migration_id) DO NOTHING
+  `;
+
+  const adoptedRows = yield* sql<{ readonly name: string }>`
+    SELECT name
+    FROM t3_custom_schema_migrations
+    WHERE migration_id = ${CUSTOM_WORK_LANES_MIGRATION_ID}
+    LIMIT 1
+  `;
+  const adoptedName = adoptedRows[0]?.name;
+  if (adoptedName !== LEGACY_WORK_LANES_MIGRATION_NAME) {
+    return yield* Effect.fail(
+      new Error(
+        `Custom migration namespace collision at id ${CUSTOM_WORK_LANES_MIGRATION_ID}: expected ${LEGACY_WORK_LANES_MIGRATION_NAME}, found ${adoptedName ?? "missing"}`,
+      ),
+    );
+  }
+
+  return {
+    legacyWorkLanesMigrationDetected: true,
+    addedTitleRegenerationColumns,
+    adoptedLegacyWorkLanesMigration: true,
+  } satisfies CustomMigrationCompatibilityResult;
+});
diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts
@@ -12,6 +12,8 @@ import * as Migrator from "effect/unstable/sql/Migrator";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 
+import { runCustomMigrationCompatibilityBridge } from "./CustomMigrationCompatibility.ts";
+
 // Import all migrations statically
 import Migration0001 from "./Migrations/001_OrchestrationEvents.ts";
 import Migration0002 from "./Migrations/002_OrchestrationCommandReceipts.ts";
@@ -147,6 +149,7 @@ export interface RunMigrationsOptions {
 export const runMigrations = Effect.fn("runMigrations")(function* ({
   toMigrationInclusive,
 }: RunMigrationsOptions = {}) {
+  yield* runCustomMigrationCompatibilityBridge();
   const executedMigrations = yield* run({ loader: makeMigrationLoader(toMigrationInclusive) });
   const migrations = executedMigrations.map(([id, name]) => `${id}_${name}`);
   yield* migrations.length === 0