diff --git a/AGENTS.md b/AGENTS.md index b560b7aa2a90..66991a904546 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -195,6 +195,12 @@ section by keeping the fork's version. does not extend to browsers or computer use. For changes that aren't visible in the web app, state concretely how to observe the result instead (command to run, log to tail, or the mobile test flow). +- Fork-only database migrations live in + `apps/server/src/persistence/ForkMigrations/` with their own IDs (001, 002, + ...) and their own ledger table, run by `runAllMigrations` after upstream's. + Never add a fork migration to `Migrations.ts`: the Effect Migrator is a + high-water mark, so a fork ID in upstream's sequence either skips upstream's + next migration or crashes on it at the following sync. - Upstream syncs are explicit and discretionary, done as a `sync-upstream-` branch PR'd into fork `main` — never a bare pull of upstream into `main`. diff --git a/apps/server/scripts/migrate-dev-db.test.ts b/apps/server/scripts/migrate-dev-db.test.ts index ddc5b7d57f86..367947ab00a1 100644 --- a/apps/server/scripts/migrate-dev-db.test.ts +++ b/apps/server/scripts/migrate-dev-db.test.ts @@ -5,7 +5,7 @@ import * as FileSystem from "effect/FileSystem"; import * as Path from "effect/Path"; import * as SqlClient from "effect/unstable/sql/SqlClient"; -import { runMigrations } from "../src/persistence/Migrations.ts"; +import { runAllMigrations } from "../src/persistence/ForkMigrations.ts"; import * as NodeSqliteClient from "../src/persistence/NodeSqliteClient.ts"; import { runMigrateDevDb } from "./migrate-dev-db.ts"; @@ -28,7 +28,7 @@ const createFixtureSource = Effect.fn("createMigrateDevDbFixtureSource")(functio databasePath, Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* runMigrations(); + yield* runAllMigrations(); // The real shared db carries this column from a branch build without a // matching migration; reproduce that drift so the filter is exercised. yield* sql`ALTER TABLE projection_threads ADD COLUMN monitor_json TEXT`; @@ -132,6 +132,36 @@ it.layer(NodeServices.layer)("migrate-dev-db", (it) => { }), ); + it.effect("fails loudly on a fork migration slot collision", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const sourceDir = yield* fs.makeTempDirectoryScoped({ prefix: "migrate-dev-db-fork-slot-" }); + const destDir = yield* fs.makeTempDirectoryScoped({ + prefix: "migrate-dev-db-fork-slot-dest-", + }); + const source = yield* createFixtureSource(sourceDir); + yield* withDatabase( + source, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`UPDATE effect_sql_migrations_fork + SET name = 'SomebodyElsesForkMigration' WHERE migration_id = 1`; + }), + ); + + const error = yield* runMigrateDevDb( + { baseDir: destDir, source, projects: 5, threadsPerProject: 10 }, + { sharedHome: sourceDir }, + ).pipe(Effect.flip); + assert.equal(error._tag, "MigrateDevDbSlotCollisionError"); + if (error._tag === "MigrateDevDbSlotCollisionError") { + assert.equal(error.ledger, "effect_sql_migrations_fork"); + assert.equal(error.slot, 1); + assert.equal(error.appliedName, "SomebodyElsesForkMigration"); + } + }), + ); + it.effect("refuses while a dev server holds the destination", () => Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; diff --git a/apps/server/scripts/migrate-dev-db.ts b/apps/server/scripts/migrate-dev-db.ts index 0958f2149f45..da0c641f65f9 100644 --- a/apps/server/scripts/migrate-dev-db.ts +++ b/apps/server/scripts/migrate-dev-db.ts @@ -38,7 +38,12 @@ import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { Command, Flag } from "effect/unstable/cli"; -import { migrationManifest, runMigrations } from "../src/persistence/Migrations.ts"; +import { + forkMigrationManifest, + forkMigrationsTable, + runAllMigrations, +} from "../src/persistence/ForkMigrations.ts"; +import { migrationManifest } from "../src/persistence/Migrations.ts"; import * as NodeSqliteClient from "../src/persistence/NodeSqliteClient.ts"; export class MigrateDevDbNotInWorktreeError extends Schema.TaggedErrorClass()( @@ -118,13 +123,14 @@ export class MigrateDevDbDestinationBusyError extends Schema.TaggedErrorClass()( "MigrateDevDbSlotCollisionError", { + ledger: Schema.String, slot: Schema.Number, codeName: Schema.String, appliedName: Schema.String, }, ) { override get message(): string { - return `Migration slot collision at ${this.slot}: this checkout registers '${this.codeName}' but the database already applied '${this.appliedName}' in that slot. Renumber the new migration to a free slot.`; + return `Migration slot collision at ${this.ledger} ${this.slot}: this checkout registers '${this.codeName}' but the database already applied '${this.appliedName}' in that slot. Renumber the new migration to a free slot.`; } } @@ -336,15 +342,22 @@ const pruneSnapshot = Effect.fn("pruneDevDbSnapshot")(function* (input: RunMigra /** Compare this checkout's migration registry against what the cloned * database recorded: same slot under a different name means the migration * was skipped, not applied. */ +const migrationLedgers = [ + ["effect_sql_migrations", migrationManifest], + [forkMigrationsTable, forkMigrationManifest], +] as const; + const verifyMigrationSlots = Effect.fn("verifyMigrationSlots")(function* () { const sql = yield* SqlClient.SqlClient; - const applied = yield* sql<{ migration_id: number; name: string }>` - SELECT migration_id, name FROM effect_sql_migrations`; - const appliedById = new Map(applied.map((row) => [Number(row.migration_id), row.name])); - for (const [slot, codeName] of migrationManifest) { - const appliedName = appliedById.get(slot); - if (appliedName !== undefined && appliedName !== codeName) { - return yield* new MigrateDevDbSlotCollisionError({ slot, codeName, appliedName }); + for (const [ledger, manifest] of migrationLedgers) { + const applied = yield* sql<{ migration_id: number; name: string }>` + SELECT migration_id, name FROM ${sql(ledger)}`; + const appliedById = new Map(applied.map((row) => [Number(row.migration_id), row.name])); + for (const [slot, codeName] of manifest) { + const appliedName = appliedById.get(slot); + if (appliedName !== undefined && appliedName !== codeName) { + return yield* new MigrateDevDbSlotCollisionError({ ledger, slot, codeName, appliedName }); + } } } }); @@ -436,7 +449,11 @@ export const runMigrateDevDb = Effect.fn("runMigrateDevDb")(function* ( const sql = yield* SqlClient.SqlClient; // Mirror server boot (persistence/Layers/Sqlite.ts). yield* sql.unsafe("PRAGMA foreign_keys = ON").unprepared; - return yield* runMigrations(); + const { upstream, fork } = yield* runAllMigrations(); + return [ + ...upstream.map(([id, name]) => `${id}_${name}`), + ...fork.map(([id, name]) => `fork/${id}_${name}`), + ]; }).pipe( Effect.provide(NodeSqliteClient.layer({ filename: snapshotPath })), wrapPhase("migrate", snapshotPath), @@ -495,7 +512,7 @@ export const runMigrateDevDb = Effect.fn("runMigrateDevDb")(function* ( sizeBytes: Number(size), projects: pruned.projects, eventCount: pruned.eventCount, - executedMigrations: executedMigrations.map(([id, name]) => `${id}_${name}`), + executedMigrations, }; }); diff --git a/apps/server/src/persistence/ForkMigrations.test.ts b/apps/server/src/persistence/ForkMigrations.test.ts new file mode 100644 index 000000000000..a688758e2e05 --- /dev/null +++ b/apps/server/src/persistence/ForkMigrations.test.ts @@ -0,0 +1,177 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Migrator from "effect/unstable/sql/Migrator"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { migrationManifest, runMigrations } from "./Migrations.ts"; +import { + forkMigrationEntries, + forkMigrationManifest, + forkMigrationsTable, + runAllMigrations, +} from "./ForkMigrations.ts"; +import * as NodeSqliteClient from "./NodeSqliteClient.ts"; + +const freshLayer = () => it.layer(Layer.fresh(Layer.mergeAll(NodeSqliteClient.layerMemory()))); + +const upstreamLatestId = migrationManifest.at(-1)![0]; +const legacyForkRow = (id: number) => 40 + id; + +const readLedger = (table: string) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + return yield* sql<{ readonly id: number; readonly name: string; readonly createdAt: string }>` + SELECT migration_id AS id, name, created_at AS "createdAt" + FROM ${sql(table)} + ORDER BY migration_id ASC + `; + }); + +const readIds = (table: string) => + Effect.map(readLedger(table), (rows) => rows.map((row) => row.id)); + +// Reproduces a database from before the split: fork migrations applied +// through upstream's ledger as 041..046. +const applyLegacyForkMigrations = Effect.gen(function* () { + yield* runMigrations(); + const legacyLoader = Migrator.fromRecord( + Object.fromEntries( + forkMigrationEntries.map(([id, name, migration]) => [ + `${legacyForkRow(id)}_${name}`, + migration, + ]), + ), + ); + yield* Migrator.make({})({ loader: legacyLoader }); +}); + +freshLayer()("ForkMigrations on a fresh database", (it) => { + it.effect("runs upstream migrations in their ledger and fork migrations in the fork ledger", () => + Effect.gen(function* () { + const result = yield* runAllMigrations(); + + assert.deepEqual( + result.upstream.map(([id]) => id), + migrationManifest.map(([id]) => id), + ); + assert.deepEqual(result.fork, forkMigrationManifest); + assert.deepEqual( + yield* readIds("effect_sql_migrations"), + migrationManifest.map(([id]) => id), + ); + assert.deepEqual( + yield* readIds(forkMigrationsTable), + forkMigrationManifest.map(([id]) => id), + ); + + const again = yield* runAllMigrations(); + assert.deepEqual(again, { upstream: [], fork: [] }); + }), + ); +}); + +freshLayer()("ForkMigrations on a database migrated before the split", (it) => { + it.effect("moves fork rows out of the upstream ledger without re-running them", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* applyLegacyForkMigrations; + yield* sql` + UPDATE effect_sql_migrations + SET created_at = '2026-08-12 03:04:05' + WHERE migration_id > ${upstreamLatestId} + `; + // A projection row that a re-run of the rebuild migrations would wipe. + yield* sql` + INSERT INTO projection_state (projector, last_applied_sequence, updated_at) + VALUES ('projection.threads', 27427, '2026-08-12T03:04:05.000Z') + `; + + const result = yield* runAllMigrations(); + + assert.deepEqual(result, { upstream: [], fork: [] }); + assert.deepEqual( + yield* readIds("effect_sql_migrations"), + migrationManifest.map(([id]) => id), + ); + assert.deepEqual( + yield* readLedger(forkMigrationsTable), + forkMigrationManifest.map(([id, name]) => ({ + id, + name, + createdAt: "2026-08-12 03:04:05", + })), + ); + const stateRows = yield* sql<{ readonly lastAppliedSequence: number }>` + SELECT last_applied_sequence AS "lastAppliedSequence" FROM projection_state + `; + assert.deepEqual(stateRows, [{ lastAppliedSequence: 27427 }]); + }), + ); +}); + +freshLayer()( + "ForkMigrations next to an upstream migration that reuses a legacy fork number", + (it) => { + it.effect("adopts only rows whose name matches a fork migration", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* applyLegacyForkMigrations; + const takenId = legacyForkRow(forkMigrationEntries.length); + yield* sql` + UPDATE effect_sql_migrations SET name = 'ApplicationEventSource' WHERE migration_id = ${takenId} + `; + + const result = yield* runAllMigrations(); + + // The renamed row stays put and, since it is above the fork's legacy + // range, the fork migration it displaced re-runs under its own ledger. + assert.deepEqual( + result.fork.map(([id]) => id), + [forkMigrationEntries.length], + ); + assert.deepEqual(yield* readIds("effect_sql_migrations"), [ + ...migrationManifest.map(([id]) => id), + takenId, + ]); + assert.deepEqual( + yield* readIds(forkMigrationsTable), + forkMigrationManifest.map(([id]) => id), + ); + }), + ); + }, +); + +freshLayer()("ForkMigrations when the fork ledger already holds a different migration", (it) => { + it.effect("refuses to adopt rather than dropping the legacy row", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* applyLegacyForkMigrations; + yield* sql` + CREATE TABLE ${sql(forkMigrationsTable)} ( + migration_id integer PRIMARY KEY NOT NULL, + created_at datetime NOT NULL DEFAULT current_timestamp, + name VARCHAR(255) NOT NULL + ) + `; + yield* sql` + INSERT INTO ${sql(forkMigrationsTable)} (migration_id, name) + VALUES (${forkMigrationEntries.length}, 'DifferentMigration') + `; + + const error = yield* runAllMigrations().pipe(Effect.flip); + + assert.equal(error._tag, "MigrationError"); + if (error._tag === "MigrationError") { + assert.equal(error.kind, "BadState"); + } + // The transaction rolled back: every legacy row is still in place. + assert.deepEqual(yield* readIds("effect_sql_migrations"), [ + ...migrationManifest.map(([id]) => id), + ...forkMigrationManifest.map(([id]) => legacyForkRow(id)), + ]); + assert.deepEqual(yield* readIds(forkMigrationsTable), [forkMigrationEntries.length]); + }), + ); +}); diff --git a/apps/server/src/persistence/ForkMigrations.ts b/apps/server/src/persistence/ForkMigrations.ts new file mode 100644 index 000000000000..fcde9af77d2d --- /dev/null +++ b/apps/server/src/persistence/ForkMigrations.ts @@ -0,0 +1,152 @@ +/** + * Fork-only migrations, kept out of upstream's ledger. + * + * The Effect Migrator is a high-water mark: it runs every loaded migration + * whose ID is above the largest ID already recorded in its table. Numbering + * fork migrations after upstream's (041, 042, ...) therefore breaks the next + * upstream sync in one of two ways: upstream's own 041 is skipped as already + * applied, or a renumbered fork migration crashes on tables that do not exist. + * Parking them in a high range is worse still, since the mark then sits above + * every future upstream migration. + * + * So fork migrations get their own ID space (1, 2, ...) and their own ledger + * table, run through a second Migrator after upstream's. `Migrations.ts` stays + * byte-identical to upstream and upstream can keep numbering from 041. + */ + +import * as Migrator from "effect/unstable/sql/Migrator"; +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "./Migrations.ts"; +import ForkMigration001 from "./ForkMigrations/001_ProjectionTurnRetractions.ts"; +import ForkMigration002 from "./ForkMigrations/002_ProjectionTurnDispatchOwnership.ts"; +import ForkMigration003 from "./ForkMigrations/003_ProjectionManagedWorktrees.ts"; +import ForkMigration004 from "./ForkMigrations/004_CleanupCompletedRetractionMessages.ts"; +import ForkMigration005 from "./ForkMigrations/005_RebuildProjectionsFromEvents.ts"; +import ForkMigration006 from "./ForkMigrations/006_RebuildProjectionsWithRetainedTurns.ts"; + +export const forkMigrationsTable = "effect_sql_migrations_fork"; +const upstreamMigrationsTable = "effect_sql_migrations"; + +export const forkMigrationEntries = [ + [1, "ProjectionTurnRetractions", ForkMigration001], + [2, "ProjectionTurnDispatchOwnership", ForkMigration002], + [3, "ProjectionManagedWorktrees", ForkMigration003], + [4, "CleanupCompletedRetractionMessages", ForkMigration004], + [5, "RebuildProjectionsFromEvents", ForkMigration005], + [6, "RebuildProjectionsWithRetainedTurns", ForkMigration006], +] as const; + +export const forkMigrationManifest = forkMigrationEntries.map(([id, name]) => [id, name] as const); + +/** + * Before the split, fork migrations 1..6 shipped as 041..046 in the upstream + * ledger. Databases that applied them there carry those rows, so the first + * boot on this code moves them into the fork ledger under their fork IDs. + * Rows are matched by ID and name, so an upstream migration that later takes + * one of those numbers is left alone. + */ +const legacyUpstreamIdOffset = 40; + +const makeForkMigrationLoader = (throughId?: number) => + Migrator.fromRecord( + Object.fromEntries( + forkMigrationEntries + .filter(([id]) => throughId === undefined || id <= throughId) + .map(([id, name, migration]) => [`${id}_${name}`, migration]), + ), + ); + +const run = Migrator.make({}); + +const adoptLegacyForkLedgerRows = Effect.fn("adoptLegacyForkLedgerRows")(function* () { + const sql = yield* SqlClient.SqlClient; + const upstreamLedger = yield* sql<{ readonly name: string }>` + SELECT name FROM sqlite_master WHERE type = 'table' AND name = ${upstreamMigrationsTable} + `; + if (upstreamLedger.length === 0) { + return; + } + // Same shape the Migrator creates for sqlite, so it adopts the table as-is. + yield* sql` + CREATE TABLE IF NOT EXISTS ${sql(forkMigrationsTable)} ( + migration_id integer PRIMARY KEY NOT NULL, + created_at datetime NOT NULL DEFAULT current_timestamp, + name VARCHAR(255) NOT NULL + ) + `; + yield* sql.withTransaction( + Effect.forEach( + forkMigrationEntries, + ([id, name]) => { + const legacyId = legacyUpstreamIdOffset + id; + return Effect.gen(function* () { + const legacy = yield* sql<{ readonly createdAt: string }>` + SELECT created_at AS "createdAt" + FROM ${sql(upstreamMigrationsTable)} + WHERE migration_id = ${legacyId} AND name = ${name} + `; + if (legacy.length === 0) { + return; + } + const existing = yield* sql<{ readonly name: string }>` + SELECT name FROM ${sql(forkMigrationsTable)} WHERE migration_id = ${id} + `; + if (existing.length === 0) { + yield* sql` + INSERT INTO ${sql(forkMigrationsTable)} (migration_id, created_at, name) + VALUES (${id}, ${legacy[0]!.createdAt}, ${name}) + `; + } else if (existing[0]!.name !== name) { + // Deleting the legacy row here would lose the only record that the + // fork migration ran; refuse instead of silently skipping it. + return yield* new Migrator.MigrationError({ + kind: "BadState", + message: `Fork migration ledger slot ${id} holds "${existing[0]!.name}" but upstream ledger row ${legacyId} records "${name}"`, + }); + } + yield* sql` + DELETE FROM ${sql(upstreamMigrationsTable)} + WHERE migration_id = ${legacyId} AND name = ${name} + `; + }); + }, + { discard: true }, + ), + ); +}); + +export interface RunForkMigrationsOptions { + readonly toMigrationInclusive?: number | undefined; +} + +/** + * Run pending fork migrations against the fork ledger. Assumes upstream + * migrations have already run; fork migrations build on upstream's tables. + */ +export const runForkMigrations = Effect.fn("runForkMigrations")(function* ({ + toMigrationInclusive, +}: RunForkMigrationsOptions = {}) { + const executedMigrations = yield* run({ + loader: makeForkMigrationLoader(toMigrationInclusive), + table: forkMigrationsTable, + }); + const migrations = executedMigrations.map(([id, name]) => `fork/${id}_${name}`); + yield* migrations.length === 0 + ? Effect.logDebug("Fork database schema is current") + : Effect.log("Fork migrations ran successfully").pipe(Effect.annotateLogs({ migrations })); + return executedMigrations; +}); + +/** + * Boot entry point: adopt legacy fork ledger rows, run upstream migrations, + * then run fork migrations. Adoption must come first so upstream's high-water + * mark drops back to upstream's own latest migration before it is consulted. + */ +export const runAllMigrations = Effect.fn("runAllMigrations")(function* () { + yield* adoptLegacyForkLedgerRows(); + const upstream = yield* runMigrations(); + const fork = yield* runForkMigrations(); + return { upstream, fork }; +}); diff --git a/apps/server/src/persistence/Migrations/041_ProjectionTurnRetractions.test.ts b/apps/server/src/persistence/ForkMigrations/001_ProjectionTurnRetractions.test.ts similarity index 89% rename from apps/server/src/persistence/Migrations/041_ProjectionTurnRetractions.test.ts rename to apps/server/src/persistence/ForkMigrations/001_ProjectionTurnRetractions.test.ts index fe6da66ec8c0..ebf9d394e220 100644 --- a/apps/server/src/persistence/Migrations/041_ProjectionTurnRetractions.test.ts +++ b/apps/server/src/persistence/ForkMigrations/001_ProjectionTurnRetractions.test.ts @@ -4,22 +4,23 @@ import * as Layer from "effect/Layer"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { runMigrations } from "../Migrations.ts"; +import { runForkMigrations } from "../ForkMigrations.ts"; import * as NodeSqliteClient from "../NodeSqliteClient.ts"; const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); -layer("041_ProjectionTurnRetractions", (it) => { +layer("fork/001_ProjectionTurnRetractions", (it) => { it.effect("upgrades an existing schema with durable, startup-indexed retraction rows", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* runMigrations({ toMigrationInclusive: 40 }); + yield* runMigrations(); const existingTables = yield* sql<{ readonly name: string }>` SELECT name FROM sqlite_master WHERE type = 'table' AND name = 'projection_threads' `; assert.equal(existingTables.length, 1); - yield* runMigrations({ toMigrationInclusive: 41 }); + yield* runForkMigrations({ toMigrationInclusive: 1 }); const columns = yield* sql<{ readonly name: string }>` PRAGMA table_info(projection_turn_retractions) diff --git a/apps/server/src/persistence/Migrations/041_ProjectionTurnRetractions.ts b/apps/server/src/persistence/ForkMigrations/001_ProjectionTurnRetractions.ts similarity index 100% rename from apps/server/src/persistence/Migrations/041_ProjectionTurnRetractions.ts rename to apps/server/src/persistence/ForkMigrations/001_ProjectionTurnRetractions.ts diff --git a/apps/server/src/persistence/Migrations/042_ProjectionTurnDispatchOwnership.test.ts b/apps/server/src/persistence/ForkMigrations/002_ProjectionTurnDispatchOwnership.test.ts similarity index 86% rename from apps/server/src/persistence/Migrations/042_ProjectionTurnDispatchOwnership.test.ts rename to apps/server/src/persistence/ForkMigrations/002_ProjectionTurnDispatchOwnership.test.ts index 261ade38b311..b26cedbf0633 100644 --- a/apps/server/src/persistence/Migrations/042_ProjectionTurnDispatchOwnership.test.ts +++ b/apps/server/src/persistence/ForkMigrations/002_ProjectionTurnDispatchOwnership.test.ts @@ -4,16 +4,18 @@ import * as Layer from "effect/Layer"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { runMigrations } from "../Migrations.ts"; +import { runForkMigrations } from "../ForkMigrations.ts"; import * as NodeSqliteClient from "../NodeSqliteClient.ts"; const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); -layer("042_ProjectionTurnDispatchOwnership", (it) => { +layer("fork/002_ProjectionTurnDispatchOwnership", (it) => { it.effect("adds durable provider-send ownership and claim storage", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* runMigrations({ toMigrationInclusive: 41 }); + yield* runMigrations(); + yield* runForkMigrations({ toMigrationInclusive: 1 }); yield* sql` INSERT INTO projection_turn_retractions ( request_id, thread_id, message_id, baseline_turn_count, @@ -26,7 +28,7 @@ layer("042_ProjectionTurnDispatchOwnership", (it) => { ) `; - yield* runMigrations({ toMigrationInclusive: 42 }); + yield* runForkMigrations({ toMigrationInclusive: 2 }); const columns = yield* sql<{ readonly name: string }>` PRAGMA table_info(projection_turn_retractions) diff --git a/apps/server/src/persistence/Migrations/042_ProjectionTurnDispatchOwnership.ts b/apps/server/src/persistence/ForkMigrations/002_ProjectionTurnDispatchOwnership.ts similarity index 100% rename from apps/server/src/persistence/Migrations/042_ProjectionTurnDispatchOwnership.ts rename to apps/server/src/persistence/ForkMigrations/002_ProjectionTurnDispatchOwnership.ts diff --git a/apps/server/src/persistence/Migrations/043_ProjectionManagedWorktrees.test.ts b/apps/server/src/persistence/ForkMigrations/003_ProjectionManagedWorktrees.test.ts similarity index 73% rename from apps/server/src/persistence/Migrations/043_ProjectionManagedWorktrees.test.ts rename to apps/server/src/persistence/ForkMigrations/003_ProjectionManagedWorktrees.test.ts index 7a6417e67dc8..ca6d081f56c4 100644 --- a/apps/server/src/persistence/Migrations/043_ProjectionManagedWorktrees.test.ts +++ b/apps/server/src/persistence/ForkMigrations/003_ProjectionManagedWorktrees.test.ts @@ -4,17 +4,19 @@ import * as Layer from "effect/Layer"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { runMigrations } from "../Migrations.ts"; +import { runForkMigrations } from "../ForkMigrations.ts"; import * as NodeSqliteClient from "../NodeSqliteClient.ts"; const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); -layer("043_ProjectionManagedWorktrees", (it) => { +layer("fork/003_ProjectionManagedWorktrees", (it) => { it.effect("adds nullable managed-worktree provenance storage", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* runMigrations({ toMigrationInclusive: 42 }); - yield* runMigrations({ toMigrationInclusive: 43 }); + yield* runMigrations(); + yield* runForkMigrations({ toMigrationInclusive: 2 }); + yield* runForkMigrations({ toMigrationInclusive: 3 }); const columns = yield* sql<{ readonly name: string }>` PRAGMA table_info(projection_threads) diff --git a/apps/server/src/persistence/Migrations/043_ProjectionManagedWorktrees.ts b/apps/server/src/persistence/ForkMigrations/003_ProjectionManagedWorktrees.ts similarity index 100% rename from apps/server/src/persistence/Migrations/043_ProjectionManagedWorktrees.ts rename to apps/server/src/persistence/ForkMigrations/003_ProjectionManagedWorktrees.ts diff --git a/apps/server/src/persistence/Migrations/044_CleanupCompletedRetractionMessages.test.ts b/apps/server/src/persistence/ForkMigrations/004_CleanupCompletedRetractionMessages.test.ts similarity index 85% rename from apps/server/src/persistence/Migrations/044_CleanupCompletedRetractionMessages.test.ts rename to apps/server/src/persistence/ForkMigrations/004_CleanupCompletedRetractionMessages.test.ts index 5049b7df1194..669139643754 100644 --- a/apps/server/src/persistence/Migrations/044_CleanupCompletedRetractionMessages.test.ts +++ b/apps/server/src/persistence/ForkMigrations/004_CleanupCompletedRetractionMessages.test.ts @@ -4,16 +4,18 @@ import * as Layer from "effect/Layer"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import { runMigrations } from "../Migrations.ts"; +import { runForkMigrations } from "../ForkMigrations.ts"; import * as NodeSqliteClient from "../NodeSqliteClient.ts"; const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); -layer("044_CleanupCompletedRetractionMessages", (it) => { +layer("fork/004_CleanupCompletedRetractionMessages", (it) => { it.effect("removes only messages belonging to completed retractions and is idempotent", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* runMigrations({ toMigrationInclusive: 43 }); + yield* runMigrations(); + yield* runForkMigrations({ toMigrationInclusive: 3 }); yield* sql` INSERT INTO projection_thread_messages ( message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at @@ -50,8 +52,8 @@ layer("044_CleanupCompletedRetractionMessages", (it) => { ) `; - const firstRun = yield* runMigrations({ toMigrationInclusive: 44 }); - assert.deepEqual(firstRun, [[44, "CleanupCompletedRetractionMessages"]]); + const firstRun = yield* runForkMigrations({ toMigrationInclusive: 4 }); + assert.deepEqual(firstRun, [[4, "CleanupCompletedRetractionMessages"]]); const rowsAfterFirstRun = yield* sql<{ readonly messageId: string }>` SELECT message_id AS "messageId" @@ -63,7 +65,7 @@ layer("044_CleanupCompletedRetractionMessages", (it) => { { messageId: "message-unrelated" }, ]); - const secondRun = yield* runMigrations({ toMigrationInclusive: 44 }); + const secondRun = yield* runForkMigrations({ toMigrationInclusive: 4 }); assert.deepEqual(secondRun, []); const rowsAfterSecondRun = yield* sql<{ readonly messageId: string }>` diff --git a/apps/server/src/persistence/Migrations/044_CleanupCompletedRetractionMessages.ts b/apps/server/src/persistence/ForkMigrations/004_CleanupCompletedRetractionMessages.ts similarity index 100% rename from apps/server/src/persistence/Migrations/044_CleanupCompletedRetractionMessages.ts rename to apps/server/src/persistence/ForkMigrations/004_CleanupCompletedRetractionMessages.ts diff --git a/apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts b/apps/server/src/persistence/ForkMigrations/005_RebuildProjectionsFromEvents.test.ts similarity index 78% rename from apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts rename to apps/server/src/persistence/ForkMigrations/005_RebuildProjectionsFromEvents.test.ts index 67c32e8c9a63..600c73a46740 100644 --- a/apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.test.ts +++ b/apps/server/src/persistence/ForkMigrations/005_RebuildProjectionsFromEvents.test.ts @@ -3,15 +3,16 @@ import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as SqlClient from "effect/unstable/sql/SqlClient"; -import { migrationManifest, runMigrations } from "../Migrations.ts"; +import { runMigrations } from "../Migrations.ts"; +import { forkMigrationManifest, runForkMigrations } from "../ForkMigrations.ts"; import rebuildProjectionsFromEvents, { projectionTableNames, -} from "./045_RebuildProjectionsFromEvents.ts"; +} from "./005_RebuildProjectionsFromEvents.ts"; import * as NodeSqliteClient from "../NodeSqliteClient.ts"; const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); -layer("045_RebuildProjectionsFromEvents", (it) => { +layer("fork/005_RebuildProjectionsFromEvents", (it) => { it.effect("clears every event-derived projection and resets cursors idempotently", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -54,21 +55,22 @@ layer("045_RebuildProjectionsFromEvents", (it) => { yield* sql`DROP TABLE projection_thread_proposed_plans`; yield* rebuildProjectionsFromEvents; - // The manifest must end on a full rebuild so boot replays every event - // through the current projectors after the wipe. - assert.deepEqual(migrationManifest.at(-1), [46, "RebuildProjectionsWithRetainedTurns"]); + // The fork manifest must end on a full rebuild so boot replays every + // event through the current projectors after the wipe. + assert.deepEqual(forkMigrationManifest.at(-1), [6, "RebuildProjectionsWithRetainedTurns"]); }), ); }); it.layer(Layer.fresh(Layer.mergeAll(NodeSqliteClient.layerMemory())))( - "045_RebuildProjectionsFromEvents registration", + "fork/005_RebuildProjectionsFromEvents registration", (it) => { - it.effect("runs after migration 044 against the full projection schema", () => + it.effect("runs after fork migration 004 against the full projection schema", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - yield* runMigrations({ toMigrationInclusive: 44 }); + yield* runMigrations(); + yield* runForkMigrations({ toMigrationInclusive: 4 }); yield* sql` INSERT INTO projection_thread_messages ( message_id, thread_id, turn_id, role, text, is_streaming, created_at, updated_at @@ -82,8 +84,8 @@ it.layer(Layer.fresh(Layer.mergeAll(NodeSqliteClient.layerMemory())))( VALUES ('projection.thread-messages', 99, '2026-01-01T00:00:00.000Z') `; - const migrations = yield* runMigrations({ toMigrationInclusive: 45 }); - assert.deepEqual(migrations, [[45, "RebuildProjectionsFromEvents"]]); + const migrations = yield* runForkMigrations({ toMigrationInclusive: 5 }); + assert.deepEqual(migrations, [[5, "RebuildProjectionsFromEvents"]]); const messageRows = yield* sql<{ readonly count: number }>` SELECT COUNT(*) AS count diff --git a/apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.ts b/apps/server/src/persistence/ForkMigrations/005_RebuildProjectionsFromEvents.ts similarity index 100% rename from apps/server/src/persistence/Migrations/045_RebuildProjectionsFromEvents.ts rename to apps/server/src/persistence/ForkMigrations/005_RebuildProjectionsFromEvents.ts diff --git a/apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts b/apps/server/src/persistence/ForkMigrations/006_RebuildProjectionsWithRetainedTurns.ts similarity index 63% rename from apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts rename to apps/server/src/persistence/ForkMigrations/006_RebuildProjectionsWithRetainedTurns.ts index b0d2c101582e..fb365c891812 100644 --- a/apps/server/src/persistence/Migrations/046_RebuildProjectionsWithRetainedTurns.ts +++ b/apps/server/src/persistence/ForkMigrations/006_RebuildProjectionsWithRetainedTurns.ts @@ -1,9 +1,9 @@ -import Migration0045 from "./045_RebuildProjectionsFromEvents.ts"; +import ForkMigration005 from "./005_RebuildProjectionsFromEvents.ts"; -// Re-run migration 045's wipe-and-replay. The first rebuild ran through a +// Re-run fork migration 005's wipe-and-replay. The first rebuild ran through a // turns projector whose thread.reverted handler kept only checkpointed turns, // deleting checkpointless turns and stranding their messages outside // turn-anchored pagination. The projector now shares the revert denylist with // the other projectors, so replaying the event history again rebuilds those // turn rows. -export default Migration0045; +export default ForkMigration005; diff --git a/apps/server/src/persistence/Layers/Sqlite.ts b/apps/server/src/persistence/Layers/Sqlite.ts index ec1ffdefac0f..b2976523c81b 100644 --- a/apps/server/src/persistence/Layers/Sqlite.ts +++ b/apps/server/src/persistence/Layers/Sqlite.ts @@ -5,7 +5,7 @@ import * as Path from "effect/Path"; import * as SqlClient from "effect/unstable/sql/SqlClient"; import type { SqlError } from "effect/unstable/sql/SqlError"; -import { runMigrations } from "../Migrations.ts"; +import { runAllMigrations } from "../ForkMigrations.ts"; import { ServerConfig } from "../../config.ts"; type RuntimeSqliteLayerConfig = { @@ -37,7 +37,7 @@ const setup = Layer.effectDiscard( yield* sql`PRAGMA busy_timeout = 5000;`; yield* sql`PRAGMA foreign_keys = ON;`; yield* sql`PRAGMA journal_mode = WAL;`; - yield* runMigrations(); + yield* runAllMigrations(); }), ); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index e57c82fdd5fe..b137cedfbedd 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -53,12 +53,6 @@ import Migration0037 from "./Migrations/037_ProjectionTurnsKeysetIndex.ts"; import Migration0038 from "./Migrations/038_ProjectionThreadsPinOrderKey.ts"; import Migration0039 from "./Migrations/039_ProjectionProjectsDefaultThreadEnvMode.ts"; import Migration0040 from "./Migrations/040_ProjectionProjectFaviconPath.ts"; -import Migration0041 from "./Migrations/041_ProjectionTurnRetractions.ts"; -import Migration0042 from "./Migrations/042_ProjectionTurnDispatchOwnership.ts"; -import Migration0043 from "./Migrations/043_ProjectionManagedWorktrees.ts"; -import Migration0044 from "./Migrations/044_CleanupCompletedRetractionMessages.ts"; -import Migration0045 from "./Migrations/045_RebuildProjectionsFromEvents.ts"; -import Migration0046 from "./Migrations/046_RebuildProjectionsWithRetainedTurns.ts"; /** * Migration loader with all migrations defined inline. @@ -111,12 +105,6 @@ export const migrationEntries = [ [38, "ProjectionThreadsPinOrderKey", Migration0038], [39, "ProjectionProjectsDefaultThreadEnvMode", Migration0039], [40, "ProjectionProjectFaviconPath", Migration0040], - [41, "ProjectionTurnRetractions", Migration0041], - [42, "ProjectionTurnDispatchOwnership", Migration0042], - [43, "ProjectionManagedWorktrees", Migration0043], - [44, "CleanupCompletedRetractionMessages", Migration0044], - [45, "RebuildProjectionsFromEvents", Migration0045], - [46, "RebuildProjectionsWithRetainedTurns", Migration0046], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/web/src/connection/storage.ts b/apps/web/src/connection/storage.ts index 474e4fc5c2ef..4f59269264cb 100644 --- a/apps/web/src/connection/storage.ts +++ b/apps/web/src/connection/storage.ts @@ -58,12 +58,12 @@ const StoredShellSnapshotJson = Schema.fromJsonString(StoredShellSnapshot); // v4 invalidates snapshots cached before the wire projection retained tool // output fields (result/aggregatedOutput); a warm cache resumes via // `afterSequence` and would otherwise show stripped payloads forever. -// v5 invalidates snapshots cached before server migration 044 deleted +// v5 invalidates snapshots cached before server fork migration 004 deleted // orphaned retracted messages out-of-band: the deletion emits no events, so // a warm cache resuming via `afterSequence` would render the ghost messages // forever. Any server-side row surgery needs a bump here to reach clients. -// v6 pairs with server migration 045's full projection rebuild. -// v7 pairs with server migration 046, which rebuilds again now that the +// v6 pairs with server fork migration 005's full projection rebuild. +// v7 pairs with server fork migration 006, which rebuilds again now that the // turns projector retains checkpointless turns across replayed reverts. const StoredThreadSnapshot = Schema.Struct({ schemaVersion: Schema.Literal(7),