diff --git a/apps/server/src/orchestration-v2/legacy/repairMultiRepoForkWorktrees.test.ts b/apps/server/src/orchestration-v2/legacy/repairMultiRepoForkWorktrees.test.ts new file mode 100644 index 000000000000..3e6e3b906ef3 --- /dev/null +++ b/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) => + 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 = []; + 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: `///`. + 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)), +); diff --git a/apps/server/src/orchestration-v2/legacy/repairMultiRepoForkWorktrees.ts b/apps/server/src/orchestration-v2/legacy/repairMultiRepoForkWorktrees.ts new file mode 100644 index 000000000000..0a9f15b88042 --- /dev/null +++ b/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 `//` + * 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` + 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; +}); diff --git a/apps/server/src/serverRuntimeStartup.autoPull.test.ts b/apps/server/src/serverRuntimeStartup.autoPull.test.ts index 8313d33dc220..67bff5023436 100644 --- a/apps/server/src/serverRuntimeStartup.autoPull.test.ts +++ b/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* () { diff --git a/apps/server/src/serverRuntimeStartup.ts b/apps/server/src/serverRuntimeStartup.ts index 88aa4298c596..5ae83d8cdfdd 100644 --- a/apps/server/src/serverRuntimeStartup.ts +++ b/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(