fix: repair legacy desktop upgrade migration and replay crashes

apps/server/src/persistence/Migrations.ts

@@ -36,6 +36,7 @@ import Migration0020 from "./Migrations/020_ProjectionThreadsForkSource.ts";
 import Migration0021 from "./Migrations/021_ProjectionThreadsAssociatedWorktree.ts";
 import Migration0022 from "./Migrations/022_ProjectionThreadsAssociatedWorktreeBranch.ts";
 import Migration0023 from "./Migrations/023_ProjectionThreadsAssociatedWorktreeRef.ts";
+import Migration0024 from "./Migrations/024_NormalizeLegacyArchivedThreadEvents.ts";
 
 /**
  * Migration loader with all migrations defined inline.
@@ -71,6 +72,7 @@ export const migrationEntries = [
   [21, "ProjectionThreadsAssociatedWorktree", Migration0021],
   [22, "ProjectionThreadsAssociatedWorktreeBranch", Migration0022],
   [23, "ProjectionThreadsAssociatedWorktreeRef", Migration0023],
+  [24, "NormalizeLegacyArchivedThreadEvents", Migration0024],
 ] as const;
 
 export const makeMigrationLoader = (throughId?: number) =>

apps/server/src/persistence/Migrations/023_ProjectionThreadsAssociatedWorktreeRef.test.ts

@@ -0,0 +1,202 @@
+import { assert, it } from "@effect/vitest";
+import { Effect, Layer } from "effect";
+import * as SqlClient from "effect/unstable/sql/SqlClient";
+
+import { runMigrations } from "../Migrations.ts";
+import * as NodeSqliteClient from "../NodeSqliteClient.ts";
+
+const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory()));
+
+const legacyMigrationLedgerRows = [
+  [17, "ProjectionThreadsArchivedAt"],
+  [18, "ProjectionThreadsArchivedAtIndex"],
+  [19, "ProjectionSnapshotLookupIndexes"],
+  [20, "AuthAccessManagement"],
+  [21, "AuthSessionClientMetadata"],
+  [22, "AuthSessionLastConnectedAt"],
+] as const;
+
+layer("023_ProjectionThreadsAssociatedWorktreeRef", (it) => {
+  it.effect(
+    "repairs legacy migration ledgers that skipped the current thread metadata columns",
+    () =>
+      Effect.gen(function* () {
+        const sql = yield* SqlClient.SqlClient;
+
+        yield* runMigrations({ toMigrationInclusive: 16 });
+
+        yield* sql`
+        INSERT INTO projection_projects (
+          project_id,
+          title,
+          workspace_root,
+          scripts_json,
+          default_model_selection_json,
+          created_at,
+          updated_at,
+          deleted_at
+        )
+        VALUES (
+          'project-legacy',
+          'Legacy project',
+          '/tmp/project-legacy',
+          '[]',
+          '{"provider":"codex","model":"gpt-5.4"}',
+          '2026-01-01T00:00:00.000Z',
+          '2026-01-01T00:00:00.000Z',
+          NULL
+        )
+      `;
+
+        yield* sql`
+        INSERT INTO projection_threads (
+          thread_id,
+          project_id,
+          title,
+          model_selection_json,
+          runtime_mode,
+          interaction_mode,
+          branch,
+          worktree_path,
+          latest_turn_id,
+          created_at,
+          updated_at,
+          deleted_at
+        )
+        VALUES
+          (
+            'thread-worktree',
+            'project-legacy',
+            'Worktree thread',
+            '{"provider":"codex","model":"gpt-5.4"}',
+            'full-access',
+            'default',
+            'feature/fix-upgrade',
+            '/tmp/project-legacy/.worktrees/thread-worktree',
+            NULL,
+            '2026-01-01T00:00:00.000Z',
+            '2026-01-01T00:00:00.000Z',
+            NULL
+          ),
+          (
+            'thread-local',
+            'project-legacy',
+            'Local thread',
+            '{"provider":"codex","model":"gpt-5.4"}',
+            'full-access',
+            'default',
+            'main',
+            NULL,
+            NULL,
+            '2026-01-01T00:00:00.000Z',
+            '2026-01-01T00:00:00.000Z',
+            NULL
+          )
+      `;
+
+        yield* sql`
+        INSERT INTO projection_thread_messages (
+          message_id,
+          thread_id,
+          turn_id,
+          role,
+          text,
+          attachments_json,
+          is_streaming,
+          created_at,
+          updated_at
+        )
+        VALUES (
+          'message-legacy',
+          'thread-worktree',
+          'turn-legacy',
+          'assistant',
+          'hello from the past',
+          NULL,
+          0,
+          '2026-01-01T00:00:00.000Z',
+          '2026-01-01T00:00:00.000Z'
+        )
+      `;
+
+        for (const [migrationId, name] of legacyMigrationLedgerRows) {
+          yield* sql`
+          INSERT INTO effect_sql_migrations (migration_id, created_at, name)
+          VALUES (${migrationId}, '2026-01-01T00:00:00.000Z', ${name})
+        `;
+        }
+
+        const executedMigrations = yield* runMigrations({ toMigrationInclusive: 23 });
+
+        assert.deepStrictEqual(executedMigrations, [
+          [23, "ProjectionThreadsAssociatedWorktreeRef"],
+        ]);
+
+        const threadRows = yield* sql<{
+          readonly threadId: string;
+          readonly envMode: string;
+          readonly associatedWorktreePath: string | null;
+          readonly associatedWorktreeBranch: string | null;
+          readonly associatedWorktreeRef: string | null;
+          readonly forkSourceThreadId: string | null;
+          readonly handoff: string | null;
+        }>`
+        SELECT
+          thread_id AS "threadId",
+          env_mode AS "envMode",
+          associated_worktree_path AS "associatedWorktreePath",
+          associated_worktree_branch AS "associatedWorktreeBranch",
+          associated_worktree_ref AS "associatedWorktreeRef",
+          fork_source_thread_id AS "forkSourceThreadId",
+          handoff_json AS "handoff"
+        FROM projection_threads
+        ORDER BY thread_id ASC
+      `;
+
+        assert.deepStrictEqual(threadRows, [
+          {
+            threadId: "thread-local",
+            envMode: "local",
+            associatedWorktreePath: null,
+            associatedWorktreeBranch: "main",
+            associatedWorktreeRef: "main",
+            forkSourceThreadId: null,
+            handoff: null,
+          },
+          {
+            threadId: "thread-worktree",
+            envMode: "worktree",
+            associatedWorktreePath: "/tmp/project-legacy/.worktrees/thread-worktree",
+            associatedWorktreeBranch: "feature/fix-upgrade",
+            associatedWorktreeRef: "feature/fix-upgrade",
+            forkSourceThreadId: null,
+            handoff: null,
+          },
+        ]);
+
+        const messageRows = yield* sql<{
+          readonly messageId: string;
+          readonly skills: string | null;
+          readonly mentions: string | null;
+          readonly source: string;
+        }>`
+        SELECT
+          message_id AS "messageId",
+          skills_json AS "skills",
+          mentions_json AS "mentions",
+          source
+        FROM projection_thread_messages
+        ORDER BY message_id ASC
+      `;
+
+        assert.deepStrictEqual(messageRows, [
+          {
+            messageId: "message-legacy",
+            skills: null,
+            mentions: null,
+            source: "native",
+          },
+        ]);
+      }),
+  );
+});

apps/server/src/persistence/Migrations/023_ProjectionThreadsAssociatedWorktreeRef.ts

@@ -8,15 +8,109 @@ import * as Effect from "effect/Effect";
 export default Effect.gen(function* () {
   const sql = yield* SqlClient.SqlClient;
 
-  yield* sql`
-    ALTER TABLE projection_threads
-    ADD COLUMN associated_worktree_ref TEXT
-  `.pipe(Effect.catchTag("SqlError", () => Effect.void));
-
-  yield* sql`
-    UPDATE projection_threads
-    SET associated_worktree_ref = associated_worktree_branch
-    WHERE associated_worktree_ref IS NULL
-      AND associated_worktree_branch IS NOT NULL
-  `;
+  // Some legacy desktop databases already recorded different 17-22 migration IDs,
+  // so this repair step must rebuild the current thread metadata shape before 23 can finish.
+  const projectionThreadsColumnExists = (columnName: string) =>
+    sql<{ readonly exists: number }>`
+      SELECT EXISTS(
+        SELECT 1
+        FROM pragma_table_info('projection_threads')
+        WHERE name = ${columnName}
+      ) AS "exists"
+    `.pipe(Effect.map(([row]) => row?.exists === 1));
+
+  const projectionThreadMessagesColumnExists = (columnName: string) =>
+    sql<{ readonly exists: number }>`
+      SELECT EXISTS(
+        SELECT 1
+        FROM pragma_table_info('projection_thread_messages')
+        WHERE name = ${columnName}
+      ) AS "exists"
+    `.pipe(Effect.map(([row]) => row?.exists === 1));
+
+  const ensureProjectionThreadsColumn = (columnName: string, definition: string) =>
+    Effect.gen(function* () {
+      const exists = yield* projectionThreadsColumnExists(columnName);
+      if (exists) {
+        return false;
+      }
+
+      yield* sql.unsafe(`
+        ALTER TABLE projection_threads
+        ADD COLUMN ${definition}
+      `);
+      return true;
+    });
+
+  const ensureProjectionThreadMessagesColumn = (columnName: string, definition: string) =>
+    Effect.gen(function* () {
+      const exists = yield* projectionThreadMessagesColumnExists(columnName);
+      if (exists) {
+        return false;
+      }
+
+      yield* sql.unsafe(`
+        ALTER TABLE projection_thread_messages
+        ADD COLUMN ${definition}
+      `);
+      return true;
+    });
+
+  yield* ensureProjectionThreadsColumn("handoff_json", "handoff_json TEXT");
+  yield* ensureProjectionThreadMessagesColumn("source", "source TEXT NOT NULL DEFAULT 'native'");
+  yield* ensureProjectionThreadMessagesColumn("skills_json", "skills_json TEXT");
+  yield* ensureProjectionThreadMessagesColumn("mentions_json", "mentions_json TEXT");
+
+  const addedEnvMode = yield* ensureProjectionThreadsColumn(
+    "env_mode",
+    "env_mode TEXT NOT NULL DEFAULT 'local'",
+  );
+  if (addedEnvMode) {
+    yield* sql`
+      UPDATE projection_threads
+      SET env_mode = CASE
+        WHEN worktree_path IS NOT NULL THEN 'worktree'
+        ELSE 'local'
+      END
+    `;
+  }
+
+  yield* ensureProjectionThreadsColumn("fork_source_thread_id", "fork_source_thread_id TEXT");
+
+  const addedAssociatedWorktreePath = yield* ensureProjectionThreadsColumn(
+    "associated_worktree_path",
+    "associated_worktree_path TEXT",
+  );
+  if (addedAssociatedWorktreePath) {
+    yield* sql`
+      UPDATE projection_threads
+      SET associated_worktree_path = worktree_path
+      WHERE associated_worktree_path IS NULL
+    `;
+  }
+
+  const addedAssociatedWorktreeBranch = yield* ensureProjectionThreadsColumn(
+    "associated_worktree_branch",
+    "associated_worktree_branch TEXT",
+  );
+  if (addedAssociatedWorktreeBranch) {
+    yield* sql`
+      UPDATE projection_threads
+      SET associated_worktree_branch = branch
+      WHERE associated_worktree_branch IS NULL
+    `;
+  }
+
+  const addedAssociatedWorktreeRef = yield* ensureProjectionThreadsColumn(
+    "associated_worktree_ref",
+    "associated_worktree_ref TEXT",
+  );
+  if (addedAssociatedWorktreeRef) {
+    yield* sql`
+      UPDATE projection_threads
+      SET associated_worktree_ref = COALESCE(associated_worktree_branch, branch)
+      WHERE associated_worktree_ref IS NULL
+        AND COALESCE(associated_worktree_branch, branch) IS NOT NULL
+    `;
+  }
 });