diff --git a/doc/DEVELOPING.md b/doc/DEVELOPING.md index 78621549793..7e0e7cc7e62 100644 --- a/doc/DEVELOPING.md +++ b/doc/DEVELOPING.md @@ -1894,6 +1894,18 @@ corrupt, or unbound sources block the import. Publication still requires an independent download and restore of the new encrypted snapshot. The old folder remains absent, and the importer does not change the original restore point. +Shared physical folders require Cloud's offline enrollment maintenance process. +A task-level hold cannot fence another workspace record pointing at the same +folder. Maintenance verifies that legacy deployments have stopped and that no +runs are active or queued, then imports nested task directories before parent +roots. A parent can omit a runtime-created nested task folder only after an +independent restore and source comparison prove that its committed snapshot +covers those bytes. The parent snapshot records that dependency, and retention +pins the child snapshot until the parent is deleted. Changed or unavailable +nested data blocks publication. Ordinary user folders remain in the archive. +Interrupted maintenance resumes the same import journals; completed records +leave the pending set. Original folders remain on the retained volume. + Supported sources include empty task folders, repository URLs, and managed mutable repositories. Authorized mentioned projects attach all their configured Git repositories as writable copies under `.paperclip-repositories/` in the task diff --git a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md index 65c7dd3d351..9bf24ecb59c 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -23,8 +23,10 @@ encrypted task-directory restoration, Git restoration recipes, and legacy branch-deletion receipts. PostgreSQL and real-file tests prove cold restoration of saved bytes, index state, and unpushed commits, including after pruning the old Git object store. Invalid or unavailable retained data still blocks import. -Shared-root migration and recovery-obligation reconciliation remain separate -gaps; this slice does not make a whole tenant eligible for enrollment. +The next slice adds leaf-first offline migration of overlapping folders, verified +nested snapshot dependencies, interruption recovery, and dependency retention. +Recovery-obligation reconciliation remains a separate gap; these slices do not +yet make the whole tenant eligible for enrollment. - Import existing files or a verified retained source. A missing folder alone is not permission to publish an empty workspace or choose a newer Git branch. diff --git a/server/src/maintenance/ephemeral-migration-data.ts b/server/src/maintenance/ephemeral-migration-data.ts index 17c713b0f12..eda8e547d6e 100644 --- a/server/src/maintenance/ephemeral-migration-data.ts +++ b/server/src/maintenance/ephemeral-migration-data.ts @@ -8,6 +8,7 @@ import { skillInstallationMaintenance } from "../services/skill-installation-mai import { skillPackageMaintenance } from "../services/skill-package-maintenance.js"; import { hasUnsettledLocalInstructionCopies } from "../services/agent-instruction-working-copies.js"; import { providerTraceStore } from "../services/provider-trace-store.js"; +import { migrateWorkspaceTrees } from "../services/remote-workspace-migration-batch.js"; /** Runs only in the maintenance process after Cloud has confirmed all legacy * deployments stopped. Each owner retains its own migration/publication rules; @@ -23,10 +24,15 @@ export async function migrateEphemeralServerData(db: Db, input: { if (await hasUnsettledLocalInstructionCopies(db)) throw new Error("local_instruction_copies_require_settlement"); const before = await readEphemeralServerDataEligibility(db); const unsupported = before.blockers.filter(blocker => ![ - "agent_files_require_migration", "run_logs_require_migration", "provider_traces_require_migration", "skills_require_migration", + "agent_files_require_migration", "run_logs_require_migration", "provider_traces_require_migration", "skills_require_migration", "workspaces_require_migration", ].includes(blocker)); if (unsupported.length) throw new Error(unsupported[0]); + phase("workspaces"); + const storage = getStorageService(); + if (storage.provider !== "s3") throw new Error("shared_object_storage_required"); + counts.workspaces = await migrateWorkspaceTrees(db, { storage, signal, phase }); + phase("agent_files"); let after: string | undefined; do { @@ -77,8 +83,6 @@ export async function migrateEphemeralServerData(db: Db, input: { } phase("verify_agent_objects"); - const storage = getStorageService(); - if (storage.provider !== "s3") throw new Error("shared_object_storage_required"); counts.verifiedAgentObjects = await verifyAgentFileObjects(db, storage, signal); phase("verify_skill_versions"); after = undefined; diff --git a/server/src/services/remote-workspace-migration-batch.ts b/server/src/services/remote-workspace-migration-batch.ts new file mode 100644 index 00000000000..327cff16d20 --- /dev/null +++ b/server/src/services/remote-workspace-migration-batch.ts @@ -0,0 +1,31 @@ +import { and, asc, desc, isNotNull, sql } from "drizzle-orm"; +import { executionWorkspaces, type Db } from "@paperclipai/db"; +import type { StorageService } from "../storage/types.js"; +import { offlineWorkspaceMigrationBlockers } from "./remote-workspace-migration-tree.js"; +import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; + +/** The Cloud maintenance process owns the stop-all-deployments precondition. + * Imported records leave the pending set, so a restart resumes the same owner + * journals without moving the task binding or importing over remote authority. */ +export async function migrateWorkspaceTrees(db: Db, input: { + storage: StorageService; signal: AbortSignal; phase: (name: string) => void; +}) { + const blockers = await offlineWorkspaceMigrationBlockers(db); + if (blockers.length) throw new Error(blockers[0]); + let migrated = 0; + for (;;) { + input.signal.throwIfAborted(); + const rows = await db.select({ companyId: executionWorkspaces.companyId, workspaceId: executionWorkspaces.id }) + .from(executionWorkspaces).where(and(isNotNull(executionWorkspaces.cwd), + sql`${executionWorkspaces.providerType} <> 'plugin'`, + sql`coalesce(${executionWorkspaces.metadata}->'remoteWorkspace'->>'schema', '') <> 'paperclip.remote-workspace.v1'`, + )).orderBy(desc(sql`length(${executionWorkspaces.cwd})`), asc(executionWorkspaces.id)).limit(25); + if (!rows.length) return migrated; + for (const workspace of rows) { + input.signal.throwIfAborted(); + input.phase(`workspace:${workspace.workspaceId}`); + await migrateRemoteWorkspace({ db, storage: input.storage, ...workspace, apply: true, writersStopped: true, signal: input.signal }); + migrated++; + } + } +} diff --git a/server/src/services/remote-workspace-migration-tree.ts b/server/src/services/remote-workspace-migration-tree.ts new file mode 100644 index 00000000000..a42c019c4b6 --- /dev/null +++ b/server/src/services/remote-workspace-migration-tree.ts @@ -0,0 +1,122 @@ +import fs from "node:fs/promises"; +import path from "node:path"; +import { and, eq, inArray, ne, sql } from "drizzle-orm"; +import { executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, workspaceOperations, type Db } from "@paperclipai/db"; +import type { StorageService } from "../storage/types.js"; +import { captureRemoteWorkspaceDirectory, restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; +import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js"; + +type ReadDb = Pick; +type Workspace = typeof executionWorkspaces.$inferSelect; +export type ImportedWorkspaceChild = { + workspaceId: string; snapshotId: string; operationId: string; relativePath: string; manifestDigest: string; +}; +const overlaps = (a: string, b: string) => a === b || a.startsWith(`${b}/`) || b.startsWith(`${a}/`); +const rootOf = (workspace: Pick) => workspace.cwd ?? workspace.providerRef; + +export async function offlineWorkspaceMigrationBlockers(db: ReadDb) { + if (process.env.PAPERCLIP_EPHEMERAL_MIGRATION !== "true") return ["offline_maintenance_required"]; + const [active] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(inArray(heartbeatRuns.status, ["running", "queued"])).limit(1); + return active ? ["active_execution_requires_drain"] : []; +} + +/** Only the maintenance entry point can acknowledge that every old deployment + * has stopped. A task-level migration hold cannot fence other path aliases. */ +export async function workspaceMigrationLayoutBlockers(db: ReadDb, workspace: Workspace, writersStopped = false) { + if (writersStopped) return offlineWorkspaceMigrationBlockers(db); + const cwd = rootOf(workspace); + if (!cwd) return []; + const local = await db.select({ id: executionWorkspaces.id, cwd: executionWorkspaces.cwd, + providerRef: executionWorkspaces.providerRef }).from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, workspace.companyId), ne(executionWorkspaces.id, workspace.id), + inArray(executionWorkspaces.providerType, ["local_fs", "git_worktree"]), + )); + if (local.some(row => rootOf(row) && overlaps(path.resolve(cwd), path.resolve(rootOf(row)!)))) { + return ["overlapping_workspace_requires_maintenance"]; + } + const migrated = await db.select({ cwd: workspaceOperations.cwd }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, workspace.companyId), eq(workspaceOperations.phase, "remote_workspace_migration"), + eq(workspaceOperations.status, "succeeded"), ne(workspaceOperations.executionWorkspaceId, workspace.id), + )); + return migrated.some(row => row.cwd && path.resolve(row.cwd).startsWith(`${path.resolve(cwd)}/.paperclip/worktrees/`)) + ? ["overlapping_workspace_requires_maintenance"] : []; +} + +/** Runtime-created nested task folders can be omitted only after an exact + * committed import covers their current bytes. Ordinary user folders stay in + * the parent archive. The snapshot owner retains the resulting dependencies. */ +export async function prepareNestedWorkspaceImports(input: { + db: Db; storage: StorageService; workspace: Workspace; scratch: string; signal?: AbortSignal; +}): Promise { + const root = rootOf(input.workspace); + if (!root) return []; + const rows = await input.db.select({ operation: workspaceOperations, snapshot: executionWorkspaceSnapshots }) + .from(workspaceOperations).innerJoin(executionWorkspaceSnapshots, and( + eq(executionWorkspaceSnapshots.companyId, workspaceOperations.companyId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, workspaceOperations.executionWorkspaceId), + eq(executionWorkspaceSnapshots.sourceOperationId, workspaceOperations.id), + )).where(and(eq(workspaceOperations.companyId, input.workspace.companyId), + ne(workspaceOperations.executionWorkspaceId, input.workspace.id), eq(workspaceOperations.phase, "remote_workspace_migration"), + eq(workspaceOperations.status, "succeeded"), inArray(executionWorkspaceSnapshots.status, ["committed", "superseded"]), + sql`${workspaceOperations.metadata}->>'snapshotId' = ${executionWorkspaceSnapshots.id}::text`, + )); + const candidates = rows.filter(row => row.operation.cwd && path.resolve(row.operation.cwd).startsWith(`${path.resolve(root)}/.paperclip/worktrees/`)) + .sort((a, b) => a.operation.cwd!.length - b.operation.cwd!.length || a.operation.id.localeCompare(b.operation.id)); + const covered: ImportedWorkspaceChild[] = []; + for (const { operation, snapshot } of candidates) { + input.signal?.throwIfAborted(); + const relativePath = path.relative(root, operation.cwd!).split(path.sep).join("/"); + if (covered.some(item => relativePath === item.relativePath || relativePath.startsWith(`${item.relativePath}/`))) continue; + const entry = await fs.lstat(operation.cwd!).catch((error: NodeJS.ErrnoException) => { if (error.code === "ENOENT") return null; throw error; }); + if (!entry) continue; + if (!entry.isDirectory() || entry.isSymbolicLink()) throw new Error("nested_workspace_source_changed"); + await verifyCoveredWorkspace({ ...input, sourcePath: operation.cwd!, snapshot, operationId: operation.id, + expectedDigest: operation.metadata?.manifestDigest, depth: 0 }); + covered.push({ workspaceId: snapshot.executionWorkspaceId, snapshotId: snapshot.id, operationId: operation.id, + relativePath, manifestDigest: snapshot.manifestDigest! }); + } + return covered; +} + +async function verifyCoveredWorkspace(input: { + db: Db; storage: StorageService; sourcePath: string; snapshot: typeof executionWorkspaceSnapshots.$inferSelect; + operationId: string; expectedDigest: unknown; scratch: string; signal?: AbortSignal; depth: number; +}) { + input.signal?.throwIfAborted(); + if (input.depth > 32) throw new Error("nested_workspace_coverage_invalid"); + const { snapshot } = input; + const reference = await getWorkspaceSnapshot(input.db, { companyId: snapshot.companyId, + workspaceId: snapshot.executionWorkspaceId, snapshotId: snapshot.id }); + if (!reference || reference.manifestDigest !== input.expectedDigest || snapshot.sourceOperationId !== input.operationId) + throw new Error("nested_workspace_snapshot_changed"); + const raw = snapshot.metadata?.importedWorkspaceChildren; + if (raw !== undefined && (!Array.isArray(raw) || raw.some(child => !child || typeof child !== "object" + || typeof child.relativePath !== "string" || !child.relativePath.startsWith(".paperclip/worktrees/") + || child.relativePath.split("/").some((part: string) => !part || part === "." || part === "..") + || !/^[a-f0-9-]{36}$/.test(child.workspaceId) || !/^[a-f0-9-]{36}$/.test(child.snapshotId) + || !/^[a-f0-9-]{36}$/.test(child.operationId) || !/^[a-f0-9]{64}$/.test(child.manifestDigest)))) { + throw new Error("nested_workspace_coverage_invalid"); + } + const children = (raw ?? []) as ImportedWorkspaceChild[]; + const area = await fs.mkdtemp(path.join(input.scratch, "covered-")); + for (const child of children) { + const [saved] = await input.db.select({ operation: workspaceOperations, snapshot: executionWorkspaceSnapshots }) + .from(executionWorkspaceSnapshots).innerJoin(workspaceOperations, and( + eq(workspaceOperations.companyId, executionWorkspaceSnapshots.companyId), eq(workspaceOperations.id, executionWorkspaceSnapshots.sourceOperationId), + )).where(and(eq(executionWorkspaceSnapshots.companyId, snapshot.companyId), eq(executionWorkspaceSnapshots.id, child.snapshotId), + eq(executionWorkspaceSnapshots.executionWorkspaceId, child.workspaceId), eq(workspaceOperations.id, child.operationId), + eq(workspaceOperations.status, "succeeded"), eq(workspaceOperations.phase, "remote_workspace_migration"), + inArray(executionWorkspaceSnapshots.status, ["committed", "superseded"]))); + const sourcePath = path.join(input.sourcePath, child.relativePath); + if (!saved || saved.operation.cwd !== sourcePath) throw new Error("nested_workspace_coverage_invalid"); + await verifyCoveredWorkspace({ ...input, sourcePath, snapshot: saved.snapshot, operationId: child.operationId, + expectedDigest: child.manifestDigest, scratch: area, depth: input.depth + 1 }); + } + const archivePath = path.join(area, "saved.pcws"), restored = path.join(area, "restored"); + await loadWorkspaceSnapshotArchive({ storage: input.storage, reference, targetPath: archivePath, signal: input.signal }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir: restored, expectedManifestDigest: reference.manifestDigest, signal: input.signal }); + const current = await captureRemoteWorkspaceDirectory({ cwd: input.sourcePath, archivePath: path.join(area, "current.pcws"), + exclude: children.map(child => child.relativePath), signal: input.signal }); + if (current.manifestDigest !== reference.manifestDigest) throw new Error("nested_workspace_source_changed"); +} diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index e2345e3cb88..a4887fbe94b 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -8,13 +8,14 @@ import { createServer, type Server } from "node:http"; import { pathToFileURL } from "node:url"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; import { eq } from "drizzle-orm"; -import { activityLog, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots, +import { activityLog, agents, heartbeatRuns, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots, executionWorkspaceRuntimeLeases, executionWorkspaces, issues, projects, projectWorkspaces, workspaceOperations, workspaceRuntimeServices } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; import { createStorageService } from "../storage/service.js"; import type { StorageService } from "../storage/types.js"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { migrateWorkspaceTrees } from "./remote-workspace-migration-batch.js"; import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace } from "./remote-workspace-migration.js"; import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js"; import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; @@ -50,7 +51,7 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await Promise.all(servers.splice(0).map(server => new Promise((resolve, reject) => server.close(error => error ? reject(error) : resolve())))); vi.unstubAllEnvs(); await db.delete(activityLog); await db.delete(workspaceOperations); await db.delete(executionWorkspaceSnapshots); await db.delete(workspaceRuntimeServices); await db.delete(executionWorkspaceRuntimeLeases); - await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); + await db.delete(heartbeatRuns); await db.delete(agents); await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); await db.delete(companies); await fs.rm(root, { recursive: true, force: true }); }); async function seed() { @@ -64,6 +65,108 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { return { companyId, workspaceId, issueId: issue!.id, cwd }; } async function workspace(id: string) { return (await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, id)))[0]!; } + async function nestedFixture() { + const parent = await seed(), workspaceId = randomUUID(); + const cwd = path.join(parent.cwd, ".paperclip", "worktrees", "child"); + await fs.mkdir(cwd, { recursive: true }); await fs.writeFile(path.join(cwd, "child.txt"), "saved child edits\n"); + const [issue] = await db.insert(issues).values({ companyId: parent.companyId, title: "Nested task" }).returning(); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: parent.companyId, sourceIssueId: issue!.id, + name: "Child workspace", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd, providerRef: cwd }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, issue!.id)); + return { parent, child: { companyId: parent.companyId, workspaceId, issueId: issue!.id, cwd } }; + } + it("requires offline maintenance for overlapping physical workspace paths", async () => { + const { parent, child } = await nestedFixture(); + await expect(migrateRemoteWorkspace({ db, storage, ...child, apply: true })).rejects.toThrow("overlapping_workspace_requires_maintenance"); + await expect(migrateRemoteWorkspace({ db, storage, ...parent, apply: true })).rejects.toThrow("overlapping_workspace_requires_maintenance"); + expect(await fs.readFile(path.join(child.cwd, "child.txt"), "utf8")).toBe("saved child edits\n"); + }); + it("imports nested task files before their parent and retains independently restorable copies", async () => { + const { parent, child } = await nestedFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await migrateRemoteWorkspace({ db, storage, ...child, apply: true, writersStopped: true }); + await migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true }); + await fs.rm(parent.cwd, { recursive: true }); + for (const [name, fixture] of [["parent", parent], ["child", child]] as const) { + const authority = readRemoteWorkspaceAuthority((await workspace(fixture.workspaceId)).metadata)!; + const ref = await getWorkspaceSnapshot(db, { ...fixture, snapshotId: authority.snapshotId! }); + const archivePath = path.join(root, `${name}.pcws`), targetDir = path.join(root, `restored-${name}`); + await loadWorkspaceSnapshotArchive({ storage, reference: ref!, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref!.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, name === "parent" ? "saved.txt" : "child.txt"), "utf8")) + .toBe(name === "parent" ? "original saved work\n" : "saved child edits\n"); + if (name === "parent") { + expect(await fs.stat(path.join(targetDir, ".paperclip", "worktrees", "child")).catch(() => null)).toBeNull(); + const [snapshot] = await db.select().from(executionWorkspaceSnapshots).where(eq(executionWorkspaceSnapshots.id, ref!.id)); + expect(snapshot.metadata?.importedWorkspaceChildren).toEqual([expect.objectContaining({ workspaceId: child.workspaceId, + relativePath: ".paperclip/worktrees/child" })]); + } + } + }); + it("requires the offline entry point even when a caller asserts writers stopped", async () => { + const { parent } = await nestedFixture(); + await expect(migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true })).rejects.toThrow("offline_maintenance_required"); + await expect(migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).rejects.toThrow("offline_maintenance_required"); + }); + it.each(["running", "queued"])("refuses offline archived imports while any execution is %s", async status => { + const f = await seed(); await retainDirectory(f); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Unrelated writer" }).returning(); + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: agent!.id, status }); + await expect(migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })) + .rejects.toThrow("active_execution_requires_drain"); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + }); + it("blocks distinct legacy records that alias the same folder", async () => { + const f = await seed(), id = randomUUID(); + await db.insert(executionWorkspaces).values({ id, companyId: f.companyId, name: "Historical alias", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd: f.cwd }); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("overlapping_workspace_requires_maintenance"); + }); + it.each(["changed", "corrupt", "late-change", "abort"])("keeps the parent local when nested coverage is unsafe (%s)", async mode => { + const { parent, child } = await nestedFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await migrateRemoteWorkspace({ db, storage, ...child, apply: true, writersStopped: true }); + const authority = readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)!; + const reference = await getWorkspaceSnapshot(db, { ...child, snapshotId: authority.snapshotId! }); + const controller = new AbortController(); + if (mode === "changed") await fs.writeFile(path.join(child.cwd, "child.txt"), "changed child bytes"); + if (mode === "corrupt") await storage.deleteObject(child.companyId, reference!.objectKey); + if (mode === "late-change" || mode === "abort") { + const put = storage.putFile.bind(storage); let once = true; + storage.putFile = async input => { const result = await put(input); if (once) { once = false; + if (mode === "abort") controller.abort(); else await fs.writeFile(path.join(child.cwd, "child.txt"), "changed during upload"); + } return result; }; + } + await expect(migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true, signal: controller.signal })).rejects.toThrow(); + expect(readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)).toBeNull(); + expect(await fs.readFile(path.join(parent.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); + if (mode === "abort") { + await migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true }); + expect(readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)?.snapshotId).toBeTruthy(); + } + }); + it("resumes an interrupted maintenance batch in leaf-first order without duplicating imports", async () => { + const { parent, child } = await nestedFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const seen: string[] = [], controller = new AbortController(); + await expect(migrateWorkspaceTrees(db, { storage, signal: controller.signal, phase: name => { + seen.push(name); if (name === `workspace:${parent.workspaceId}`) controller.abort(); + } })).rejects.toThrow(); + expect(seen).toEqual([`workspace:${child.workspaceId}`, `workspace:${parent.workspaceId}`]); + const childAuthority = readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata); + expect(await migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).toBe(1); + expect(await migrateWorkspaceTrees(db, { storage, signal: new AbortController().signal, phase: () => {} })).toBe(0); + expect(readRemoteWorkspaceAuthority((await workspace(child.workspaceId)).metadata)).toEqual(childAuthority); + }); + it("checks transitive nested coverage before omitting a child tree", async () => { + const { parent, child } = await nestedFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const id = randomUUID(), cwd = path.join(child.cwd, ".paperclip", "worktrees", "grandchild"); + await fs.mkdir(cwd, { recursive: true }); await fs.writeFile(path.join(cwd, "deep.txt"), "deep work"); + const [issue] = await db.insert(issues).values({ companyId: parent.companyId, title: "Grandchild task" }).returning(); + await db.insert(executionWorkspaces).values({ id, companyId: parent.companyId, sourceIssueId: issue!.id, + name: "Grandchild", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd }); + await migrateRemoteWorkspace({ db, storage, companyId: parent.companyId, workspaceId: id, apply: true, writersStopped: true }); + await migrateRemoteWorkspace({ db, storage, ...child, apply: true, writersStopped: true }); + await fs.writeFile(path.join(cwd, "deep.txt"), "new deep work"); + await expect(migrateRemoteWorkspace({ db, storage, ...parent, apply: true, writersStopped: true })).rejects.toThrow("nested_workspace_source_changed"); + expect(readRemoteWorkspaceAuthority((await workspace(parent.workspaceId)).metadata)).toBeNull(); + }); async function gitArchiveFixture(repoUrl = "https://example.invalid/repository.git") { const f = await seed(), projectId = randomUUID(), projectWorkspaceId = randomUUID(), branchName = "saved-task"; vi.stubEnv("PAPERCLIP_HOME", root); vi.stubEnv("PAPERCLIP_INSTANCE_ID", "default"); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 47dd4c7435f..b0ccbd46f05 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -17,6 +17,7 @@ import { notifyRemoteWorkspaceSnapshots } from "./remote-workspace-snapshot-rete import { materializeTaskDirectorySnapshot, removeTaskDirectoryTree } from "./isolated-task-directory-snapshot.js"; import { loadTaskDirectoryRestorePoint, mayBeIsolatedTaskDirectoryWorkspace, readTaskDirectoryRetention, validateTaskDirectoryRestorePointRef } from "./isolated-task-directory-retention.js"; import { inspectArchivedGitMigrationSource, materializeArchivedGitMigrationSource, type ArchivedGitMigrationSource } from "./remote-workspace-migration-git.js"; +import { prepareNestedWorkspaceImports, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js"; const SCHEMA = "paperclip.remote-workspace-migration.v1"; type Workspace = typeof executionWorkspaces.$inferSelect; @@ -38,7 +39,7 @@ const gitSourceProof = (source: ArchivedGitMigrationSource | null) => source ? { commit: source.commit, sourcePath: source.sourcePath, cleanupOperationId: source.cleanupOperationId, } : null; -async function inspect(db: ReadDb, input: { companyId: string; workspaceId: string; operationId?: string }) { +async function inspect(db: ReadDb, input: { companyId: string; workspaceId: string; operationId?: string; writersStopped?: boolean }) { 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); @@ -77,6 +78,7 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri ? "restore_point_expired" : "archived_git_source_unavailable"); } } if (!retainedDirectory && !archivedGitSource && (!entry?.isDirectory() || entry.isSymbolicLink())) reasons.push("local_source_unsafe_or_missing"); + if (entry?.isDirectory() && !entry.isSymbolicLink()) reasons.push(...await workspaceMigrationLayoutBlockers(db, workspace, input.writersStopped)); } const [issue] = workspace.sourceIssueId ? await db.select().from(issues).where(and(eq(issues.companyId, workspace.companyId), eq(issues.id, workspace.sourceIssueId))) : []; // Historical records still own their saved files after the task selects a @@ -142,7 +144,7 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource }; } -export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string }) { +export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string; writersStopped?: boolean }) { const result = await inspect(db, input); return { companyId: input.companyId, workspaceId: input.workspaceId, eligible: result.reasons.length === 0, alreadyMigrated: Boolean(result.authority), reasons: result.reasons, localPath: result.workspace.cwd ?? result.workspace.providerRef, @@ -153,6 +155,7 @@ export async function inventoryRemoteWorkspaceMigration(db: Db, input: { company /** Administrative import. The CLI enforces external deployment prerequisites. */ export async function migrateRemoteWorkspace(input: { 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 }; return withExecutionWorkspaceLifecycleLock(input.db, input.workspaceId, async lock => { @@ -198,22 +201,32 @@ export async function migrateRemoteWorkspace(input: { operationId: operation.id, generation: 1, epoch: 1 }); const scratch = await fs.mkdtemp(path.join(os.tmpdir(), "pc-workspace-migration-")); try { + input.signal?.throwIfAborted(); const retained = reserved.retainedDirectory?.restorePoint; const sourceRoot = retained ? await materializeTaskDirectorySnapshot({ payload: await loadTaskDirectoryRestorePoint({ storage: input.storage, workspace: reserved.workspace, ref: retained }), 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 nested = input.writersStopped && !retained && !reserved.archivedGitSource + ? await prepareNestedWorkspaceImports({ ...input, workspace: reserved.workspace, scratch }) : []; + const exclude = nested.map(child => child.relativePath); const archivePath = path.join(scratch, "workspace.pcws"); - const captured = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot, archivePath }); + const captured = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot, archivePath, exclude, signal: input.signal }); const ref = await uploadWorkspaceSnapshot({ db: input.db, storage: input.storage, snapshotId: snapshot.id, - companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest }); + companyId: input.companyId, workspaceId: input.workspaceId, archivePath, manifestDigest: captured.manifestDigest, signal: input.signal }); const restoredArchive = path.join(scratch, "verified.pcws"), restored = path.join(scratch, "restored"); - await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: restoredArchive }); - await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest }); + await loadWorkspaceSnapshotArchive({ storage: input.storage, reference: ref, targetPath: restoredArchive, signal: input.signal }); + await restoreRemoteWorkspaceArchive({ archivePath: restoredArchive, targetDir: restored, expectedManifestDigest: ref.manifestDigest, signal: input.signal }); const currentArchive = path.join(scratch, "current.pcws"); - const current = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot, archivePath: currentArchive }); + const current = await captureRemoteWorkspaceDirectory({ cwd: sourceRoot, archivePath: currentArchive, exclude, 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 prepareNestedWorkspaceImports({ ...input, workspace: reserved.workspace, scratch: verifyScratch }); + if (canonical(verified) !== canonical(nested)) throw new Error("nested_workspace_source_changed"); + } + input.signal?.throwIfAborted(); await lock.execute(sql`select 1`); const publication = await input.db.transaction(async tx => { const [workspace] = await tx.select().from(executionWorkspaces).where(and(eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.workspaceId))).for("update"); @@ -231,11 +244,21 @@ export async function migrateRemoteWorkspace(input: { 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"); + } 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; await notifyRemoteWorkspaceSnapshots(tx, `workspace:${workspace.id}`); - await tx.update(executionWorkspaceSnapshots).set({ status: "committed", deleteAfter: null, updatedAt: new Date() }).where(eq(executionWorkspaceSnapshots.id, ref.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(), diff --git a/server/src/services/remote-workspace-snapshot-retention.test.ts b/server/src/services/remote-workspace-snapshot-retention.test.ts index d008d393d38..1bfb4d42996 100644 --- a/server/src/services/remote-workspace-snapshot-retention.test.ts +++ b/server/src/services/remote-workspace-snapshot-retention.test.ts @@ -131,6 +131,24 @@ describe.skipIf(!support.supported)("remote workspace snapshot retention", () => for (const key of f.keys) expect((await storage.headObject(f.companyId, key)).exists).toBe(false); expect(await owner.nextDeadlineAt()).toBeNull(); }); + it("pins imported child bytes until the covering parent snapshot is deleted", async () => { + const child = await seed(), parentId = randomUUID(), workspaceId = randomUUID(); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: child.companyId, name: "Imported parent", mode: "isolated_workspace", strategyType: "project_primary", + metadata: { remoteWorkspace: { snapshotId: parentId } } }); + await db.insert(executionWorkspaceSnapshots).values({ id: parentId, companyId: child.companyId, executionWorkspaceId: workspaceId, + sourceOperationId: randomUUID(), generation: 1, epoch: 1, status: "committed", + metadata: { importedWorkspaceChildren: [{ workspaceId: child.workspaceId, snapshotId: child.row.id }] } }); + const owner = service(); + expect(await owner.tick()).toMatchObject({ pinned: 1, deleted: 0 }); + expect(await owner.nextDeadlineAt()).toBeNull(); + expect((await storage.headObject(child.companyId, child.keys[0]!)).exists).toBe(true); + await db.update(executionWorkspaces).set({ metadata: {} }).where(eq(executionWorkspaces.id, workspaceId)); + await db.update(executionWorkspaceSnapshots).set({ status: "superseded", deleteAfter: new Date(now.getTime() - 1) }) + .where(eq(executionWorkspaceSnapshots.id, parentId)); + await owner.tick(); await owner.tick(); + expect((await storage.headObject(child.companyId, child.keys[0]!)).exists).toBe(false); + expect(await owner.nextDeadlineAt()).toBeNull(); + }); it("purges exact-key S3 versions and delete markers before settling retention", async () => { const s3 = await versionedS3Fixture(); try { diff --git a/server/src/services/remote-workspace-snapshot-retention.ts b/server/src/services/remote-workspace-snapshot-retention.ts index 22977698304..0b087a1ee14 100644 --- a/server/src/services/remote-workspace-snapshot-retention.ts +++ b/server/src/services/remote-workspace-snapshot-retention.ts @@ -23,6 +23,13 @@ async function pinned(db: ReadDb, row: Snapshot) { eq(executionWorkspaces.companyId, row.companyId), eq(executionWorkspaces.id, row.executionWorkspaceId), )); if ((workspace?.metadata?.remoteWorkspace as { snapshotId?: unknown } | undefined)?.snapshotId === row.id) return true; + const [importedBy] = await db.select({ id: executionWorkspaceSnapshots.id }).from(executionWorkspaceSnapshots).where(and( + eq(executionWorkspaceSnapshots.companyId, row.companyId), ne(executionWorkspaceSnapshots.status, "deleted"), + sql`${executionWorkspaceSnapshots.metadata}->'importedWorkspaceChildren' @> ${JSON.stringify([ + { workspaceId: row.executionWorkspaceId, snapshotId: row.id }, + ])}::jsonb`, + )).limit(1); + if (importedBy) return true; const [operation] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( eq(workspaceOperations.companyId, row.companyId), eq(workspaceOperations.executionWorkspaceId, row.executionWorkspaceId), or(inArray(workspaceOperations.status, ["running", "pending"]), @@ -138,7 +145,8 @@ export function remoteWorkspaceSnapshotRetentionService(db: Db, options: { .where(and(eq(executionWorkspaceSnapshots.id, claimed.id), eq(executionWorkspaceSnapshots.status, "delete_requested"))); // A parent inspected earlier in this page can now lose its last pin. // Revisit it once instead of waiting for an unrelated execution event. - if (claimed.parentSnapshotId) rescanRequested = true; + if (claimed.parentSnapshotId || Array.isArray(claimed.metadata?.importedWorkspaceChildren) + && claimed.metadata.importedWorkspaceChildren.length > 0) rescanRequested = true; result.deleted++; } catch (error) { if (signal?.aborted) throw error;