diff --git a/doc/plans/2026-11-07-workspace-cohort-capture.md b/doc/plans/2026-11-07-workspace-cohort-capture.md new file mode 100644 index 00000000000..ab731ba6031 --- /dev/null +++ b/doc/plans/2026-11-07-workspace-cohort-capture.md @@ -0,0 +1,107 @@ +# Bounded workspace cohort capture + +`migrateRemoteWorkspaceCohort` in `server/src/services/remote-workspace-migration.ts` +imports one through eight leaf workspaces from one proven physical source +component. It uses the same online source, lifecycle, recovery, archive, storage, +and publication owners as `migrateRemoteWorkspace`. It does not stop unrelated +work or replay a provider turn. + +```ts +await migrateRemoteWorkspaceCohort({ + db, + storage, + companyId, + workspaceIds, + apply: true, + concurrency: 2, + signal, +}); +``` + +An apply caller must supply an original `createDb` pool and an `AbortSignal`. +`concurrency` defaults to two and accepts values from one through eight. +`apply: false` checks the selected membership without reserving operations or +capturing files or using storage. A verified read-only transaction can supply +this observation's database reads. It does not attest held source fences or +promise that a later apply will remain eligible. Its generated cohort ID is not +an apply token; a saved journal's complete membership remains authoritative. + +## Execution and resource limits + +Each member must have the same complete project, issue, and workspace source +component. A selected root cannot contain another selected root. Aliases of the +same root are allowed. Existing nested-import and recovery blockers still apply. +The batch acquires an independent complete source fence, all component lifecycle +locks, and the union of required physical copyback locks. It refuses to borrow +an enclosing transaction because a savepoint cannot confirm the outer commit. + +Reservation runs serially on the original pool. Capture, encryption, upload, +authenticated readback, restore, and recapture run concurrently on that pool. +Under the complete fence, preflight compares every selected workspace fingerprint +with the aggregate source proof's full preimage before the first reservation. +Each reservation repeats its member check before writing a journal. Earlier +source changes refuse the attempt; a fresh attempt can replan when no journal +was reserved. Source changes after a partial reservation remain blocked by the +saved aggregate proof. +One nested-child verifier belongs to the complete cohort fence. Its prepare +calls are queued; each call retains the existing two-child bound and completed +authenticated immutable proof. Every member still rechecks current local bytes, +Git metadata, database references and recovery bindings. The verifier closes +only after all member work joins, before the fence can leave. +The caller must budget for the number of logical captures, even though there is +only one Node process. Include up to two nested-child verification workers in +addition to the configured member concurrency. The first hosted qualification +uses two same-component leaves without imported children. Eight captures can use much more scratch space, CPU, +storage bandwidth, and database capacity than one capture. Start with a +two-member canary and measure the actual host before increasing concurrency. + +The first failure aborts the shared internal signal. The batch stops taking new +members and waits for every active capture and its scratch cleanup. It retains +the complete fence until those peers settle. Cancellation does not promise an +instant return from a storage callback that ignores its signal. + +## Publication and resume + +Each reserved operation stores the same versioned cohort identifier, logical +source fingerprint, and exact sorted workspace/operation membership. Only those +exact journals are admitted together. Other running or uncertain journals block +the batch. Every saved pending member must still be running and unfinished; +cancelled, failed or finished journals are refused before either preflight +eligibility or snapshot work. Publication repeats that check. The owner does +not reset or reopen those journals. A single-workspace import refuses a cohort +journal. + +After capture, the owner transaction locks and freshly validates every member, +source, journal, recovery binding, and uploaded snapshot before it writes any +workspace authority. It then publishes members serially. The outer commit must +succeed before the batch returns success or publishes live activity events. +The originals remain intact. + +After a failed or interrupted attempt, call the batch again with the **same full +workspace ID set**. A persisted member journal supplies the original cohort and +preallocated operation IDs, including an interruption halfway through +reservation. A changed or incomplete membership is rejected. Fresh physical +source checks still run after a process or volume handover; the durable logical +fingerprint does not treat an inode number as lasting authority. + +An uncertain commit must first be observed through the canonical inventory and +journals. An entirely remote set is an idempotent replay only when its selected +snapshots and succeeded journals prove the same complete cohort membership. +A partial, unrelated or changed committed set is refused. A mixed remote/local +set is refused for explicit investigation. The API does not repair or clear +foreign journals, recovery actions, or remote provider obligations. + +## Local proof + +The focused PostgreSQL and real filesystem suite exercises overlapping uploads, +held source and copyback fences, cold restore after original removal, foreign +journals, journal mutation, partial reservation, caller cancellation, peer +failure, fence connection loss, and an actual deferred outer-commit failure. +The existing single-workspace and recovery suites remain part of the affected +regression check. Local macOS evidence does not replace a source Linux/Node 24 +two-member canary or prove a speedup. An earlier six-run macOS comparison was +inconclusive under host contention, including slower cohort runs. Its twelve +cold restores passed, but it is not performance acceptance. Hosted use remains +opt-in: measure two members on the same host, including peak memory and scratch +headroom, before considering a larger cohort. This API adds no caller, scheduler +or automatic batch selection to the running app. diff --git a/server/src/services/remote-workspace-cohort-migration.test.ts b/server/src/services/remote-workspace-cohort-migration.test.ts new file mode 100644 index 00000000000..7f5c932933a --- /dev/null +++ b/server/src/services/remote-workspace-cohort-migration.test.ts @@ -0,0 +1,402 @@ +import { execFile } from "node:child_process"; +import { createHash, randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { promisify } from "node:util"; +import { eq, inArray, sql } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; +import { activityLog, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, nativeRunFinalizations, executionWorkspaceSnapshots, executionWorkspaces, issues, projects, workspaceOperations, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { resolveManagedProjectWorkspaceDir } from "../home-paths.js"; +import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { createStorageService } from "../storage/service.js"; +import type { StorageService } from "../storage/types.js"; +import { migrateRemoteWorkspace, migrateRemoteWorkspaceCohort } from "./remote-workspace-migration.js"; +import { inspectOnlineWorkspaceSource } from "./online-workspace-source.js"; +import { IsolatedRepositorySourceBusy, withWorkspaceSourceLocks } from "./workspace-source-lock.js"; +import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js"; +import { tryLockNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalization-ownership.js"; +import { subscribeCompanyLiveEvents } from "./live-events.js"; +import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; +import * as workspaceArchive from "./remote-workspace-snapshot-archive.js"; +import * as onlineWorkspaceSource from "./online-workspace-source.js"; + +const support = await getEmbeddedPostgresTestSupport(), execute = promisify(execFile); +const git = async (cwd: string, ...args: string[]) => (await execute("git", ["-c", `core.hooksPath=${os.devNull}`, "-C", cwd, ...args], { + env: { ...process.env, GIT_CONFIG_NOSYSTEM: "1", GIT_CONFIG_GLOBAL: os.devNull, GIT_TERMINAL_PROMPT: "0", + GIT_AUTHOR_NAME: "Cohort", GIT_AUTHOR_EMAIL: "cohort@example.test", GIT_COMMITTER_NAME: "Cohort", GIT_COMMITTER_EMAIL: "cohort@example.test" }, +})).stdout.trim(); + +describe.skipIf(!support.supported)("bounded workspace cohort migration", () => { + let database: Awaited>, db: ReturnType, home: string, storage: StorageService; + beforeAll(async () => { database = await startEmbeddedPostgresTestDatabase("workspace-cohort-"); db = createDb(database.connectionString, { maxConnections: 8 }); }, 90_000); + afterAll(async () => { await db?.$client.end({ timeout: 1 }); await database?.cleanup(); }); + beforeEach(async () => { + home = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-cohort-"))); + vi.stubEnv("PAPERCLIP_HOME", home); vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "61".repeat(32)); + storage = createStorageService(createLocalDiskStorageProvider(path.join(home, "objects"))); + }); + afterEach(async () => { + vi.restoreAllMocks(); + await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); + await db.execute(sql`drop trigger if exists cohort_publish_probe on execution_workspaces`); + await db.execute(sql`drop function if exists cohort_publish_probe()`); await db.execute(sql`drop sequence if exists cohort_publish_attempts`); + await db.execute(sql`drop trigger if exists cohort_commit_fault on workspace_operations`); await db.execute(sql`drop function if exists cohort_commit_fault()`); + await db.execute(sql`drop trigger if exists cohort_reserve_fault on workspace_operations`); await db.execute(sql`drop function if exists cohort_reserve_fault()`); + await db.delete(nativeRunFinalizations); await db.delete(environmentLeases); await db.delete(environments); await db.delete(heartbeatRuns); + await db.delete(issues); await db.delete(agents); await db.delete(executionWorkspaces); await db.delete(projects); await db.delete(companies); + vi.unstubAllEnvs(); await fs.rm(home, { recursive: true, force: true }); + }); + async function fixture() { + const companyId = randomUUID(), projectId = randomUUID(), root = resolveManagedProjectWorkspaceDir({ companyId, projectId }); + await db.insert(companies).values({ id: companyId, name: "Cohort", issuePrefix: `C${companyId.slice(0, 6)}` }); + await db.insert(projects).values({ id: projectId, companyId, name: "Shared Git" }); + await fs.mkdir(root, { recursive: true }); await fs.writeFile(path.join(root, "saved.txt"), "committed base\n"); + await git(root, "init", "-qb", "main"); await git(root, "add", "."); await git(root, "commit", "-qm", "Original history"); + const repoUrl = "https://example.invalid/cohort.git"; await git(root, "remote", "add", "origin", repoUrl); + const members = []; + for (const name of ["base", "one", "two"]) { + const cwd = name === "base" ? root : path.join(root, ".paperclip", "worktrees", name); + if (name !== "base") await git(root, "worktree", "add", "-qb", name, cwd); + const [issue] = await db.insert(issues).values({ companyId, projectId, title: name }).returning(); + const [workspace] = await db.insert(executionWorkspaces).values({ companyId, projectId, sourceIssueId: issue.id, name, + mode: name === "base" ? "shared_workspace" : "isolated_workspace", strategyType: name === "base" ? "project_primary" : "git_worktree", + providerType: name === "base" ? "local_fs" : "git_worktree", cwd, providerRef: cwd, repoUrl, branchName: name === "base" ? "main" : name }).returning(); + await db.update(issues).set({ executionWorkspaceId: workspace.id }).where(eq(issues.id, issue.id)); + if (name !== "base") { await fs.writeFile(path.join(cwd, "saved.txt"), `${name} dirty tracked work\n`); await fs.writeFile(path.join(cwd, "untracked.txt"), `${name} untracked work\n`); } + members.push(workspace); + } + return { companyId, projectId, root, parent: members[0], leaves: members.slice(1) }; + } + const run = (f: Awaited>, extra: Partial[0]> = {}) => + migrateRemoteWorkspaceCohort({ db, storage, companyId: f.companyId, workspaceIds: f.leaves.map(row => row.id), + apply: true, concurrency: 2, signal: AbortSignal.timeout(50_000), ...extra }); + const rows = (f: Awaited>) => db.select().from(executionWorkspaces).where(inArray(executionWorkspaces.id, f.leaves.map(row => row.id))); + async function assertLocal(f: Awaited>) { + expect((await rows(f)).every(row => !readRemoteWorkspaceAuthority(row.metadata))).toBe(true); + for (const member of f.leaves) expect(await fs.readFile(path.join(member.cwd!, "saved.txt"), "utf8")).toBe(`${member.name} dirty tracked work\n`); + } + async function recoverable(f: Awaited>) { + const workspace = f.leaves[0], runId = randomUUID(), leaseId = randomUUID(), providerLeaseId = randomUUID(); + const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Held recovery", adapterType: "paperclip_runner" }).returning(); + const [environment] = await db.insert(environments).values({ name: runId, driver: "daytona" }).returning(); + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: agent.id, nativeIssueId: workspace.sourceIssueId, + runtimeMode: "native", status: "failed", runnerProfileJson: { nativeWorkspaceSync: { schema: "paperclip.native-workspace-sync/v3", + state: "prepared", workspaceId: workspace.id, leaseId, providerLeaseId, remoteCwd: "/workspaces/held", + descriptorSha256: "12".repeat(32), baselineSha256: "34".repeat(32), resourceDisposition: "keep_running" } } }); + await db.insert(environmentLeases).values({ id: leaseId, companyId: f.companyId, issueId: workspace.sourceIssueId, + heartbeatRunId: runId, executionWorkspaceId: workspace.id, environmentId: environment.id, provider: "daytona", providerLeaseId, + status: "retained", leasePolicy: "retain_on_failure", metadata: { remoteCwd: "/workspaces/held" } }); + const [owner] = await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: workspace.sourceIssueId!, + runId, phase: "terminal_failure", failureCode: "native_session_interrupted" }).returning(); + return { runId, owner }; + } + it("captures two shared-Git leaves concurrently and restores both after outer commit", async () => { + const f = await fixture(), held = await recoverable(f), workspaceIds = f.leaves.map(row => row.id), put = storage.putFile.bind(storage); + let entered = 0, active = 0, maximum = 0, release!: () => void; + const barrier = new Promise(resolve => { release = resolve; }), timer = setTimeout(() => release(), 2000); + vi.spyOn(storage, "putFile").mockImplementation(async input => { + active++; maximum = Math.max(maximum, active); if (++entered === 2) release(); + if (entered === 1) { + const other = createDb(database.connectionString, { maxConnections: 1 }); + try { + await expect(withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => true, true)) + .rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); + await other.transaction(async tx => expect(await tryLockNativeWorkspaceFinalization(tx as unknown as Db, { companyId: f.companyId, runId: held.runId })).toBe(false)); + } finally { await other.$client.end({ timeout: 1 }); } + } + try { + await barrier; + const rows = await db.select().from(executionWorkspaces).where(inArray(executionWorkspaces.id, workspaceIds)); + expect(rows.every(row => !readRemoteWorkspaceAuthority(row.metadata))).toBe(true); + return await put(input); + } finally { active--; } + }); + try { + await expect(migrateRemoteWorkspaceCohort({ db, storage, companyId: f.companyId, workspaceIds, + apply: true, signal: AbortSignal.timeout(50_000) })).resolves.toMatchObject({ applied: true }); + } finally { clearTimeout(timer); } + expect(maximum).toBe(2); + await expect(run(f)).resolves.toMatchObject({ applied: true, replayed: true }); + await expect(run(f, { workspaceIds: [f.leaves[0].id] })).rejects.toThrow("workspace_cohort_membership_changed"); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, held.runId)))[0]).toEqual(held.owner); + const archives = []; + for (const workspace of f.leaves) { + const [row] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspace.id)); + const authority = readRemoteWorkspaceAuthority(row.metadata)!; + const ref = (await getWorkspaceSnapshot(db, { companyId: f.companyId, workspaceId: workspace.id, snapshotId: authority.snapshotId! }))!; + const archivePath = path.join(home, `${workspace.name}.pcws`); await loadWorkspaceSnapshotArchive({ storage, reference: ref, targetPath: archivePath }); + archives.push({ workspace, ref, archivePath }); + } + await fs.rm(f.root, { recursive: true }); + for (const { workspace, ref, archivePath } of archives) { + const targetDir = path.join(home, `${workspace.name}-restored`); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe(`${workspace.name} dirty tracked work\n`); + expect(await fs.readFile(path.join(targetDir, "untracked.txt"), "utf8")).toBe(`${workspace.name} untracked work\n`); + expect(await git(targetDir, "log", "-1", "--format=%s")).toBe("Original history"); + } + }, 60_000); + it("shares the merged child verifier across parallel aliases under the complete cohort fence", async () => { + const f = await fixture(), parent = f.leaves[0], cwd = path.join(parent.cwd!, ".paperclip", "worktrees", "nested"); + await git(f.root, "worktree", "add", "-qb", "nested", cwd); + const [childIssue, aliasIssue] = await db.insert(issues).values([ + { companyId: f.companyId, projectId: f.projectId, title: "Nested child" }, + { companyId: f.companyId, projectId: f.projectId, title: "Leaf alias" }, + ]).returning(); + const [child, alias] = await db.insert(executionWorkspaces).values([ + { companyId: f.companyId, projectId: f.projectId, sourceIssueId: childIssue.id, name: "nested", mode: "isolated_workspace", + strategyType: "git_worktree", providerType: "git_worktree", cwd, providerRef: cwd, repoUrl: parent.repoUrl, branchName: "nested" }, + { companyId: f.companyId, projectId: f.projectId, sourceIssueId: aliasIssue.id, name: "alias", mode: parent.mode, + strategyType: parent.strategyType, providerType: parent.providerType, cwd: parent.cwd, providerRef: parent.providerRef, + repoUrl: parent.repoUrl, branchName: parent.branchName }, + ]).returning(); + await db.update(issues).set({ executionWorkspaceId: child.id }).where(eq(issues.id, childIssue.id)); + await db.update(issues).set({ executionWorkspaceId: alias.id }).where(eq(issues.id, aliasIssue.id)); + await fs.writeFile(path.join(cwd, "untracked.txt"), "nested dirty bytes\n"); + await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: child.id, apply: true, + signal: AbortSignal.timeout(50_000) }); + const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.executionWorkspaceId, child.id)); + const get = storage.getObject.bind(storage), head = storage.headObject.bind(storage); + let reads = 0; + vi.spyOn(storage, "headObject").mockImplementation(async (...args) => { + const result = await head(...args); + if (!result.exists) return result; + const bytes = await fs.readFile(path.join(home, "objects", args[1])); + return { ...result, etag: `"${createHash("sha256").update(bytes).digest("hex")}"` }; + }); + vi.spyOn(storage, "getObject").mockImplementation(async (...args) => { + if (args[1] === snapshot.objectKey) reads++; + return get(...args); + }); + const capture = vi.spyOn(workspaceArchive, "captureRemoteWorkspaceDirectory"); + await expect(run(f, { workspaceIds: [parent.id, alias.id] })).resolves.toMatchObject({ applied: true }); + expect(reads).toBe(1); + expect(capture.mock.calls.filter(([input]) => input.cwd === cwd && path.basename(input.archivePath) === "current.pcws")).toHaveLength(4); + expect(await fs.readFile(path.join(cwd, "untracked.txt"), "utf8")).toBe("nested dirty bytes\n"); + }, 60_000); + it.each(["foreign", "journal", "finished"])("refuses a changed %s operation before any authority write", async mode => { + const f = await fixture(), put = storage.putFile.bind(storage); let uploaded = 0; + await db.execute(sql`create sequence cohort_publish_attempts minvalue 0 start 0`); + await db.execute(sql`create function cohort_publish_probe() returns trigger language plpgsql as $$ begin perform nextval('cohort_publish_attempts'); return new; end $$`); + await db.execute(sql`create trigger cohort_publish_probe before update on execution_workspaces for each row when (new.metadata->'remoteWorkspace' is not null) execute function cohort_publish_probe()`); + vi.spyOn(storage, "putFile").mockImplementation(async input => { + const result = await put(input); + if (++uploaded === 2) { + if (mode === "foreign") await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.parent.id, + issueId: f.parent.sourceIssueId, cwd: f.root, phase: "worktree_prepare", status: "running" }); + else if (mode === "finished") await db.update(workspaceOperations).set({ finishedAt: new Date() }) + .where(eq(workspaceOperations.executionWorkspaceId, f.leaves[1].id)); + else await db.update(workspaceOperations).set({ metadata: sql`${workspaceOperations.metadata} || '{"unexpected":"changed journal"}'::jsonb` }) + .where(eq(workspaceOperations.executionWorkspaceId, f.leaves[1].id)); + } + return result; + }); + await expect(run(f)).rejects.toThrow(); + expect((await db.execute(sql`select is_called from cohort_publish_attempts`))[0].is_called).toBe(false); + expect(await db.select().from(activityLog)).toHaveLength(0); + await assertLocal(f); + expect(await db.select().from(workspaceOperations)).toHaveLength(mode === "foreign" ? 3 : 2); + }); + it("resumes the exact membership after reservation stops halfway", async () => { + const f = await fixture(), second = [...f.leaves].sort((a, b) => a.id.localeCompare(b.id))[1]; + await db.execute(sql.raw(`create function cohort_reserve_fault() returns trigger language plpgsql as $$ begin if new.execution_workspace_id='${second.id}'::uuid then raise exception 'cohort_reservation_fault'; end if; return new; end $$`)); + await db.execute(sql`create trigger cohort_reserve_fault before insert on workspace_operations for each row execute function cohort_reserve_fault()`); + await expect(run(f)).rejects.toThrow(); + const [operation] = await db.select().from(workspaceOperations); + expect(operation.status).toBe("running"); + const membership = operation.metadata!.cohortMigration; + expect(await db.select().from(workspaceOperations)).toHaveLength(1); + await assertLocal(f); + const noStorage = new Proxy(storage, { get() { throw new Error("readonly_preflight_storage_forbidden"); } }); + await db.transaction(async tx => { + await tx.execute(sql`set local statement_timeout='8s'`); + await tx.execute(sql`set local lock_timeout='1s'`); + expect((await tx.execute(sql`select current_setting('transaction_read_only') as ro`))[0].ro).toBe("on"); + await expect(run(f, { db: tx as unknown as Db, storage: noStorage, apply: false })) + .resolves.toMatchObject({ applied: false, eligible: true, cohortId: (membership as { id: string }).id }); + }, { accessMode: "read only" }); + expect(await db.select().from(workspaceOperations)).toHaveLength(1); + expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0); + await expect(run(f, { workspaceIds: [operation.executionWorkspaceId!] })).rejects.toThrow("workspace_cohort_membership_changed"); + await expect(migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: operation.executionWorkspaceId!, + apply: true, signal: AbortSignal.timeout(50_000) })).rejects.toThrow("remote_workspace_cohort_resume_required"); + await db.execute(sql`drop trigger cohort_reserve_fault on workspace_operations`); + await expect(run(f)).resolves.toMatchObject({ applied: true, cohortId: (membership as { id: string }).id }); + const operations = await db.select().from(workspaceOperations); + expect(operations).toHaveLength(2); + expect(operations.every(row => row.status === "succeeded" && JSON.stringify(row.metadata!.cohortMigration) === JSON.stringify(membership))).toBe(true); + }); + it.each(["first", "last"] as const)("does not persist a stale cohort source proof when the %s member changes before reservation", async position => { + const f = await fixture(), [first, second] = [...f.leaves].sort((a, b) => a.id.localeCompare(b.id)); + const changedMember = position === "first" ? first : second; + await db.execute(sql.raw(`create function cohort_reserve_fault() returns trigger language plpgsql as $$ begin if new.execution_workspace_id='${second.id}'::uuid then raise exception 'cohort_reservation_fault'; end if; return new; end $$`)); + await db.execute(sql`create trigger cohort_reserve_fault before insert on workspace_operations for each row execute function cohort_reserve_fault()`); + const inspect = onlineWorkspaceSource.inspectOnlineWorkspaceSource; let changed = false; + vi.spyOn(onlineWorkspaceSource, "inspectOnlineWorkspaceSource").mockImplementation(async (...args) => { + if (!changed) { + changed = true; + await db.update(executionWorkspaces).set({ metadata: { ...changedMember.metadata, preservedEvidence: "new source metadata" } }) + .where(eq(executionWorkspaces.id, changedMember.id)); + } + return inspect(...args); + }); + await expect.soft(run(f)).rejects.toThrow("workspace_cohort_source_changed"); + expect.soft(await db.select().from(workspaceOperations)).toHaveLength(0); + expect.soft(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0); + await assertLocal(f); + vi.restoreAllMocks(); + await db.execute(sql`drop trigger cohort_reserve_fault on workspace_operations`); + await expect(run(f)).resolves.toMatchObject({ applied: true }); + }); + it("refuses changed source fingerprints after partial reservation before storage or more journal work", async () => { + const f = await fixture(), second = [...f.leaves].sort((a, b) => a.id.localeCompare(b.id))[1]; + await db.execute(sql.raw(`create function cohort_reserve_fault() returns trigger language plpgsql as $$ begin if new.execution_workspace_id='${second.id}'::uuid then raise exception 'cohort_reservation_fault'; end if; return new; end $$`)); + await db.execute(sql`create trigger cohort_reserve_fault before insert on workspace_operations for each row execute function cohort_reserve_fault()`); + await expect(run(f)).rejects.toThrow(); + await db.execute(sql`drop trigger cohort_reserve_fault on workspace_operations`); + const operations = await db.select().from(workspaceOperations); + expect(operations).toHaveLength(1); + await db.update(executionWorkspaces).set({ metadata: { ...second.metadata, preservedEvidence: "changed after reservation" } }) + .where(eq(executionWorkspaces.id, second.id)); + const calls: string[] = [], noStorage = new Proxy(storage, { get(_target, key) { + calls.push(String(key)); throw new Error("unexpected_storage_operation"); + } }); + await expect(run(f, { storage: noStorage, apply: false })).rejects.toThrow("workspace_cohort_membership_changed"); + await expect(run(f, { storage: noStorage })).rejects.toThrow("workspace_cohort_membership_changed"); + expect(calls).toEqual([]); + expect(await db.select().from(workspaceOperations)).toEqual(operations); + expect(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0); + await assertLocal(f); + }); + it("refuses mixed local and independently imported members before storage or journal work", async () => { + const f = await fixture(), [first, second] = f.leaves; + await migrateRemoteWorkspace({ db, storage, companyId: f.companyId, workspaceId: first.id, apply: true, + signal: AbortSignal.timeout(50_000) }); + const operations = await db.select().from(workspaceOperations), snapshots = await db.select().from(executionWorkspaceSnapshots); + const workspaces = await rows(f); + expect(readRemoteWorkspaceAuthority(workspaces.find(row => row.id === first.id)!.metadata)).not.toBeNull(); + const calls: string[] = [], noStorage = new Proxy(storage, { get(_target, key) { + calls.push(String(key)); throw new Error("unexpected_storage_operation"); + } }); + await expect(run(f, { storage: noStorage, apply: false })).rejects.toThrow("workspace_cohort_membership_changed"); + await expect(run(f, { storage: noStorage })).rejects.toThrow("workspace_cohort_membership_changed"); + expect(calls).toEqual([]); + expect(await db.select().from(workspaceOperations)).toEqual(operations); + expect(await db.select().from(executionWorkspaceSnapshots)).toEqual(snapshots); + expect(await rows(f)).toEqual(workspaces); + expect(await fs.readFile(path.join(second.cwd!, "saved.txt"), "utf8")).toBe(`${second.name} dirty tracked work\n`); + }); + it.each(["cancelled", "failed", "running-finished"] as const)("refuses a %s member journal before preflight or snapshot work", async status => { + const f = await fixture(), second = [...f.leaves].sort((a, b) => a.id.localeCompare(b.id))[1]; + await db.execute(sql.raw(`create function cohort_reserve_fault() returns trigger language plpgsql as $$ begin if new.execution_workspace_id='${second.id}'::uuid then raise exception 'cohort_reservation_fault'; end if; return new; end $$`)); + await db.execute(sql`create trigger cohort_reserve_fault before insert on workspace_operations for each row execute function cohort_reserve_fault()`); + await expect(run(f)).rejects.toThrow(); + await db.execute(sql`drop trigger cohort_reserve_fault on workspace_operations`); + const [operation] = await db.select().from(workspaceOperations); + expect(operation.status).toBe("running"); + await db.update(workspaceOperations).set({ status: status === "running-finished" ? "running" : status, finishedAt: new Date() }) + .where(eq(workspaceOperations.id, operation.id)); + const calls: string[] = [], noStorage = new Proxy(storage, { get(_target, key) { + calls.push(String(key)); throw new Error("unexpected_storage_operation"); + } }); + await db.transaction(async tx => { + expect((await tx.execute(sql`select current_setting('transaction_read_only') as ro`))[0].ro).toBe("on"); + await expect.soft(run(f, { db: tx as unknown as Db, storage: noStorage, apply: false })) + .rejects.toThrow("workspace_cohort_journal_changed"); + }, { accessMode: "read only" }); + await expect.soft(run(f, { storage: noStorage })).rejects.toThrow("workspace_cohort_journal_changed"); + expect.soft(calls).toEqual([]); + expect.soft(await db.select().from(executionWorkspaceSnapshots)).toHaveLength(0); + expect.soft(await db.select().from(workspaceOperations)).toHaveLength(1); + await assertLocal(f); + }); + it.each(["failure", "cancel"] as const)("joins active capture peers and keeps the fence on %s", async mode => { + const f = await fixture(), controller = new AbortController(); let entered = 0, firstFailed!: () => void, release!: () => void; + const both = new Promise(resolve => { firstFailed = resolve; }), peer = new Promise(resolve => { release = resolve; }); + vi.spyOn(storage, "putFile").mockImplementation(async () => { + const index = ++entered; + if (index === 2) { if (mode === "cancel") controller.abort(new Error("caller_stop")); firstFailed(); } + if (index === 1) { await both; if (mode === "failure") throw new Error("capture_upload_failure"); } + await peer; throw new Error("peer_finished_after_abort"); + }); + let settled = false; + const pending = run(f, { signal: controller.signal }).then(value => ({ value }), error => ({ error })).finally(() => { settled = true; }); + await both; + await new Promise(resolve => setTimeout(resolve, 25)); + expect(settled).toBe(false); + const other = createDb(database.connectionString, { maxConnections: 1 }); + try { await expect(withWorkspaceSourceLocks(other, { companyId: f.companyId, projectIds: [f.projectId] }, async () => true, true)) + .rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); } + finally { release(); await other.$client.end({ timeout: 1 }); } + expect(await pending).toHaveProperty("error"); + await assertLocal(f); + expect(await db.select().from(workspaceOperations)).toHaveLength(2); + expect(await db.select().from(activityLog)).toHaveLength(0); + }); + it("does not publish after its actual fence connection is lost", async () => { + const f = await fixture(), put = storage.putFile.bind(storage); let killed = false; + vi.spyOn(storage, "putFile").mockImplementation(async input => { + const result = await put(input); + if (!killed) { + killed = true; const key = `managed_project_source:${f.companyId}:${f.projectId}`; + const held = await db.execute(sql`select distinct pid from pg_locks where locktype='advisory' and granted + and classid=((hashtextextended(${key},0)>>32)&4294967295)::oid and objid=(hashtextextended(${key},0)&4294967295)::oid`); + expect(held).toHaveLength(1); await db.execute(sql`select pg_terminate_backend(${Number(held[0].pid)})`); + } + return result; + }); + await expect(run(f)).rejects.toThrow(); + await assertLocal(f); + expect(await db.select().from(activityLog)).toHaveLength(0); + vi.restoreAllMocks(); + await expect(run(f)).resolves.toMatchObject({ applied: true }); + expect((await rows(f)).every(row => readRemoteWorkspaceAuthority(row.metadata))).toBe(true); + }); + it("refuses unreviewed membership, a parent/child pair, and a foreign journal without captures", async () => { + const f = await fixture(), put = vi.spyOn(storage, "putFile"); + await expect(run(f, { workspaceIds: Array(9).fill(f.leaves[0].id) })).rejects.toThrow(); + await expect(run(f, { concurrency: 9 })).rejects.toThrow(); + await expect(run(f, { workspaceIds: [f.leaves[0].id, f.leaves[0].id.toUpperCase()] })).rejects.toThrow("duplicate_member"); + await expect(run(f, { workspaceIds: [f.parent.id, f.leaves[0].id] })).rejects.toThrow("leaf_members_required"); + await db.insert(workspaceOperations).values({ companyId: f.companyId, executionWorkspaceId: f.parent.id, + issueId: f.parent.sourceIssueId, cwd: f.root, phase: "workspace_finalize", status: "running", metadata: { uncertain: true } }); + await expect(run(f)).rejects.toThrow("workspace_operation_unresolved"); + expect(put).not.toHaveBeenCalled(); + expect(await db.select().from(workspaceOperations)).toHaveLength(1); + await assertLocal(f); + }); + + it("keeps every authority and activity unpublished when the actual outer commit fails", async () => { + const f = await fixture(), events: unknown[] = [], unsubscribe = subscribeCompanyLiveEvents(f.companyId, event => { + if (event.type === "activity.logged") events.push(event); + }); + await db.execute(sql`create function cohort_commit_fault() returns trigger language plpgsql as $$ begin raise exception 'outer_commit_fault'; end $$`); + await db.execute(sql`create constraint trigger cohort_commit_fault after update on workspace_operations deferrable initially deferred + for each row when (new.status='succeeded') execute function cohort_commit_fault()`); + try { + await expect(run(f)).rejects.toThrow(); + expect(events).toEqual([]); + await assertLocal(f); + expect(await db.select().from(activityLog)).toHaveLength(0); + const journals = await db.select().from(workspaceOperations); + expect(journals.every(row => row.status === "running")).toBe(true); + expect((await db.select().from(executionWorkspaceSnapshots)).every(row => row.status === "uploaded")).toBe(true); + await db.execute(sql`drop trigger cohort_commit_fault on workspace_operations`); + await expect(run(f)).resolves.toMatchObject({ applied: true }); + expect(events).toHaveLength(2); + expect((await db.select().from(workspaceOperations)).map(row => row.id).sort()).toEqual(journals.map(row => row.id).sort()); + } finally { unsubscribe(); } + }); + it("does not return a savepoint success inside somebody else's source transaction", async () => { + const f = await fixture(), put = vi.spyOn(storage, "putFile"); + await withWorkspaceSourceLocks(db, { companyId: f.companyId, projectIds: [f.projectId] }, async () => { + await expect(run(f)).rejects.toBeInstanceOf(IsolatedRepositorySourceBusy); + expect(await db.select().from(workspaceOperations)).toHaveLength(0); + }); + expect(put).not.toHaveBeenCalled(); await assertLocal(f); + }); + +}); diff --git a/server/src/services/remote-workspace-migration-tree.ts b/server/src/services/remote-workspace-migration-tree.ts index 0695ffe5a31..76521ab43aa 100644 --- a/server/src/services/remote-workspace-migration-tree.ts +++ b/server/src/services/remote-workspace-migration-tree.ts @@ -62,6 +62,12 @@ export async function offlineWorkspaceMigrationBlockers(db: ReadDb) { return active ? ["active_execution_requires_drain"] : []; } +export type WorkspaceMigrationCohort = { + schema: "paperclip.remote-workspace-cohort.v1"; + id: string; companyId: string; sourceProof: string; + members: Array<{ workspaceId: string; operationId: string }>; +}; + export type WorkspaceSourceScope = { root: string; projectId: string | null; issueIds: string[]; workspaceIds: string[]; device: number; inode: number; physicalRoots?: string[]; gitSourceRoots?: string[]; settledPreparedWorkspaceIds?: string[]; settledRecoveryIssueIds?: string[]; settledFinalizationRunIds?: string[]; @@ -153,12 +159,16 @@ export async function inspectOnlineOrdinarySource(db: ReadDb, workspace: Workspa } export async function onlineOrdinarySourceBlockers(db: Pick, companyId: string, - source: WorkspaceSourceScope, workspace: Pick, operationId?: string): Promise { + source: WorkspaceSourceScope, workspace: Pick, operationId?: string, cohort?: WorkspaceMigrationCohort): Promise { const workspaceIds = sql.join(source.workspaceIds.map(id => sql`${id}::uuid`), sql`, `); const issueIds = sql.join(source.issueIds.map(id => sql`${id}::uuid`), sql`, `); const settledWorkspaces = sql.join((source.settledPreparedWorkspaceIds?.length ? source.settledPreparedWorkspaceIds : ["00000000-0000-0000-0000-000000000000"]).map(id => sql`${id}`), sql`, `); const settledFinalizers = sql.join((source.settledFinalizationRunIds?.length ? source.settledFinalizationRunIds : ["00000000-0000-0000-0000-000000000000"]).map(id => sql`${id}::uuid`), sql`, `); const settledRecoveries = sql.join((source.settledRecoveryIssueIds?.length ? source.settledRecoveryIssueIds : ["00000000-0000-0000-0000-000000000000"]).map(id => sql`${id}::uuid`), sql`, `); + const admitted = cohort ? sql`o.phase='remote_workspace_migration' and o.metadata->'cohortMigration'=${JSON.stringify(cohort)}::jsonb + and exists (select 1 from jsonb_to_recordset(${JSON.stringify(cohort.members)}::jsonb) as m("workspaceId" uuid,"operationId" uuid) + where m."workspaceId"=o.execution_workspace_id and m."operationId"=o.id)` : sql`false`; + const otherOperation = cohort ? sql`true` : sql`(${operationId ?? null}::uuid is null or o.id<>${operationId ?? null}::uuid)`; const rows = await db.execute<{ reason: string }>(sql` select 'active_execution' reason where exists (select 1 from heartbeat_runs r where r.company_id=${companyId}::uuid and r.status in ('queued','running') and (r.issue_id in (${issueIds}) or r.native_issue_id in (${issueIds}) @@ -167,8 +177,8 @@ export async function onlineOrdinarySourceBlockers(db: Pick, comp union all select 'task_writer_bound' where exists (select 1 from issues i where i.company_id=${companyId}::uuid and i.id in (${issueIds}) and (i.execution_run_id is not null or i.checkout_run_id is not null)) union all select 'workspace_operation_unresolved' where exists (select 1 from workspace_operations o where o.company_id=${companyId}::uuid - and o.execution_workspace_id in (${workspaceIds}) and (${operationId ?? null}::uuid is null or o.id<>${operationId ?? null}::uuid) - and (o.status in ('running','pending') or o.metadata->>'uncertain'='true')) + and o.execution_workspace_id in (${workspaceIds}) and ${otherOperation} + and not coalesce((${admitted}),false) and (o.status in ('running','pending') or o.metadata->>'uncertain'='true')) union all select 'workspace_runtime_owned' where exists (select 1 from execution_workspace_runtime_leases l where l.company_id=${companyId}::uuid and l.execution_workspace_id in (${workspaceIds}) and l.expires_at>clock_timestamp()) union all select 'workspace_runtime_service_unsettled' where exists (select 1 from workspace_runtime_services s where s.company_id=${companyId}::uuid diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 1633032a835..4ce163e8c71 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -1,4 +1,5 @@ -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; +import { z } from "zod"; import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; @@ -22,10 +23,10 @@ import { loadTaskDirectoryRestorePoint, mayBeIsolatedTaskDirectoryWorkspace, rea import { inspectArchivedGitMigrationSource, lockArchivedGitMigrationSources, materializeArchivedGitMigrationSource, type ArchivedGitMigrationSource } from "./remote-workspace-migration-git.js"; import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { isLegacyWorkspaceTaskRelated, readLegacyWorkspaceRecoveryAllocations, LEGACY_WORKSPACE_RECOVERY_MIGRATION, type LegacyWorkspaceRecoverySource } from "./legacy-workspace-recovery-migration.js"; -import { createNestedWorkspaceImportVerifier, onlineOrdinarySourceBlockers, offlineWorkspaceMigrationBlockers, prepareNestedWorkspaceImports, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js"; +import { createNestedWorkspaceImportVerifier, onlineOrdinarySourceBlockers, offlineWorkspaceMigrationBlockers, prepareNestedWorkspaceImports, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers, type WorkspaceMigrationCohort } from "./remote-workspace-migration-tree.js"; import { inspectOnlineWorkspaceSource, requiresOnlineWorkspaceSource, type OnlineWorkspaceSource } from "./online-workspace-source.js"; -import { withWorkspaceSourceLocks, IsolatedRepositorySourceBusy } from "./workspace-source-lock.js"; +import { withWorkspaceSourceLocks, withIndependentWorkspaceSourceLocks, IsolatedRepositorySourceBusy } from "./workspace-source-lock.js"; import { inspectOnlineLegacyWorkspaceRecoverySource, inspectOnlineLegacyWorkspaceRecoveryHold, lockOnlineLegacyWorkspaceRecoverySources, type OnlineLegacyWorkspaceRecoveryHold } from "./online-legacy-workspace-recovery.js"; import { inspectNativeWorkspaceMigration, nativeWorkspaceMigrationMarker } from "./native-runtime/native-workspace-migration.js"; @@ -86,7 +87,7 @@ async function isUnmaterializedRepositoryRequest(workspace: Workspace, }); } -async function inspect(db: ReadDb, input: { companyId: string; workspaceId: string; operationId?: string; writersStopped?: boolean; onlineSource?: OnlineWorkspaceSource; onlineSourceUnproven?: boolean }) { +async function inspect(db: ReadDb, input: { companyId: string; workspaceId: string; operationId?: string; writersStopped?: boolean; onlineSource?: OnlineWorkspaceSource; onlineSourceUnproven?: boolean }, cohort?: WorkspaceMigrationCohort) { const [workspace] = await db.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))); if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); const authority = readRemoteWorkspaceAuthority(workspace.metadata); @@ -99,7 +100,7 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri if (input.onlineSource) { const currentScope = await inspectOnlineWorkspaceSource(db, workspace); if (canonical(currentScope) !== canonical(input.onlineSource)) reasons.push("online_workspace_source_scope_changed"); - else reasons.push(...await onlineOrdinarySourceBlockers(db as Db, workspace.companyId, currentScope!, workspace, allowedOperation)); + else reasons.push(...await onlineOrdinarySourceBlockers(db as Db, workspace.companyId, currentScope!, workspace, allowedOperation, cohort)); } if (!["local_fs", "git_worktree"].includes(workspace.providerType)) reasons.push("workspace_provider_unsupported"); if (!["active", "idle", "in_review", "archived"].includes(workspace.status)) reasons.push("workspace_not_active"); @@ -322,12 +323,13 @@ export async function inventoryRemoteWorkspaceMigration(db: Db, input: { company snapshotId: result.authority?.snapshotId ?? null }; } -/** Administrative import. The CLI enforces external deployment prerequisites. */ -export async function migrateRemoteWorkspace(input: { +type MigrationInput = { db: Db; storage: StorageService; companyId: string; workspaceId: string; apply?: boolean; source?: RemoteWorkspaceSource; writersStopped?: boolean; signal?: AbortSignal; -}) { - if (!input.apply) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: false }; +}; +type MigrationTransaction = Parameters[0]>[0]; + +async function resolveMigrationSource(input: MigrationInput) { const [sourceWorkspace] = !input.writersStopped ? await input.db.select().from(executionWorkspaces).where(and( eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))) : []; const onlineSource = sourceWorkspace && !readRemoteWorkspaceAuthority(sourceWorkspace.metadata) @@ -336,243 +338,306 @@ export async function migrateRemoteWorkspace(input: { && await requiresOnlineWorkspaceSource(sourceWorkspace)) { throw new Error("remote_workspace_migration_blocked:online_workspace_source_scope_unproven"); } - let committedPublication: Parameters[0] | undefined; - const migrate = async (lock: Parameters[2]>[0]) => { - const reserved = await input.db.transaction(async tx => { - const [selected] = await tx.select({ sourceIssueId: executionWorkspaces.sourceIssueId }).from(executionWorkspaces) - .where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))); - if (!selected) throw new Error("remote_workspace_migration_workspace_missing"); - await lockMigrationParents(tx, input.companyId, selected.sourceIssueId); - const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))).for("update"); - if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); - if (workspace.sourceIssueId !== selected.sourceIssueId) throw new Error("remote_workspace_migration_source_changed"); - const current = await inspect(tx as unknown as Db, { ...input, onlineSource }); - if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[], onlineRecoveryHolds: [] as OnlineRecoveryHold[] }; - if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); - const nativeSessionContinuity = await inspectNativeWorkspaceMigration(tx as unknown as Db, workspace); - const source = remoteWorkspaceSourceSchema.parse(input.source ?? current.source); - const sourceFingerprint = fingerprint(workspace); - const marker = current.marker; - if (marker.schema === SCHEMA && typeof marker.operationId === "string") { - if (marker.sourceFingerprint !== sourceFingerprint || canonical(marker.source) !== canonical(source)) throw new Error("remote_workspace_migration_source_changed"); - const [operation] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, workspace.companyId), - eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"))); - if (!operation) throw new Error("remote_workspace_migration_operation_missing"); - if (canonical(operation.metadata?.nativeSessionContinuity ?? null) !== canonical(nativeSessionContinuity)) throw new Error("native_workspace_migration_source_changed"); - // Older operations only admitted the task's current workspace. Preserve - // that binding when resuming an import journaled before this field existed. - const issueBinding = Object.hasOwn(operation.metadata ?? {}, "sourceIssueBinding") - ? operation.metadata!.sourceIssueBinding : workspace.id; - if (issueBinding !== current.issueBinding) throw new Error("remote_workspace_migration_source_changed"); - if (Boolean(operation.metadata?.retainedDirectory) !== Boolean(current.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); - if (canonical(operation.metadata?.archivedGitSource ?? null) !== canonical(gitSourceProof(current.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); - if (canonical(operation.metadata?.finalizedRecoveryHolds ?? []) !== canonical(current.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); - if (canonical(operation.metadata?.legacyRecoverySources ?? []) !== canonical(current.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed"); - if (canonical(operation.metadata?.onlineRecoveryHolds ?? []) !== canonical(current.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); - return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds }; - } - const [operation] = await tx.insert(workspaceOperations).values({ companyId: workspace.companyId, executionWorkspaceId: workspace.id, - issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running", - metadata: { schema: SCHEMA, sourceFingerprint, source, sourceIssueBinding: current.issueBinding, - ...(nativeSessionContinuity ? { nativeSessionContinuity } : {}), - nestedWorkspaceScope: { version: 1, state: "pending" }, - ...(current.finalizedRecoveryHolds.length ? { finalizedRecoveryHolds: current.finalizedRecoveryHolds } : {}), - ...(current.legacyRecoverySources.length ? { legacyRecoverySources: current.legacyRecoverySources } : {}), - ...(current.onlineRecoveryHolds.length ? { onlineRecoveryHolds: current.onlineRecoveryHolds } : {}), - ...(current.archivedGitSource ? { archivedGitSource: gitSourceProof(current.archivedGitSource) } : {}), - ...(current.retainedDirectory ? { retainedDirectory: { restorePointId: current.retainedDirectory.restorePoint.id, - generation: current.retainedDirectory.generation, digest: current.retainedDirectory.restorePoint.digest } } : {}) } }).returning(); - await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); - await tx.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspaceMigration: { - schema: SCHEMA, operationId: operation!.id, state: "running", sourceFingerprint, source, - } }, updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); - return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds }; - }); - if (reserved.authority) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: true, replayed: true }; - let operation = reserved.operation!; - if (onlineSource) { - if (!await lockOnlineLegacyWorkspaceRecoverySources(lock as unknown as Db, { companyId: input.companyId, - sources: [...reserved.legacyRecoverySources, ...onlineSource.companions.flatMap(proof => proof.sources)], - holds: [...reserved.onlineRecoveryHolds, ...onlineSource.companions.flatMap(proof => proof.holds)] })) - throw new Error("remote_workspace_migration_blocked:remote_copyback_owned"); - const current = await inspect(input.db, { ...input, onlineSource, operationId: operation.id }); - if (current.reasons.length || canonical(current.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources) - || canonical(current.onlineRecoveryHolds) !== canonical(reserved.onlineRecoveryHolds)) - throw new Error("remote_workspace_migration_recovery_source_changed"); + return onlineSource; +} + +async function reserveMigration(input: MigrationInput, onlineSource?: OnlineWorkspaceSource, cohort?: WorkspaceMigrationCohort, + expectedSourceFingerprint?: string) { + return input.db.transaction(async tx => { + const [selected] = await tx.select({ sourceIssueId: executionWorkspaces.sourceIssueId }).from(executionWorkspaces) + .where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))); + if (!selected) throw new Error("remote_workspace_migration_workspace_missing"); + await lockMigrationParents(tx, input.companyId, selected.sourceIssueId); + const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))).for("update"); + if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); + if (workspace.sourceIssueId !== selected.sourceIssueId) throw new Error("remote_workspace_migration_source_changed"); + const current = await inspect(tx as unknown as Db, { ...input, onlineSource }, cohort); + if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[], onlineRecoveryHolds: [] as OnlineRecoveryHold[] }; + if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); + const nativeSessionContinuity = await inspectNativeWorkspaceMigration(tx as unknown as Db, workspace); + const source = remoteWorkspaceSourceSchema.parse(input.source ?? current.source); + const sourceFingerprint = fingerprint(workspace); + if (cohort && sourceFingerprint !== expectedSourceFingerprint) throw new Error("workspace_cohort_source_changed"); + const marker = current.marker; + if (marker.schema === SCHEMA && typeof marker.operationId === "string") { + if (marker.sourceFingerprint !== sourceFingerprint || canonical(marker.source) !== canonical(source)) throw new Error("remote_workspace_migration_source_changed"); + const [operation] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, workspace.companyId), + eq(workspaceOperations.executionWorkspaceId, workspace.id), eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"))); + if (!operation) throw new Error("remote_workspace_migration_operation_missing"); + if (canonical(operation.metadata?.cohortMigration ?? null) !== canonical(cohort ?? null)) throw new Error("remote_workspace_cohort_resume_required"); + if (canonical(operation.metadata?.nativeSessionContinuity ?? null) !== canonical(nativeSessionContinuity)) throw new Error("native_workspace_migration_source_changed"); + // Older operations only admitted the task's current workspace. Preserve + // that binding when resuming an import journaled before this field existed. + const issueBinding = Object.hasOwn(operation.metadata ?? {}, "sourceIssueBinding") + ? operation.metadata!.sourceIssueBinding : workspace.id; + if (issueBinding !== current.issueBinding) throw new Error("remote_workspace_migration_source_changed"); + if (Boolean(operation.metadata?.retainedDirectory) !== Boolean(current.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); + if (canonical(operation.metadata?.archivedGitSource ?? null) !== canonical(gitSourceProof(current.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); + if (canonical(operation.metadata?.finalizedRecoveryHolds ?? []) !== canonical(current.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); + if (canonical(operation.metadata?.legacyRecoverySources ?? []) !== canonical(current.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed"); + if (canonical(operation.metadata?.onlineRecoveryHolds ?? []) !== canonical(current.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); + return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds }; } - const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.companyId, workspaceId: input.workspaceId, - operationId: operation.id, generation: 1, epoch: 1 }); - const scratch = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-"))); - const childVerifier = onlineSource?.kind === "git" ? createNestedWorkspaceImportVerifier({ - db: input.db, storage: input.storage, companyId: input.companyId, sourceLock: lock, - projectIds: onlineSource.projectIds, issueIds: onlineSource.issueIds, signal: input.signal, + const [operation] = await tx.insert(workspaceOperations).values({ ...(cohort ? { id: cohort.members.find(member => member.workspaceId === input.workspaceId)!.operationId } : {}), companyId: workspace.companyId, executionWorkspaceId: workspace.id, + issueId: workspace.sourceIssueId, cwd: workspace.cwd ?? workspace.providerRef, phase: "remote_workspace_migration", status: "running", + metadata: { schema: SCHEMA, sourceFingerprint, source, sourceIssueBinding: current.issueBinding, + ...(cohort ? { cohortMigration: cohort } : {}), + ...(nativeSessionContinuity ? { nativeSessionContinuity } : {}), + nestedWorkspaceScope: { version: 1, state: "pending" }, + ...(current.finalizedRecoveryHolds.length ? { finalizedRecoveryHolds: current.finalizedRecoveryHolds } : {}), + ...(current.legacyRecoverySources.length ? { legacyRecoverySources: current.legacyRecoverySources } : {}), + ...(current.onlineRecoveryHolds.length ? { onlineRecoveryHolds: current.onlineRecoveryHolds } : {}), + ...(current.archivedGitSource ? { archivedGitSource: gitSourceProof(current.archivedGitSource) } : {}), + ...(current.retainedDirectory ? { retainedDirectory: { restorePointId: current.retainedDirectory.restorePoint.id, + generation: current.retainedDirectory.generation, digest: current.retainedDirectory.restorePoint.digest } } : {}) } }).returning(); + await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); + await tx.update(executionWorkspaces).set({ metadata: { ...workspace.metadata, remoteWorkspaceMigration: { + schema: SCHEMA, operationId: operation!.id, state: "running", sourceFingerprint, source, + } }, updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); + return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds, onlineRecoveryHolds: current.onlineRecoveryHolds }; + }); +} +type ReservedMigration = Awaited>; +type MigrationPlan = { input: MigrationInput; onlineSource?: OnlineWorkspaceSource; reserved: ReservedMigration }; + +async function lockMigrationRecovery(lock: MigrationTransaction, plans: MigrationPlan[], cohort?: WorkspaceMigrationCohort) { + const online = plans.filter(plan => plan.onlineSource); + if (!online.length) return; + if (!await lockOnlineLegacyWorkspaceRecoverySources(lock as unknown as Db, { companyId: online[0].input.companyId, + sources: online.flatMap(({ reserved, onlineSource }) => [...reserved.legacyRecoverySources, ...onlineSource!.companions.flatMap(proof => proof.sources)]), + holds: online.flatMap(({ reserved, onlineSource }) => [...reserved.onlineRecoveryHolds, ...onlineSource!.companions.flatMap(proof => proof.holds)]) })) + throw new Error("remote_workspace_migration_blocked:remote_copyback_owned"); + for (const { input, onlineSource, reserved } of online) { + const current = await inspect(input.db, { ...input, onlineSource, operationId: reserved.operation!.id }, cohort); + if (current.reasons.length || canonical(current.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources) + || canonical(current.onlineRecoveryHolds) !== canonical(reserved.onlineRecoveryHolds)) + throw new Error("remote_workspace_migration_recovery_source_changed"); + } +} + +function createMigrationChildVerifier(input: MigrationInput, onlineSource: OnlineWorkspaceSource | undefined, sourceLock: MigrationTransaction) { + if (onlineSource?.kind !== "git") return undefined; + const verifier = createNestedWorkspaceImportVerifier({ db: input.db, storage: input.storage, companyId: input.companyId, + sourceLock, projectIds: onlineSource.projectIds, issueIds: onlineSource.issueIds, signal: input.signal }); + let pending = Promise.resolve(); + return { + prepare(options: Parameters[0]) { + const work = pending.then(() => verifier.prepare(options)); + pending = work.then(() => undefined, () => undefined); + return work; + }, + async close() { await pending; await verifier.close(); }, + }; +} + +async function captureMigration(input: MigrationInput, reserved: ReservedMigration, onlineSource?: OnlineWorkspaceSource, + childVerifier?: ReturnType) { + let operation = reserved.operation!; + const snapshot = await beginWorkspaceSnapshot(input.db, { companyId: input.companyId, workspaceId: input.workspaceId, + operationId: operation.id, generation: 1, epoch: 1 }); + const scratch = await fs.realpath(await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-"))); + try { + input.signal?.throwIfAborted(); + const retained = reserved.retainedDirectory?.restorePoint; + const archivePath = path.join(scratch, "workspace.pcws"); + const recovered = reserved.archivedGitSource ? await recoverWorkspaceSnapshotArchive({ + db: input.db, storage: input.storage, companyId: input.companyId, workspaceId: input.workspaceId, + snapshotId: snapshot.id, targetPath: archivePath, signal: input.signal, }) : null; - const prepareChildren = (options: Parameters[0]) => - childVerifier ? childVerifier.prepare(options) : prepareNestedWorkspaceImports(options); + const sourceRoot = recovered ? null : retained ? await materializeTaskDirectorySnapshot({ + payload: await loadTaskDirectoryRestorePoint({ storage: input.storage, workspace: reserved.workspace, ref: retained, recoverMissingEmpty: true, signal: input.signal }), + target: path.join(scratch, "retained-source"), expectedDigest: retained.digest, + }) : reserved.archivedGitSource ? await materializeArchivedGitMigrationSource({ db: input.db, storage: input.storage, + workspace: reserved.workspace, source: reserved.archivedGitSource, scratch }) : reserved.workspace.cwd ?? reserved.workspace.providerRef!; + const scope = object(operation.metadata?.nestedWorkspaceScope); + if (operation.metadata?.nestedWorkspaceScope !== undefined && (scope.version !== 1 || !["pending", "fixed"].includes(String(scope.state)))) + throw new Error("nested_workspace_scope_invalid"); + // Journals predating scoped captures included every child. Preserve that + // archive identity even if a later batch has already imported the leaves. + const recordedChildren = scope.state === "fixed" ? validateImportedWorkspaceChildren(scope.children) + : scope.state === "pending" ? undefined : []; + const nested = (input.writersStopped || onlineSource?.kind === "git") && !retained && !reserved.archivedGitSource + ? await (childVerifier?.prepare ?? prepareNestedWorkspaceImports)({ ...input, workspace: reserved.workspace, scratch, recordedChildren, gitSourceRoots: onlineSource?.gitSourceRoots }) : []; + if (scope.state === "pending") { + const [saved] = await input.db.update(workspaceOperations).set({ metadata: { ...operation.metadata, + nestedWorkspaceScope: { version: 1, state: "fixed", children: nested } }, updatedAt: new Date() }) + .where(and(eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.id, operation.id), + eq(workspaceOperations.status, "running"))).returning(); + if (!saved) throw new Error("remote_workspace_migration_operation_changed"); + operation = saved; + } + const exclude = nested.map(child => child.relativePath); + const captured = recovered ?? await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); + const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); + const verifiedArchivePath = process.platform === "win32" ? undefined : restoredArchive; + const ref = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage, snapshotId: snapshot.id, + companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal, + verifiedArchivePath, + // Private Git reconstruction can change administrative bytes between + // attempts. Its journaled logical recipe is checked at reservation and + // again under source locks before authority publication below. + ...(reserved.archivedGitSource ? { unpublishedRecapture: { sourceOperationId: operation.id } } : {}), + }); + if (!verifiedArchivePath) await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, + targetPath: restoredArchive, signal: input.signal }); + await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal }); + if (!recovered) { + const currentArchive = path.join(scratch, "current.pcws"); + const current = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath: currentArchive, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); + if (current.manifestDigest !== captured.manifestDigest) throw new Error("remote_workspace_migration_source_changed"); + } + if (nested.length) { + const verifyScratch = path.join(scratch, "verify-nested"); await fs.mkdir(verifyScratch); + const verified = await (childVerifier?.prepare ?? prepareNestedWorkspaceImports)({ ...input, workspace: reserved.workspace, + scratch: verifyScratch, recordedChildren: nested, gitSourceRoots: onlineSource?.gitSourceRoots }); + if (canonical(verified) !== canonical(nested)) throw new Error("nested_workspace_source_changed"); + } + input.signal?.throwIfAborted(); + return { operation, ref, nested, scratch }; + } catch (error) { await removeTaskDirectoryTree(scratch); throw error; } +} +type CapturedMigration = Awaited>; + +async function validateMigrationPublication(tx: MigrationTransaction, input: MigrationInput, reserved: ReservedMigration, + captured: CapturedMigration, onlineSource?: OnlineWorkspaceSource, cohort?: WorkspaceMigrationCohort) { + const { operation, ref, nested } = captured; + if (cohort) { + const [journal] = await tx.select().from(workspaceOperations).where(and(eq(workspaceOperations.companyId, input.companyId), + eq(workspaceOperations.id, operation.id), eq(workspaceOperations.executionWorkspaceId, input.workspaceId))).for("update", { noWait: true }); + if (!journal || journal.status !== "running" || journal.finishedAt !== null || journal.phase !== "remote_workspace_migration" + || journal.issueId !== operation.issueId || journal.cwd !== operation.cwd + || canonical(journal.metadata) !== canonical(operation.metadata)) throw new Error("workspace_cohort_journal_changed"); + } + await lockMigrationParents(tx, input.companyId, reserved.workspace.sourceIssueId, onlineSource ? [ + ...reserved.legacyRecoverySources.map(source => source.issueId), + ...onlineSource.companions.flatMap(proof => proof.sources.map(source => source.issueId)), + ] : []); + for (const actionId of [...new Set([...reserved.finalizedRecoveryHolds, ...reserved.onlineRecoveryHolds].map(held => held.actionId))].sort()) { + await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, actionId))).for("update", { noWait: true }); + } + const protectedRunIds = [...new Set([...reserved.legacyRecoverySources.map(source => source.runId), + ...reserved.finalizedRecoveryHolds.map(held => held.runId)])].sort(); + for (const runId of protectedRunIds) { + await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, runId))).for("update", { noWait: true }); + await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, runId))).for("update", { noWait: true }); + } + const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))).for("update"); + if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); + const existing = readRemoteWorkspaceAuthority(workspace.metadata); + if (existing?.snapshotId === ref.id) return null; + if (existing || fingerprint(workspace) !== reserved.sourceFingerprint + || object(workspace.metadata?.remoteWorkspaceMigration).operationId !== operation.id) throw new Error("remote_workspace_migration_publication_changed"); + for (const source of reserved.legacyRecoverySources) { + for (const allocation of source.recoveryAllocations ?? []) await tx.select({ id: environmentLeases.id }) + .from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), + eq(environmentLeases.id, allocation.leaseId))).for("update", { noWait: true }); + } + const validation = await inspect(tx as unknown as Db, { ...input, operationId: operation.id, onlineSource }, cohort); + if (!input.source && canonical(validation.source) !== canonical(reserved.source)) throw new Error("remote_workspace_migration_source_changed"); + if (validation.issueBinding !== reserved.issueBinding) throw new Error("remote_workspace_migration_source_changed"); + if (Boolean(validation.retainedDirectory) !== Boolean(reserved.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); + if (canonical(gitSourceProof(validation.archivedGitSource)) !== canonical(gitSourceProof(reserved.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); + if (validation.reasons.length) throw new Error(`remote_workspace_migration_blocked:${validation.reasons.join(",")}`); + if (reserved.archivedGitSource) await lockArchivedGitMigrationSources(tx as unknown as Db, workspace, reserved.archivedGitSource); + const [uploaded] = await tx.select().from(executionWorkspaceSnapshots).where(and(eq(executionWorkspaceSnapshots.companyId, workspace.companyId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.id, ref.id))).for("update"); + if (!uploaded || uploaded.status !== "uploaded" || uploaded.sourceOperationId !== operation.id) throw new Error("remote_workspace_migration_snapshot_changed"); + for (const child of [...nested].sort((a, b) => a.snapshotId.localeCompare(b.snapshotId))) { + const [saved] = await tx.select().from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.id, child.snapshotId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, child.workspaceId), + )).for("update"); + if (!saved || !["committed", "superseded"].includes(saved.status) || saved.manifestDigest !== child.manifestDigest + || saved.sourceOperationId !== child.operationId) throw new Error("nested_workspace_snapshot_changed"); + } + if (canonical(validation.finalizedRecoveryHolds) !== canonical(reserved.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); + if (canonical(validation.onlineRecoveryHolds) !== canonical(reserved.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_recovery_source_changed"); + if (canonical(validation.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed"); + const continuity = await inspectNativeWorkspaceMigration(tx as unknown as Db, workspace); + if (canonical(continuity) !== canonical(operation.metadata?.nativeSessionContinuity ?? null)) throw new Error("native_workspace_migration_source_changed"); + return { workspace, uploaded, continuity }; +} +type ValidatedPublication = NonNullable>>; + +async function publishMigration(tx: MigrationTransaction, reserved: ReservedMigration, captured: CapturedMigration, validated: ValidatedPublication) { + const { operation, ref, nested } = captured, { workspace, uploaded, continuity } = validated; + const authority: RemoteWorkspaceAuthority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: ref.id, + source: reserved.source, location: null, writerRunId: null, dirty: false }; + const metadata: Record = { ...workspace.metadata, remoteWorkspace: authority }; delete metadata.remoteWorkspaceMigration; + if (continuity) metadata.nativeWorkspaceMigration = nativeWorkspaceMigrationMarker(operation.id); + if (reserved.legacyRecoverySources.length) { + metadata.legacyWorkspaceRecoveryMigration = { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: operation.id }; + for (const source of reserved.legacyRecoverySources) { + const markerKey = source.reference?.state === "finalized" + ? "finalizedWorkspaceRecoveryMigration" : "legacyWorkspaceRecoveryMigration"; + // The offline deployment fence has stopped the old controller. The + // locked validation above must still match its journaled expired + // lease exactly. Release only that coordinator lease so explicit + // file recovery can proceed; its phase/checkpoint and the provider's + // independent stop obligation remain unchanged. + if (source.unresolvedFinalization?.leaseOwner) await tx.update(nativeRunFinalizations).set({ + leaseOwner: null, leaseExpiresAt: null, + recoveryHistory: sql`${nativeRunFinalizations.recoveryHistory} || ${JSON.stringify([{ + event: "legacy_workspace_migration_owner_released", operationId: operation.id, + workspaceId: workspace.id, previousOwner: source.unresolvedFinalization, + }])}::jsonb`, updatedAt: new Date(), + }).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId), + eq(nativeRunFinalizations.runId, source.runId), eq(nativeRunFinalizations.issueId, source.issueId))); + await tx.update(heartbeatRuns).set({ + runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ + [markerKey]: { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, + operationId: operation.id, workspaceId: workspace.id }, + })}::jsonb`, updatedAt: new Date(), + }).where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, source.runId))); + } + const prior = Array.isArray(metadata.remoteWorkspaceRecoverySourceRunIds) ? metadata.remoteWorkspaceRecoverySourceRunIds : []; + metadata.remoteWorkspaceRecoverySourceRunIds = [...new Set([...prior, ...reserved.legacyRecoverySources.map(source => source.runId)])]; + } + await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); + await tx.update(executionWorkspaceSnapshots).set({ status: "committed", deleteAfter: null, + metadata: { ...uploaded.metadata, ...(nested.length ? { importedWorkspaceChildren: nested } : {}) }, + updatedAt: new Date() }).where(eq(executionWorkspaceSnapshots.id, ref.id)); + await tx.update(executionWorkspaces).set({ providerType: "cloud_sandbox", providerRef: null, cwd: `/workspaces/${workspace.id}`, metadata, + updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); + await tx.update(workspaceOperations).set({ status: "succeeded", finishedAt: new Date(), updatedAt: new Date(), + metadata: { ...operation.metadata, snapshotId: ref.id, manifestDigest: ref.manifestDigest } }).where(eq(workspaceOperations.id, operation.id)); + return (await persistActivity(tx as unknown as Db, { companyId: workspace.companyId, actorType: "system", actorId: "workspace-migration", + action: "workspace.remote_storage_migrated", entityType: "execution_workspace", entityId: workspace.id, + details: { snapshotId: ref.id, operationId: operation.id, manifestDigest: ref.manifestDigest } })).publication; +} + +function publicationError(error: unknown): never { + const failure = object(error), cause = object(failure.cause); + if ((cause.code ?? failure.code) !== "55P03") throw error; + throw new Error("remote_workspace_migration_blocked:execution_rows_busy", { cause: error }); +} + +/** Administrative import. The CLI enforces external deployment prerequisites. */ +export async function migrateRemoteWorkspace(input: MigrationInput) { + if (!input.apply) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: false }; + const onlineSource = await resolveMigrationSource(input); + let committedPublication: Parameters[0] | undefined; + const migrate = async (lock: MigrationTransaction) => { + const reserved = await reserveMigration(input, onlineSource); + if (reserved.authority) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: true, replayed: true }; + await lockMigrationRecovery(lock, [{ input, onlineSource, reserved }]); + const childVerifier = createMigrationChildVerifier(input, onlineSource, lock); + let scratch: string | undefined; try { - input.signal?.throwIfAborted(); - const retained = reserved.retainedDirectory?.restorePoint; - const archivePath = path.join(scratch, "workspace.pcws"); - const recovered = reserved.archivedGitSource ? await recoverWorkspaceSnapshotArchive({ - db: input.db, storage: input.storage, companyId: input.companyId, workspaceId: input.workspaceId, - snapshotId: snapshot.id, targetPath: archivePath, signal: input.signal, - }) : null; - const sourceRoot = recovered ? null : retained ? await materializeTaskDirectorySnapshot({ - payload: await loadTaskDirectoryRestorePoint({ storage: input.storage, workspace: reserved.workspace, ref: retained, recoverMissingEmpty: true, signal: input.signal }), - target: path.join(scratch, "retained-source"), expectedDigest: retained.digest, - }) : reserved.archivedGitSource ? await materializeArchivedGitMigrationSource({ db: input.db, storage: input.storage, - workspace: reserved.workspace, source: reserved.archivedGitSource, scratch }) : reserved.workspace.cwd ?? reserved.workspace.providerRef!; - const scope = object(operation.metadata?.nestedWorkspaceScope); - if (operation.metadata?.nestedWorkspaceScope !== undefined && (scope.version !== 1 || !["pending", "fixed"].includes(String(scope.state)))) - throw new Error("nested_workspace_scope_invalid"); - // Journals predating scoped captures included every child. Preserve that - // archive identity even if a later batch has already imported the leaves. - const recordedChildren = scope.state === "fixed" ? validateImportedWorkspaceChildren(scope.children) - : scope.state === "pending" ? undefined : []; - const nested = (input.writersStopped || onlineSource?.kind === "git") && !retained && !reserved.archivedGitSource - ? await prepareChildren({ ...input, workspace: reserved.workspace, scratch, recordedChildren, gitSourceRoots: onlineSource?.gitSourceRoots }) : []; - if (scope.state === "pending") { - const [saved] = await input.db.update(workspaceOperations).set({ metadata: { ...operation.metadata, - nestedWorkspaceScope: { version: 1, state: "fixed", children: nested } }, updatedAt: new Date() }) - .where(and(eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.id, operation.id), - eq(workspaceOperations.status, "running"))).returning(); - if (!saved) throw new Error("remote_workspace_migration_operation_changed"); - operation = saved; - } - const exclude = nested.map(child => child.relativePath); - const captured = recovered ?? await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); - const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); - const verifiedArchivePath = process.platform === "win32" ? undefined : restoredArchive; - const ref = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage, snapshotId: snapshot.id, - companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal, - verifiedArchivePath, - // Private Git reconstruction can change administrative bytes between - // attempts. Its journaled logical recipe is checked at reservation and - // again under source locks before authority publication below. - ...(reserved.archivedGitSource ? { unpublishedRecapture: { sourceOperationId: operation.id } } : {}), - }); - if (!verifiedArchivePath) await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, - targetPath: restoredArchive, signal: input.signal }); - await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal }); - if (!recovered) { - const currentArchive = path.join(scratch, "current.pcws"); - const current = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot!, archivePath: currentArchive, exclude, requireNoGit: onlineSource?.kind === "ordinary", gitSourceRoots: onlineSource?.gitSourceRoots, signal: input.signal }); - if (current.manifestDigest !== captured.manifestDigest) throw new Error("remote_workspace_migration_source_changed"); - } - if (nested.length) { - const verifyScratch = path.join(scratch, "verify-nested"); await fs.mkdir(verifyScratch); - const verified = await prepareChildren({ ...input, workspace: reserved.workspace, scratch: verifyScratch, recordedChildren: nested, - gitSourceRoots: onlineSource?.gitSourceRoots }); - if (canonical(verified) !== canonical(nested)) throw new Error("nested_workspace_source_changed"); - } - input.signal?.throwIfAborted(); + const captured = await captureMigration(input, reserved, onlineSource, childVerifier); + scratch = captured.scratch; await lock.execute(sql`select 1`); const publicationDb = onlineSource ? lock as unknown as Db : input.db; const publication = await publicationDb.transaction(async tx => { - await lockMigrationParents(tx, input.companyId, reserved.workspace.sourceIssueId, onlineSource ? [ - ...reserved.legacyRecoverySources.map(source => source.issueId), - ...onlineSource.companions.flatMap(proof => proof.sources.map(source => source.issueId)), - ] : []); - for (const actionId of [...new Set([...reserved.finalizedRecoveryHolds, ...reserved.onlineRecoveryHolds].map(held => held.actionId))].sort()) { - await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, actionId))).for("update", { noWait: true }); - } - const protectedRunIds = [...new Set([...reserved.legacyRecoverySources.map(source => source.runId), - ...reserved.finalizedRecoveryHolds.map(held => held.runId)])].sort(); - for (const runId of protectedRunIds) { - await tx.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations) - .where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, runId))).for("update", { noWait: true }); - await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) - .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, runId))).for("update", { noWait: true }); - } - const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))).for("update"); - if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); - const existing = readRemoteWorkspaceAuthority(workspace.metadata); - if (existing?.snapshotId === ref.id) return null; - if (existing || fingerprint(workspace) !== reserved.sourceFingerprint - || object(workspace.metadata?.remoteWorkspaceMigration).operationId !== operation.id) throw new Error("remote_workspace_migration_publication_changed"); - for (const source of reserved.legacyRecoverySources) { - for (const allocation of source.recoveryAllocations ?? []) await tx.select({ id: environmentLeases.id }) - .from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), - eq(environmentLeases.id, allocation.leaseId))).for("update", { noWait: true }); - } - const validation = await inspect(tx as unknown as Db, { ...input, operationId: operation.id, onlineSource }); - if (!input.source && canonical(validation.source) !== canonical(reserved.source)) throw new Error("remote_workspace_migration_source_changed"); - if (validation.issueBinding !== reserved.issueBinding) throw new Error("remote_workspace_migration_source_changed"); - if (Boolean(validation.retainedDirectory) !== Boolean(reserved.retainedDirectory)) throw new Error("remote_workspace_migration_source_changed"); - if (canonical(gitSourceProof(validation.archivedGitSource)) !== canonical(gitSourceProof(reserved.archivedGitSource))) throw new Error("remote_workspace_migration_source_changed"); - if (validation.reasons.length) throw new Error(`remote_workspace_migration_blocked:${validation.reasons.join(",")}`); - if (reserved.archivedGitSource) await lockArchivedGitMigrationSources(tx as unknown as Db, workspace, reserved.archivedGitSource); - const [uploaded] = await tx.select().from(executionWorkspaceSnapshots).where(and(eq(executionWorkspaceSnapshots.companyId, workspace.companyId), - eq(executionWorkspaceSnapshots.executionWorkspaceId, workspace.id), eq(executionWorkspaceSnapshots.id, ref.id))).for("update"); - if (!uploaded || uploaded.status !== "uploaded" || uploaded.sourceOperationId !== operation.id) throw new Error("remote_workspace_migration_snapshot_changed"); - for (const child of [...nested].sort((a, b) => a.snapshotId.localeCompare(b.snapshotId))) { - const [saved] = await tx.select().from(executionWorkspaceSnapshots).where(and( - eq(executionWorkspaceSnapshots.companyId, workspace.companyId), eq(executionWorkspaceSnapshots.id, child.snapshotId), - eq(executionWorkspaceSnapshots.executionWorkspaceId, child.workspaceId), - )).for("update"); - if (!saved || !["committed", "superseded"].includes(saved.status) || saved.manifestDigest !== child.manifestDigest - || saved.sourceOperationId !== child.operationId) throw new Error("nested_workspace_snapshot_changed"); - } - if (canonical(validation.finalizedRecoveryHolds) !== canonical(reserved.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_source_changed"); - if (canonical(validation.onlineRecoveryHolds) !== canonical(reserved.onlineRecoveryHolds)) throw new Error("remote_workspace_migration_recovery_source_changed"); - if (canonical(validation.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources)) throw new Error("remote_workspace_migration_source_changed"); - const continuity = await inspectNativeWorkspaceMigration(tx as unknown as Db, workspace); - if (canonical(continuity) !== canonical(operation.metadata?.nativeSessionContinuity ?? null)) throw new Error("native_workspace_migration_source_changed"); - const authority: RemoteWorkspaceAuthority = { schema: "paperclip.remote-workspace.v1", epoch: 1, snapshotId: ref.id, - source: reserved.source, location: null, writerRunId: null, dirty: false }; - const metadata: Record = { ...workspace.metadata, remoteWorkspace: authority }; delete metadata.remoteWorkspaceMigration; - if (continuity) metadata.nativeWorkspaceMigration = nativeWorkspaceMigrationMarker(operation.id); - if (reserved.legacyRecoverySources.length) { - metadata.legacyWorkspaceRecoveryMigration = { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: operation.id }; - for (const source of reserved.legacyRecoverySources) { - const markerKey = source.reference?.state === "finalized" - ? "finalizedWorkspaceRecoveryMigration" : "legacyWorkspaceRecoveryMigration"; - // The offline deployment fence has stopped the old controller. The - // locked validation above must still match its journaled expired - // lease exactly. Release only that coordinator lease so explicit - // file recovery can proceed; its phase/checkpoint and the provider's - // independent stop obligation remain unchanged. - if (source.unresolvedFinalization?.leaseOwner) await tx.update(nativeRunFinalizations).set({ - leaseOwner: null, leaseExpiresAt: null, - recoveryHistory: sql`${nativeRunFinalizations.recoveryHistory} || ${JSON.stringify([{ - event: "legacy_workspace_migration_owner_released", operationId: operation.id, - workspaceId: workspace.id, previousOwner: source.unresolvedFinalization, - }])}::jsonb`, updatedAt: new Date(), - }).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId), - eq(nativeRunFinalizations.runId, source.runId), eq(nativeRunFinalizations.issueId, source.issueId))); - await tx.update(heartbeatRuns).set({ - runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ - [markerKey]: { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, - operationId: operation.id, workspaceId: workspace.id }, - })}::jsonb`, updatedAt: new Date(), - }).where(and(eq(heartbeatRuns.companyId, workspace.companyId), eq(heartbeatRuns.id, source.runId))); - } - const prior = Array.isArray(metadata.remoteWorkspaceRecoverySourceRunIds) ? metadata.remoteWorkspaceRecoverySourceRunIds : []; - metadata.remoteWorkspaceRecoverySourceRunIds = [...new Set([...prior, ...reserved.legacyRecoverySources.map(source => source.runId)])]; - } - await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); - await tx.update(executionWorkspaceSnapshots).set({ status: "committed", deleteAfter: null, - metadata: { ...uploaded.metadata, ...(nested.length ? { importedWorkspaceChildren: nested } : {}) }, - updatedAt: new Date() }).where(eq(executionWorkspaceSnapshots.id, ref.id)); - await tx.update(executionWorkspaces).set({ providerType: "cloud_sandbox", providerRef: null, cwd: `/workspaces/${workspace.id}`, metadata, - updatedAt: new Date() }).where(eq(executionWorkspaces.id, workspace.id)); - await tx.update(workspaceOperations).set({ status: "succeeded", finishedAt: new Date(), updatedAt: new Date(), - metadata: { ...operation.metadata, snapshotId: ref.id, manifestDigest: ref.manifestDigest } }).where(eq(workspaceOperations.id, operation.id)); - return (await persistActivity(tx as unknown as Db, { companyId: workspace.companyId, actorType: "system", actorId: "workspace-migration", - action: "workspace.remote_storage_migrated", entityType: "execution_workspace", entityId: workspace.id, - details: { snapshotId: ref.id, operationId: operation.id, manifestDigest: ref.manifestDigest } })).publication; - }).catch(error => { - const failure = object(error), cause = object(failure.cause); - if ((cause.code ?? failure.code) !== "55P03") throw error; - throw new Error("remote_workspace_migration_blocked:execution_rows_busy", { cause: error }); - }); - if (publication) { - if (onlineSource) committedPublication = publication; - else publishActivity(publication); - } + const validated = await validateMigrationPublication(tx, input, reserved, captured, onlineSource); + return validated ? publishMigration(tx, reserved, captured, validated) : null; + }).catch(publicationError); + if (publication) { if (onlineSource) committedPublication = publication; else publishActivity(publication); } return { ...await inventoryRemoteWorkspaceMigration(publicationDb, input), applied: true, replayed: false }; - } finally { await childVerifier?.close(); await removeTaskDirectoryTree(scratch); } + } finally { await childVerifier?.close(); if (scratch) await removeTaskDirectoryTree(scratch); } }; if (onlineSource) { const result = await withWorkspaceSourceLocks(input.db, { @@ -590,3 +655,155 @@ export async function migrateRemoteWorkspace(input: { } return withExecutionWorkspaceLifecycleLock(input.db, input.workspaceId, migrate); } + +const cohortSchema = z.object({ + schema: z.literal("paperclip.remote-workspace-cohort.v1"), id: z.string().uuid(), companyId: z.string().uuid(), + sourceProof: z.string().regex(/^[a-f0-9]{64}$/), + members: z.array(z.object({ workspaceId: z.string().uuid(), operationId: z.string().uuid() }).strict()).min(1).max(8), +}).strict(); + +export async function migrateRemoteWorkspaceCohort(input: { + db: Db; storage: StorageService; companyId: string; workspaceIds: readonly string[]; + apply?: boolean; concurrency?: number; signal: AbortSignal; +}) { + const companyId = z.string().uuid().parse(input.companyId).toLowerCase(); + const workspaceIds = z.array(z.string().uuid()).min(1).max(8).parse(input.workspaceIds).map(id => id.toLowerCase()).sort(); + const concurrency = z.number().int().min(1).max(8).parse(input.concurrency ?? 2); + if (new Set(workspaceIds).size !== workspaceIds.length) throw new Error("workspace_cohort_duplicate_member"); + input.signal.throwIfAborted(); + const initial = await input.db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, companyId), inArray(executionWorkspaces.id, workspaceIds))).orderBy(executionWorkspaces.id); + if (initial.length !== workspaceIds.length) throw new Error("workspace_cohort_member_missing"); + const imported = initial.map(workspace => Boolean(readRemoteWorkspaceAuthority(workspace.metadata))); + if (imported.every(Boolean)) { + const snapshotIds = initial.flatMap(workspace => readRemoteWorkspaceAuthority(workspace.metadata)?.snapshotId ?? []); + if (snapshotIds.length !== workspaceIds.length) throw new Error("workspace_cohort_membership_changed"); + const saved = await input.db.select({ snapshot: executionWorkspaceSnapshots, operation: workspaceOperations }) + .from(executionWorkspaceSnapshots).innerJoin(workspaceOperations, and( + eq(workspaceOperations.companyId, executionWorkspaceSnapshots.companyId), + eq(workspaceOperations.id, executionWorkspaceSnapshots.sourceOperationId))) + .where(and(eq(executionWorkspaceSnapshots.companyId, companyId), inArray(executionWorkspaceSnapshots.id, snapshotIds), + inArray(executionWorkspaceSnapshots.executionWorkspaceId, workspaceIds))); + const parsed = cohortSchema.safeParse(saved[0]?.operation.metadata?.cohortMigration); + if (!parsed.success) throw new Error("workspace_cohort_membership_changed"); + const cohort = parsed.data; + if (saved.length !== workspaceIds.length || cohort.companyId !== companyId + || canonical(cohort.members.map(member => member.workspaceId)) !== canonical(workspaceIds) + || new Set(cohort.members.map(member => member.operationId)).size !== workspaceIds.length + || saved.some(({ snapshot, operation }) => operation.status !== "succeeded" || operation.phase !== "remote_workspace_migration" + || operation.executionWorkspaceId !== snapshot.executionWorkspaceId || !["committed", "superseded"].includes(snapshot.status) + || operation.metadata?.snapshotId !== snapshot.id || operation.metadata?.manifestDigest !== snapshot.manifestDigest + || cohort.members.find(member => member.workspaceId === snapshot.executionWorkspaceId)?.operationId !== operation.id + || canonical(operation.metadata?.cohortMigration) !== canonical(cohort))) throw new Error("workspace_cohort_membership_changed"); + return { companyId, cohortId: cohort.id, applied: Boolean(input.apply), replayed: true, + members: await Promise.all(workspaceIds.map(workspaceId => inventoryRemoteWorkspaceMigration(input.db, { companyId, workspaceId }))) }; + } + if (imported.some(Boolean)) throw new Error("workspace_cohort_membership_changed"); + const stop = new AbortController(), signal = AbortSignal.any([input.signal, stop.signal]); + const members = await Promise.all(workspaceIds.map(async workspaceId => { + const memberInput: MigrationInput = { db: input.db, storage: input.storage, companyId, workspaceId, apply: true, signal }; + const onlineSource = await resolveMigrationSource(memberInput); + if (!onlineSource) throw new Error("workspace_cohort_online_source_required"); + return { input: memberInput, onlineSource }; + })); + const scopeIdentity = (scope: OnlineWorkspaceSource) => canonical({ kind: scope.kind, + projectIds: scope.projectIds, issueIds: scope.issueIds, workspaceIds: scope.workspaceIds }); + if (members.some(member => scopeIdentity(member.onlineSource) !== scopeIdentity(members[0].onlineSource))) + throw new Error("workspace_cohort_source_scope_mismatch"); + for (const member of members) for (const other of members) { + if (member !== other && member.onlineSource.root !== other.onlineSource.root + && member.onlineSource.root.startsWith(other.onlineSource.root + path.sep)) throw new Error("workspace_cohort_leaf_members_required"); + } + const sourceProof = createHash("sha256").update(canonical({ scope: scopeIdentity(members[0].onlineSource), + members: initial.map(workspace => ({ workspaceId: workspace.id, fingerprint: fingerprint(workspace) })) })).digest("hex"); + const pending = []; + for (const workspace of initial) { + const marker = object(workspace.metadata?.remoteWorkspaceMigration); + if (marker.schema !== SCHEMA || typeof marker.operationId !== "string") continue; + const [operation] = await input.db.select().from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, companyId), eq(workspaceOperations.executionWorkspaceId, workspace.id), + eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"))); + if (!operation?.metadata?.cohortMigration) throw new Error("workspace_cohort_foreign_journal"); + if (operation.status !== "running" || operation.finishedAt !== null) throw new Error("workspace_cohort_journal_changed"); + pending.push(cohortSchema.parse(operation.metadata.cohortMigration)); + } + const cohort: WorkspaceMigrationCohort = pending[0] ?? { schema: "paperclip.remote-workspace-cohort.v1", id: randomUUID(), companyId, + sourceProof, members: workspaceIds.map(workspaceId => ({ workspaceId, operationId: randomUUID() })) }; + if (cohort.companyId !== companyId || cohort.sourceProof !== sourceProof + || canonical(cohort.members.map(member => member.workspaceId)) !== canonical(workspaceIds) + || new Set(cohort.members.map(member => member.operationId)).size !== workspaceIds.length + || pending.some(marker => canonical(marker) !== canonical(cohort))) throw new Error("workspace_cohort_membership_changed"); + const preflight = async (db: ReadDb) => { + for (const member of members) { + const current = await inspect(db, { ...member.input, onlineSource: member.onlineSource }, cohort); + if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); + if (fingerprint(current.workspace) !== fingerprint(initial.find(workspace => workspace.id === member.input.workspaceId)!)) + throw new Error("workspace_cohort_source_changed"); + } + }; + if (!input.apply) { await preflight(input.db); return { companyId, cohortId: cohort.id, applied: false, eligible: true, workspaceIds }; } + const publications: Parameters[0][] = []; + const result = await withIndependentWorkspaceSourceLocks(input.db, { companyId, + projectIds: members[0].onlineSource.projectIds, issueIds: members[0].onlineSource.issueIds }, async lock => { + for (const workspaceId of members[0].onlineSource.workspaceIds) { + if (!await tryAcquireExecutionWorkspaceLifecycleLock(lock, workspaceId)) throw new IsolatedRepositorySourceBusy("workspace_source_busy"); + } + await preflight(input.db); + const plans: MigrationPlan[] = []; + for (const member of members) { + const reserved = await reserveMigration(member.input, member.onlineSource, cohort, + fingerprint(initial.find(workspace => workspace.id === member.input.workspaceId)!)); + if (reserved.authority) throw new Error("workspace_cohort_membership_changed"); + plans.push({ ...member, reserved }); + } + await lockMigrationRecovery(lock, plans, cohort); + const captures: Array = Array(plans.length); + const childVerifier = createMigrationChildVerifier(plans[0].input, plans[0].onlineSource, lock); + let next = 0, failure: unknown; + const worker = async () => { + while (!signal.aborted && next < plans.length) { + const index = next++, plan = plans[index]; + try { captures[index] = await captureMigration(plan.input, plan.reserved, plan.onlineSource, childVerifier); } + catch (error) { failure ??= error; stop.abort(error); } + } + }; + try { + const peers = await Promise.allSettled(Array.from({ length: Math.min(concurrency, plans.length) }, worker)); + for (const peer of peers) if (peer.status === "rejected") failure ??= peer.reason; + if (failure) throw failure; + signal.throwIfAborted(); + await lock.execute(sql`select 1`); + await lock.transaction(async tx => { + await lockMigrationParents(tx, companyId, null, plans.flatMap(plan => [ + plan.reserved.workspace.sourceIssueId!, ...plan.reserved.legacyRecoverySources.map(source => source.issueId), + ...plan.onlineSource!.companions.flatMap(proof => proof.sources.map(source => source.issueId)), + ])); + const validated = []; + for (const [index, plan] of plans.entries()) { + signal.throwIfAborted(); + const current = await validateMigrationPublication(tx, plan.input, plan.reserved, captures[index]!, plan.onlineSource, cohort); + if (!current) throw new Error("workspace_cohort_publication_changed"); + validated.push(current); + } + for (const [index, plan] of plans.entries()) { + signal.throwIfAborted(); + publications.push(await publishMigration(tx, plan.reserved, captures[index]!, validated[index])); + } + signal.throwIfAborted(); + }).catch(publicationError); + const results = []; + for (const workspaceId of workspaceIds) results.push(await inventoryRemoteWorkspaceMigration(lock as unknown as Db, { companyId, workspaceId })); + return { companyId, cohortId: cohort.id, applied: true, replayed: false, members: results }; + } catch (error) { failure = error; throw error; } + finally { + await childVerifier?.close(); + const cleanup = await Promise.allSettled(captures.filter((capture): capture is CapturedMigration => Boolean(capture)) + .map(capture => removeTaskDirectoryTree(capture.scratch))); + const errors = cleanup.flatMap(outcome => outcome.status === "rejected" ? [outcome.reason] : []); + if (errors.length) throw new AggregateError(failure ? [failure, ...errors] : errors, "workspace_cohort_cleanup_failed"); + if (!failure) signal.throwIfAborted(); + } + }, true); + for (const publication of publications) publishActivity(publication); + return result; +}