fix(server): threads from the pre-v2 multi-repo fork reach every repository again

apps/server/src/orchestration-v2/legacy/repairMultiRepoForkWorktrees.test.ts

@@ -0,0 +1,189 @@
+import * as NodeServices from "@effect/platform-node/NodeServices";
+import { assert, it } from "@effect/vitest";
+import { CommandId, type OrchestrationV2ServerCommand, ThreadId } from "@t3tools/contracts";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
+import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
+import * as Layer from "effect/Layer";
+import * as Path from "effect/Path";
+import * as SqlClient from "effect/sql/SqlClient";
+
+import * as ServerConfig from "../../config.ts";
+import * as WorkspaceWorktrees from "../../git/WorkspaceWorktrees.ts";
+import * as GitVcsDriver from "../../vcs/GitVcsDriver.ts";
+import * as VcsProcess from "../../vcs/VcsProcess.ts";
+import * as WorkspaceRepositories from "../../workspace/WorkspaceRepositories.ts";
+import * as ThreadManagement from "../ThreadManagementService.ts";
+import { repairMultiRepoForkWorktrees } from "./repairMultiRepoForkWorktrees.ts";
+
+const git = (cwd: string, args: ReadonlyArray<string>) =>
+  Effect.flatMap(VcsProcess.VcsProcess, (process) =>
+    process.run({ operation: "repairMultiRepoForkWorktrees.test", command: "git", cwd, args }),
+  );
+
+const initRepository = Effect.fn("initRepository")(function* (cwd: string) {
+  const fileSystem = yield* FileSystem.FileSystem;
+  yield* fileSystem.makeDirectory(cwd, { recursive: true });
+  yield* git(cwd, ["init", "--initial-branch=main"]);
+  yield* git(cwd, ["config", "user.email", "test@test.com"]);
+  yield* git(cwd, ["config", "user.name", "Test"]);
+  yield* fileSystem.writeFileString(`${cwd}/README.md`, "# test\n");
+  yield* git(cwd, ["add", "."]);
+  yield* git(cwd, ["commit", "-m", "initial commit"]);
+});
+
+const insertThread = Effect.fn("insertThread")(function* (input: {
+  readonly threadId: string;
+  readonly v1WorktreePath: string;
+  readonly v2WorktreePath: string;
+  readonly worktrees: ReadonlyArray<{ readonly repoRoot: string; readonly worktreePath: string }>;
+}) {
+  const sql = yield* SqlClient.SqlClient;
+  yield* sql`
+    INSERT INTO projection_threads (thread_id, worktree_path, worktrees_json)
+    VALUES (${input.threadId}, ${input.v1WorktreePath}, ${JSON.stringify(input.worktrees)})
+  `;
+  yield* sql`
+    INSERT INTO orchestration_v2_projection_threads (thread_id, deleted_at, payload_json)
+    VALUES (
+      ${input.threadId},
+      NULL,
+      ${JSON.stringify({ worktreePath: input.v2WorktreePath })}
+    )
+  `;
+});
+
+it.effect("moves fork threads still on their first repository to the whole isolated run", () =>
+  Effect.gen(function* () {
+    const fileSystem = yield* FileSystem.FileSystem;
+    const path = yield* Path.Path;
+    const baseDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-fork-worktrees-" });
+    const dispatched: Array<OrchestrationV2ServerCommand> = [];
+    const layer = Layer.mergeAll(
+      WorkspaceWorktrees.layer,
+      WorkspaceRepositories.layer,
+      Layer.mock(ThreadManagement.ThreadManagementService)({
+        dispatch: (command) => {
+          dispatched.push(command);
+          return Effect.succeed({ sequence: dispatched.length, storedEvents: [] });
+        },
+      }),
+      NodeSqliteClient.layer({ filename: ":memory:" }),
+    ).pipe(
+      Layer.provideMerge(GitVcsDriver.layer),
+      Layer.provideMerge(VcsProcess.layer),
+      Layer.provideMerge(ServerConfig.layerTest(baseDir, baseDir)),
+      Layer.provideMerge(NodeServices.layer),
+    );
+
+    yield* Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      const config = yield* ServerConfig.ServerConfig;
+      yield* sql`
+        CREATE TABLE projection_threads (
+          thread_id TEXT PRIMARY KEY,
+          worktree_path TEXT,
+          worktrees_json TEXT NOT NULL DEFAULT '[]'
+        )
+      `;
+      yield* sql`
+        CREATE TABLE orchestration_v2_projection_threads (
+          thread_id TEXT PRIMARY KEY,
+          deleted_at TEXT,
+          payload_json TEXT NOT NULL
+        )
+      `;
+
+      const api = path.join(baseDir, "repos", "api");
+      const web = path.join(baseDir, "repos", "web");
+      yield* initRepository(api);
+      yield* initRepository(web);
+      // The fork's layout: `<worktrees>/<project>/<thread>/<repository>`.
+      const container = path.join(config.worktreesDir, "project-1", "thread-1");
+      const worktrees = [
+        { repoRoot: api, worktreePath: path.join(container, "api") },
+        { repoRoot: web, worktreePath: path.join(container, "web") },
+      ];
+      for (const worktree of worktrees) {
+        yield* git(worktree.repoRoot, [
+          "worktree",
+          "add",
+          "-b",
+          "t3code/thread-1",
+          worktree.worktreePath,
+        ]);
+      }
+      yield* insertThread({
+        threadId: "thread-1",
+        v1WorktreePath: worktrees[0]!.worktreePath,
+        v2WorktreePath: worktrees[0]!.worktreePath,
+        worktrees,
+      });
+      // Moved after the import, so it is left alone.
+      yield* insertThread({
+        threadId: "thread-moved",
+        v1WorktreePath: worktrees[0]!.worktreePath,
+        v2WorktreePath: path.join(config.worktreesDir, "elsewhere"),
+        worktrees,
+      });
+      // An ordinary single-repository isolated run.
+      yield* insertThread({
+        threadId: "thread-single",
+        v1WorktreePath: worktrees[1]!.worktreePath,
+        v2WorktreePath: worktrees[1]!.worktreePath,
+        worktrees: [worktrees[1]!],
+      });
+
+      assert.strictEqual(yield* repairMultiRepoForkWorktrees(), 1);
+      assert.deepStrictEqual(dispatched, [
+        {
+          type: "thread.metadata.update",
+          commandId: CommandId.make("migration:v1:multi-repo-worktrees:thread-1"),
+          threadId: ThreadId.make("thread-1"),
+          worktreePath: container,
+          expectedWorktreePath: worktrees[0]!.worktreePath,
+        },
+      ]);
+
+      const repositories = yield* (yield* WorkspaceRepositories.WorkspaceRepositories).list(
+        container,
+      );
+      assert.deepStrictEqual(
+        repositories.map((repository) => [repository.relativePath, repository.path]),
+        [
+          ["api", path.join(container, "api")],
+          ["web", path.join(container, "web")],
+        ],
+      );
+
+      // Thread deletion can now clean the run up.
+      const workspaceWorktrees = yield* WorkspaceWorktrees.WorkspaceWorktrees;
+      assert.isTrue(yield* workspaceWorktrees.isContainer(container));
+      yield* workspaceWorktrees.remove({ path: container, force: false });
+      assert.isFalse(yield* fileSystem.exists(container));
+      assert.isFalse(
+        yield* fileSystem.exists(WorkspaceRepositories.isolatedRunLayoutPath(container)),
+      );
+    }).pipe(Effect.provide(layer));
+  }).pipe(Effect.provide(NodeServices.layer)),
+);
+
+it.effect("does nothing on databases that never ran the fork", () =>
+  Effect.gen(function* () {
+    const fileSystem = yield* FileSystem.FileSystem;
+    const baseDir = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-fork-worktrees-" });
+    yield* Effect.gen(function* () {
+      const sql = yield* SqlClient.SqlClient;
+      yield* sql`CREATE TABLE projection_threads (thread_id TEXT PRIMARY KEY, worktree_path TEXT)`;
+      assert.strictEqual(yield* repairMultiRepoForkWorktrees(), 0);
+    }).pipe(
+      Effect.provide(
+        Layer.mergeAll(
+          NodeSqliteClient.layer({ filename: ":memory:" }),
+          Layer.mock(ThreadManagement.ThreadManagementService)({}),
+          ServerConfig.layerTest(baseDir, baseDir),
+        ).pipe(Layer.provideMerge(NodeServices.layer)),
+      ),
+    );
+  }).pipe(Effect.provide(NodeServices.layer)),
+);

apps/server/src/orchestration-v2/legacy/repairMultiRepoForkWorktrees.ts

@@ -0,0 +1,110 @@
+/**
+ * Points threads imported from the pre-V2 multi-repo fork at their whole isolated run.
+ *
+ * The fork gave an isolated run one worktree per repository under `<worktrees>/<project>/<thread>`
+ * and recorded them in `projection_threads.worktrees_json`, but its `worktree_path` named only
+ * the first repository's worktree. The V2 import copies `worktree_path`, so those threads came
+ * across as single-repo threads. That per-thread folder already has V2's container layout, so
+ * this writes the container's layout file and moves the thread's worktree path up to it.
+ *
+ * Fork-only; not intended for upstream.
+ *
+ * @module repairMultiRepoForkWorktrees
+ */
+import { CommandId, ThreadId } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
+import * as FileSystem from "effect/FileSystem";
+import * as Path from "effect/Path";
+import * as Schema from "effect/Schema";
+import * as SqlClient from "effect/sql/SqlClient";
+
+import * as ServerConfig from "../../config.ts";
+import * as WorkspaceRepositories from "../../workspace/WorkspaceRepositories.ts";
+import * as ThreadManagement from "../ThreadManagementService.ts";
+
+const ForkWorktrees = Schema.Array(
+  Schema.Struct({ repoRoot: Schema.String, worktreePath: Schema.String }),
+);
+const decodeForkWorktrees = Schema.decodeUnknownOption(Schema.fromJsonString(ForkWorktrees));
+const encodeLayout = Schema.encodeEffect(
+  Schema.fromJsonString(WorkspaceRepositories.IsolatedRunLayout),
+);
+
+interface CandidateRow {
+  readonly thread_id: string;
+  readonly worktree_path: string;
+  readonly worktrees_json: string;
+}
+
+/** Repairs every affected thread that still has its imported worktree path; returns how many. */
+export const repairMultiRepoForkWorktrees = Effect.fn("repairMultiRepoForkWorktrees")(function* () {
+  const sql = yield* SqlClient.SqlClient;
+  const fileSystem = yield* FileSystem.FileSystem;
+  const path = yield* Path.Path;
+  const config = yield* ServerConfig.ServerConfig;
+  const threads = yield* ThreadManagement.ThreadManagementService;
+
+  const columns = yield* sql<{ readonly name: string }>`PRAGMA table_info(projection_threads)`;
+  if (!columns.some((column) => column.name === "worktrees_json")) return 0;
+
+  // A thread whose V2 worktree path moved since the import has been handled already, or the
+  // user moved it, so only threads still on the imported path are candidates.
+  const rows = yield* sql<CandidateRow>`
+    SELECT thread.thread_id, thread.worktree_path, thread.worktrees_json
+    FROM projection_threads AS thread
+    INNER JOIN orchestration_v2_projection_threads AS projection
+      ON projection.thread_id = thread.thread_id
+    WHERE projection.deleted_at IS NULL
+      AND thread.worktree_path IS NOT NULL
+      AND json_extract(projection.payload_json, '$.worktreePath') = thread.worktree_path
+      AND CASE WHEN json_valid(thread.worktrees_json)
+        THEN json_array_length(thread.worktrees_json) ELSE 0 END > 1
+  `;
+
+  let repaired = 0;
+  for (const row of rows) {
+    const worktrees = decodeForkWorktrees(row.worktrees_json);
+    if (worktrees._tag === "None") continue;
+    const container = path.dirname(row.worktree_path);
+    const relativeToWorktrees = path.relative(config.worktreesDir, container);
+    const isRepairable =
+      relativeToWorktrees !== "" &&
+      !relativeToWorktrees.startsWith("..") &&
+      !path.isAbsolute(relativeToWorktrees) &&
+      worktrees.value.some((worktree) => worktree.worktreePath === row.worktree_path) &&
+      worktrees.value.every((worktree) => path.dirname(worktree.worktreePath) === container) &&
+      (yield* fileSystem.exists(container).pipe(Effect.orElseSucceed(() => false)));
+    if (!isRepairable) continue;
+
+    yield* Effect.gen(function* () {
+      // The layout file is what V2 discovery and thread deletion read, so it goes first.
+      const layoutPath = WorkspaceRepositories.isolatedRunLayoutPath(container);
+      if (!(yield* fileSystem.exists(layoutPath))) {
+        const contents = yield* encodeLayout({
+          repositories: worktrees.value.map((worktree) => ({
+            relativePath: path.basename(worktree.worktreePath),
+            name: path.basename(worktree.worktreePath),
+            source: worktree.repoRoot,
+          })),
+        });
+        yield* fileSystem.writeFileString(layoutPath, contents);
+      }
+      yield* threads.dispatch({
+        type: "thread.metadata.update",
+        commandId: CommandId.make(`migration:v1:multi-repo-worktrees:${row.thread_id}`),
+        threadId: ThreadId.make(row.thread_id),
+        worktreePath: container,
+        expectedWorktreePath: row.worktree_path,
+      });
+      repaired += 1;
+    }).pipe(
+      Effect.catch((cause) =>
+        Effect.logWarning("Could not repair a multi-repo fork thread's worktree path", {
+          threadId: row.thread_id,
+          cause,
+        }),
+      ),
+    );
+  }
+  return repaired;
+});

apps/server/src/serverRuntimeStartup.autoPull.test.ts

@@ -1,6 +1,8 @@
 import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
+import * as NodeFileSystem from "@effect/platform-node/NodeFileSystem";
 import { expect, it } from "@effect/vitest";
 import { DEFAULT_SERVER_SETTINGS, EnvironmentId, ProjectId } from "@t3tools/contracts";
+import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
 import { DEFAULT_SIGNAL_EXPORT } from "@t3tools/shared/observability";
 import * as OtelEnvironment from "@t3tools/shared/otelEnvironment";
 import * as Deferred from "effect/Deferred";
@@ -170,6 +172,8 @@ it.effect("parks automatic pull until activation without delaying command readin
           address: NetAddress.inetAddressFromIpStringUnsafe("127.0.0.1", 3773),
         }),
         Path.layer,
+        NodeFileSystem.layer,
+        Layer.orDie(NodeSqliteClient.layer({ filename: ":memory:" })),
       );
 
       yield* Effect.gen(function* () {

apps/server/src/serverRuntimeStartup.ts

@@ -32,6 +32,7 @@ import * as Keybindings from "./keybindings.ts";
 import * as ExternalLauncher from "./process/externalLauncher.ts";
 import * as EffectWorker from "./orchestration-v2/EffectWorker.ts";
 import * as LegacyV1ThreadImporter from "./orchestration-v2/legacy/LegacyV1ThreadImporter.ts";
+import { repairMultiRepoForkWorktrees } from "./orchestration-v2/legacy/repairMultiRepoForkWorktrees.ts";
 import * as Orchestrator from "./orchestration-v2/Orchestrator.ts";
 import * as ProviderRuntimeRecovery from "./orchestration-v2/ProviderRuntimeRecoveryService.ts";
 import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts";
@@ -537,6 +538,21 @@ const make = (options?: StartupOptions) =>
         ).pipe(Effect.map((targets): AutoBootstrapWelcomeTargets => targets)),
       });
       yield* Effect.logInfo("V2 orchestration recovery completed", recovery);
+      yield* runStartupPhase(
+        "orchestration-v2.legacy-v1.multi-repo-fork-worktrees",
+        repairMultiRepoForkWorktrees().pipe(
+          Effect.tap((repaired) =>
+            repaired === 0
+              ? Effect.void
+              : Effect.logInfo("Moved multi-repo fork threads to their isolated runs", {
+                  repaired,
+                }),
+          ),
+          Effect.catch((cause) =>
+            Effect.logWarning("Failed to repair multi-repo fork thread worktrees", { cause }),
+          ),
+        ),
+      );
       // Runs after activation: the status check fetches every enabled project's
       // remote, and awaiting it here held command readiness for that long.
       yield* runStartupPhase(