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 9bf24ecb59c..b3fbce2fe1c 100644 --- a/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md +++ b/doc/plans/2026-11-03-existing-workspace-ephemeral-migration.md @@ -45,3 +45,27 @@ yet make the whole tenant eligible for enrollment. bindings, archived sources, overlap/writer races, and unavailable sources. - Unrelated tenants, production services, and agent task execution remain outside the migration's mutation scope. + +## Legacy recovery storage transition + +The migration owner must distinguish file location from recovery completion. +An offline import can preserve a terminal failed run's local workspace without +claiming that its remote files were recovered. The snapshot and migration +receipt bind the original run and sync descriptor; the failed run, provider +allocation, recovery action, and unresolved state stay recorded. + +After publication, the native sync owner must reject copyback to the old local +folder. Explicit recovery still validates the exact provider allocation and +stop evidence, then preserves remote files as an immutable recovery attachment +instead of merging into retired local authority. New remote allocations attach +those verified recovery copies through the existing attachment owner. Missing +or ambiguous remote sources remain unresolved; migration never authorizes +replaying the old agent turn or destroying its allocation. + +Publication requires the offline deployment fence, terminal unowned runs, +unchanged descriptors and task bindings, verified local snapshot bytes, and a +single transaction publishing the migration receipt and remote authority. +Controller leases, in-flight recovery, active writers, and uncertain workspace +operations remain blockers. Tests must prove migration leaves failure state +intact, rejects stale copyback even if the old folder exists, and preserves +recovery source validation and company boundaries. diff --git a/server/src/services/execution-workspaces.ts b/server/src/services/execution-workspaces.ts index ba0d2fa8353..b4e3313667a 100644 --- a/server/src/services/execution-workspaces.ts +++ b/server/src/services/execution-workspaces.ts @@ -272,7 +272,7 @@ function preserveWorkspaceMetadataForPatch( if (patch[key] !== undefined && patch[key] !== current[key]) throw conflict("Remote workspace coordinates belong to its sandbox storage"); } } - const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles"] as const; + const remoteAuthorityKeys = ["remoteWorkspace", "remoteWorkspaceMigration", "remoteWorkspaceRecoverySourceRunIds", "remoteWorkspaceRecoveredFiles", "legacyWorkspaceRecoveryMigration"] as const; if (patch.metadata !== undefined && remoteAuthorityKeys.some(key => current.metadata?.[key] !== undefined || patch.metadata?.[key] !== undefined)) { nextPatch.metadata = { ...patch.metadata }; for (const key of remoteAuthorityKeys) { diff --git a/server/src/services/legacy-workspace-recovery-migration.ts b/server/src/services/legacy-workspace-recovery-migration.ts new file mode 100644 index 00000000000..e4b17f07fb9 --- /dev/null +++ b/server/src/services/legacy-workspace-recovery-migration.ts @@ -0,0 +1,51 @@ +import { isDeepStrictEqual } from "node:util"; +import { and, eq } from "drizzle-orm"; +import { executionWorkspaces, issues, workspaceOperations, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import type { NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; + +export type LegacyWorkspaceRecoverySource = { runId: string; issueId: string; reference: NativeWorkspaceSyncReference }; +export const LEGACY_WORKSPACE_RECOVERY_MIGRATION = "paperclip.legacy-workspace-recovery-migration.v1"; + +/** Child tasks can explicitly reuse an ancestor's workspace. The run's own + * task remains the scope for its recovery files; migration never rebinds it. */ +export async function isLegacyWorkspaceTaskRelated(db: Pick, input: { + companyId: string; issueId: string | null; sourceIssueId: string | null; +}) { + let current = input.issueId; + const visited = new Set(); + while (current && visited.size < 32 && !visited.has(current)) { + if (current === input.sourceIssueId) return true; + visited.add(current); + const [issue] = await db.select({ parentId: issues.parentId }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, current))); + current = issue?.parentId ?? null; + } + return false; +} + +/** This receipt changes where recovery may save files; it never certifies that + * a provider stopped or that its files were recovered. The recovery owner must + * still validate allocation provenance and termination before exporting. */ +export async function hasMigratedLegacyWorkspaceSource(db: Pick, input: { + companyId: string; runId: string; reference: NativeWorkspaceSyncReference; +}) { + if (input.reference.state !== "prepared") return false; + const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, input.reference.workspaceId), + )); + if (!readRemoteWorkspaceAuthority(workspace?.metadata)) return false; + const marker = workspace?.metadata?.legacyWorkspaceRecoveryMigration as { schema?: unknown; operationId?: unknown } | undefined; + if (marker?.schema !== LEGACY_WORKSPACE_RECOVERY_MIGRATION || typeof marker.operationId !== "string") + throw new Error("legacy_workspace_migration_receipt_invalid"); + const [operation] = await db.select({ metadata: workspaceOperations.metadata }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.executionWorkspaceId, input.reference.workspaceId), + eq(workspaceOperations.id, marker.operationId), eq(workspaceOperations.phase, "remote_workspace_migration"), + eq(workspaceOperations.status, "succeeded"), + )); + const sources = operation?.metadata?.legacyRecoverySources; + if (!Array.isArray(sources) || !sources.some(source => source && typeof source === "object" + && source.runId === input.runId && isDeepStrictEqual(source.reference, input.reference))) + throw new Error("legacy_workspace_migration_receipt_invalid"); + return true; +} diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index aa5a0c76e2a..a684714b53a 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/server/src/services/native-runtime/native-workspace-sync.ts @@ -1,3 +1,4 @@ +import { hasMigratedLegacyWorkspaceSource } from "../legacy-workspace-recovery-migration.js"; import { notifyCancelledWorkspacePreservation } from "../cancelled-workspace-preservation-notifications.js"; import { notifyEnvironmentCleanup } from "../environment-cleanup-notifications.js"; import { measureRunPhase } from "../execution-preparation-timing.js"; @@ -921,7 +922,7 @@ export async function prepareNativeWorkspaceSync(input: { throw new Error("native_remote_recovery_lease_mismatch"); } const run = await measureRunPhase("workspace_sync.run_read", () => input.db - .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) + .select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1) @@ -1184,7 +1185,7 @@ export async function resumeNativeWorkspaceSync(input: { throw new Error("workspace_sync_out_unrecoverable"); } const run = await input.db - .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson }) + .select({ companyId: heartbeatRuns.companyId, runnerProfileJson: heartbeatRuns.runnerProfileJson }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1) @@ -1193,6 +1194,8 @@ export async function resumeNativeWorkspaceSync(input: { parseObject(run?.runnerProfileJson).nativeWorkspaceSync, ); if (!reference) return false; + if (run && await hasMigratedLegacyWorkspaceSource(input.db, { companyId: run.companyId, runId: input.runId, reference })) + throw new Error("native_workspace_sync_source_migrated"); const existing = await readDescriptor({ runId: input.runId, reference }); const providerLeaseId = input.target.sandboxLeaseAcquisition?.providerLeaseId ?? diff --git a/server/src/services/remote-workspace-migration.test.ts b/server/src/services/remote-workspace-migration.test.ts index bce64e9e66d..b94a9a4fa57 100644 --- a/server/src/services/remote-workspace-migration.test.ts +++ b/server/src/services/remote-workspace-migration.test.ts @@ -8,7 +8,7 @@ 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, agents, heartbeatRuns, applyPendingMigrations, closeRegisteredClients, companies, createDb, executionWorkspaceSnapshots, +import { activityLog, agents, heartbeatRuns, nativeRunFinalizations, 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"; @@ -25,6 +25,7 @@ import type { TaskDirectoryRestorePointRef } from "./isolated-task-directory-ret import { resolveManagedProjectWorkspaceDir } from "../home-paths.js"; import { cleanupExecutionWorkspaceArtifacts } from "./workspace-runtime.js"; import { workspaceOperationService } from "./workspace-operations.js"; +import { resumeNativeWorkspaceSync } from "./native-runtime/native-workspace-sync.js"; import { managedProjectCheckoutName } from "./managed-project-checkout-name.js"; import { captureGitWorkspaceContents, readManagedCheckoutContents } from "./git-workspace-contents.js"; @@ -51,7 +52,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(heartbeatRuns); await db.delete(agents); await db.delete(issues); await db.delete(executionWorkspaces); await db.delete(projectWorkspaces); await db.delete(projects); + await db.delete(nativeRunFinalizations); 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() { @@ -140,6 +141,75 @@ describe.skipIf(!support.supported)("remote workspace migration", () => { await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: reference!.manifestDigest }); expect(await fs.readFile(path.join(targetDir, ".paperclip/worktrees/child/child.txt"), "utf8")).toBe("saved child edits\n"); }); + async function failedCopybackFixture() { + const f = await seed(), runId = randomUUID(); + const [agent] = await db.insert(agents).values({ companyId: f.companyId, name: "Historical agent" }).returning(); + const reference = { schema: "paperclip.native-workspace-sync/v3", state: "prepared", workspaceId: f.workspaceId, + leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/workspaces/task", + descriptorSha256: "12".repeat(32), baselineSha256: "34".repeat(32), finalHostSha256: null, resourceDisposition: "stop_and_retain" }; + await db.insert(heartbeatRuns).values({ id: runId, companyId: f.companyId, agentId: agent!.id, nativeIssueId: f.issueId, + runtimeMode: "native", status: "failed", finishedAt: new Date(), runnerProfileJson: { nativeWorkspaceSync: reference } }); + await db.insert(nativeRunFinalizations).values({ companyId: f.companyId, issueId: f.issueId, runId, phase: "terminal_failure" }); + return { ...f, runId, reference }; + } + it("carries unresolved copyback forward during offline migration without declaring recovery complete", async () => { + const f = await failedCopybackFixture(); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true })).rejects.toThrow("remote_copyback_unresolved"); + vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + const row = await workspace(f.workspaceId); + expect(readRemoteWorkspaceAuthority(row.metadata)?.snapshotId).toBeTruthy(); + expect(row.metadata?.remoteWorkspaceRecoverySourceRunIds).toEqual([f.runId]); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + expect(run).toMatchObject({ status: "failed", runnerProfileJson: { nativeWorkspaceSync: f.reference } }); + expect(await fs.readFile(path.join(f.cwd, "saved.txt"), "utf8")).toBe("original saved work\n"); + // An old copyback must not recreate or mutate the predecessor's local folder. + await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } })) + .rejects.toThrow("native_workspace_sync_source_migrated"); + }); + it.each(["controller", "finalizer", "retryable", "task", "descriptor"])("keeps unsafe legacy copyback blocked (%s)", async mode => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + if (mode === "controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "finalizer") await db.update(nativeRunFinalizations).set({ leaseOwner: randomUUID() }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "retryable") await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "task") { + const [other] = await db.insert(issues).values({ companyId: f.companyId, title: "Different task" }).returning(); + await db.update(executionWorkspaces).set({ sourceIssueId: other!.id }).where(eq(executionWorkspaces.id, f.workspaceId)); + } + if (mode === "descriptor") await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, descriptorSha256: "invalid" } } }).where(eq(heartbeatRuns.id, f.runId)); + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })).rejects.toThrow("remote_copyback_unresolved"); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + }); + it("rejects a legacy descriptor changed while its local files are uploading", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const put = storage.putFile.bind(storage); + storage.putFile = async input => { + const result = await put(input); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, providerLeaseId: randomUUID() } } }).where(eq(heartbeatRuns.id, f.runId)); + return result; + }; + await expect(migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true })).rejects.toThrow("remote_workspace_migration_source_changed"); + expect(readRemoteWorkspaceAuthority((await workspace(f.workspaceId)).metadata)).toBeNull(); + }); + it("preserves a child task's recovery reference to its ancestor's workspace", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + const [parent] = await db.insert(issues).values({ companyId: f.companyId, title: "Parent task" }).returning(); + await db.update(issues).set({ parentId: parent!.id }).where(eq(issues.id, f.issueId)); + await db.update(executionWorkspaces).set({ sourceIssueId: parent!.id }).where(eq(executionWorkspaces.id, f.workspaceId)); + await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + expect((await workspace(f.workspaceId)).sourceIssueId).toBe(parent!.id); + expect((await db.select().from(issues).where(eq(issues.id, f.issueId)))[0]!.executionWorkspaceId).toBe(f.workspaceId); + await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } })) + .rejects.toThrow("native_workspace_sync_source_migrated"); + }); + it("fails closed if the durable legacy migration receipt changes", async () => { + const f = await failedCopybackFixture(); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + await migrateRemoteWorkspace({ db, storage, ...f, apply: true, writersStopped: true }); + const row = await workspace(f.workspaceId); + await db.update(executionWorkspaces).set({ metadata: { ...row.metadata, legacyWorkspaceRecoveryMigration: { operationId: randomUUID() } } }).where(eq(executionWorkspaces.id, f.workspaceId)); + await expect(resumeNativeWorkspaceSync({ db, runId: f.runId, target: { kind: "remote", transport: "sandbox", remoteCwd: f.reference.remoteCwd } })) + .rejects.toThrow("legacy_workspace_migration_receipt_invalid"); + }); 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 }); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 91cdc407c22..aed67144fb3 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -17,7 +17,9 @@ 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, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js"; +import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; +import { isLegacyWorkspaceTaskRelated, LEGACY_WORKSPACE_RECOVERY_MIGRATION, type LegacyWorkspaceRecoverySource } from "./legacy-workspace-recovery-migration.js"; +import { offlineWorkspaceMigrationBlockers, prepareNestedWorkspaceImports, validateImportedWorkspaceChildren, workspaceMigrationLayoutBlockers } from "./remote-workspace-migration-tree.js"; const SCHEMA = "paperclip.remote-workspace-migration.v1"; type Workspace = typeof executionWorkspaces.$inferSelect; @@ -46,7 +48,8 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri const marker = object(workspace.metadata?.remoteWorkspaceMigration); const allowedOperation = input.operationId ?? (marker.schema === SCHEMA && typeof marker.operationId === "string" ? marker.operationId : undefined); const reasons: string[] = []; - if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null }; + if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] }; + if (input.writersStopped) reasons.push(...await offlineWorkspaceMigrationBlockers(db)); 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"); if (!workspace.sourceIssueId) reasons.push("task_binding_required"); @@ -96,11 +99,25 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri or ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'workspaceId' = ${workspace.id})`, )).limit(1); if (run) reasons.push("active_execution"); - const [unexported] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, workspace.companyId), + const unexported = await db.select({ id: heartbeatRuns.id, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + issueId: heartbeatRuns.nativeIssueId, controllerLeaseExpiresAt: heartbeatRuns.controllerLeaseExpiresAt, + reference: sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'`, + }).from(heartbeatRuns).where(and(eq(heartbeatRuns.companyId, workspace.companyId), sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'workspaceId' = ${workspace.id}`, sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' = 'prepared'`, - )).limit(1); - if (unexported) reasons.push("remote_copyback_unresolved"); + )); + const legacyRecoverySources: LegacyWorkspaceRecoverySource[] = []; + for (const run of unexported.sort((a, b) => a.id.localeCompare(b.id))) { + const reference = readNativeWorkspaceSyncReference(run.reference); + const [owner] = await db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner }) + .from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.runId, run.id))); + if (!input.writersStopped || !reference || run.runtimeMode !== "native" || !run.issueId + || !await isLegacyWorkspaceTaskRelated(db, { companyId: workspace.companyId, issueId: run.issueId, sourceIssueId: workspace.sourceIssueId }) + || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(run.status) + || run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date() + || !owner || owner.leaseOwner || !["committed", "terminal_failure"].includes(owner.phase)) reasons.push("remote_copyback_unresolved"); + else legacyRecoverySources.push({ runId: run.id, issueId: run.issueId, reference }); + } const [finalization] = workspace.sourceIssueId ? await db.select({ id: nativeRunFinalizations.runId }).from(nativeRunFinalizations).where(and( eq(nativeRunFinalizations.companyId, workspace.companyId), eq(nativeRunFinalizations.issueId, workspace.sourceIssueId), sql`${nativeRunFinalizations.phase} not in ('committed', 'terminal_failure')`, @@ -129,11 +146,14 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri ) : undefined), )).limit(1); if (service) reasons.push("workspace_runtime_service_unsettled"); - const [recovery] = workspace.sourceIssueId ? await db.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( - eq(issueRecoveryActions.companyId, workspace.companyId), eq(issueRecoveryActions.sourceIssueId, workspace.sourceIssueId), + const recoveryIssueIds = [...new Set([workspace.sourceIssueId, ...legacyRecoverySources.map(source => source.issueId)].filter((id): id is string => Boolean(id)))]; + const recoveries = recoveryIssueIds.length ? await db.select({ id: issueRecoveryActions.id, evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, workspace.companyId), inArray(issueRecoveryActions.sourceIssueId, recoveryIssueIds), inArray(issueRecoveryActions.status, ["active", "escalated"]), - )).limit(1) : []; - if (recovery) reasons.push("workspace_recovery_unresolved"); + )) : []; + if (recoveries.some(recovery => !input.writersStopped + || !legacyRecoverySources.some(source => source.runId === recovery.evidence?.runId) + || (recovery.evidence?.systemRecovery as { state?: unknown } | undefined)?.state === "running")) reasons.push("workspace_recovery_unresolved"); const repositories = await db.select().from(executionWorkspaceRepositories).where(and(eq(executionWorkspaceRepositories.companyId, workspace.companyId), eq(executionWorkspaceRepositories.executionWorkspaceId, workspace.id))); if (repositories.some(repository => repository.state !== "ready")) reasons.push("repository_preparation_unresolved"); @@ -141,7 +161,7 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri ref: archivedGitSource?.commit ?? workspace.baseRef, branchName: workspace.branchName } : { kind: "empty" }; if (repositories.length) source.repositories = repositories.map(repository => ({ id: repository.id, relativePath: repository.relativePath, repoUrl: repository.repoUrl, ref: repository.requestedRef, ...(repository.pinnedCommit ? { pinnedCommit: repository.pinnedCommit } : {}) })); - return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource }; + return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource, legacyRecoverySources }; } export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string; writersStopped?: boolean }) { @@ -164,7 +184,7 @@ export async function migrateRemoteWorkspace(input: { if (!workspace) throw new Error("remote_workspace_migration_workspace_missing"); if (workspace.sourceIssueId) await tx.select({ id: issues.id }).from(issues).where(and(eq(issues.companyId, workspace.companyId), eq(issues.id, workspace.sourceIssueId))).for("update"); const current = await inspect(tx as unknown as Db, input); - if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null }; + if (current.authority) return { workspace, authority: current.authority, operation: null, source: current.source, sourceFingerprint: "", issueBinding: current.issueBinding, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] }; if (current.reasons.length) throw new Error(`remote_workspace_migration_blocked:${current.reasons.join(",")}`); const source = remoteWorkspaceSourceSchema.parse(input.source ?? current.source); const sourceFingerprint = fingerprint(workspace); @@ -181,12 +201,14 @@ export async function migrateRemoteWorkspace(input: { 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"); - return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource }; + if (canonical(operation.metadata?.legacyRecoverySources ?? []) !== canonical(current.legacyRecoverySources)) 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 }; } 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, nestedWorkspaceScope: { version: 1, state: "pending" }, + ...(current.legacyRecoverySources.length ? { legacyRecoverySources: current.legacyRecoverySources } : {}), ...(current.archivedGitSource ? { archivedGitSource: gitSourceProof(current.archivedGitSource) } : {}), ...(current.retainedDirectory ? { retainedDirectory: { restorePointId: current.retainedDirectory.restorePoint.id, generation: current.retainedDirectory.generation, digest: current.retainedDirectory.restorePoint.digest } } : {}) } }).returning(); @@ -194,7 +216,7 @@ export async function migrateRemoteWorkspace(input: { 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 }; + return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources }; }); if (reserved.authority) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: true, replayed: true }; let operation = reserved.operation!; @@ -245,6 +267,8 @@ export async function migrateRemoteWorkspace(input: { input.signal?.throwIfAborted(); await lock.execute(sql`select 1`); const publication = await input.db.transaction(async tx => { + for (const source of reserved.legacyRecoverySources) await tx.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, source.runId))).for("update"); 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); @@ -268,9 +292,15 @@ export async function migrateRemoteWorkspace(input: { 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.legacyRecoverySources) !== canonical(reserved.legacyRecoverySources)) throw new Error("remote_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 (reserved.legacyRecoverySources.length) { + metadata.legacyWorkspaceRecoveryMigration = { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: operation.id }; + 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 } : {}) }, diff --git a/server/src/services/workspace-recovery-source.ts b/server/src/services/workspace-recovery-source.ts index ad65d438d2b..87b27c4409d 100644 --- a/server/src/services/workspace-recovery-source.ts +++ b/server/src/services/workspace-recovery-source.ts @@ -1,3 +1,4 @@ +import { hasMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; import { retiredReusableLeasePredecessors } from "./environment-lease-handoff.js"; import { and, eq, inArray, ne, or, sql } from "drizzle-orm"; import { environmentLeases, environments, executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; @@ -211,7 +212,10 @@ export async function inspectWorkspaceRecoverySource(db: Db, input: WorkspaceRec } const [environment] = await db.select().from(environments).where(eq(environments.id, lease.environmentId)); if (!environment) throw new SystemRecoveryFailure("source_unavailable", "service_required"); - sources.push({ lease, environment, reference: legacyDescriptor, remoteReference: remoteDescriptor, + const preservationOnly = legacyDescriptor ? await hasMigratedLegacyWorkspaceSource(db, { + companyId: input.companyId, runId: run.id, reference: legacyDescriptor, + }) : false; + sources.push({ lease, environment, reference: legacyDescriptor, preservationOnly, remoteReference: remoteDescriptor, remoteWorkspace: remoteDescriptor ? remoteWorkspace : null, remoteCwd }); } return { run, action, sources }; diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 3768ae1528b..182be104090 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/server/src/services/workspace-recovery.test.ts @@ -33,6 +33,7 @@ import { GitShallowMetadataError } from "@paperclipai/adapter-utils/git-workspac import { readSystemRecovery } from "./system-recovery-state.js"; import { adoptSystemRecovery, updateSystemRecovery } from "./system-recovery.js"; import { getStorageService } from "../storage/index.js"; +import { migrateRemoteWorkspace } from "./remote-workspace-migration.js"; import { attachPreservedWorkspaceSnapshots } from "./workspace-recovery-snapshot.js"; import { adoptHistoricalWorkspaceRecoveries, historicalRecoveryCounts } from "./workspace-recovery-backfill.js"; import { createSystemRecovery, failSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; @@ -194,6 +195,43 @@ describe("automatic retained workspace recovery", () => { const action = async (id: string) => (await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, id)))[0]!; const run = async (id: string) => (await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, id)))[0]!; + it.each([false, true])("preserves an old sandbox after storage migration without writing its old host (removed=%s)", async removeHost => { + const f = await seed(); + // The fixture's command runner supports native copyback; the provider + // snapshot boundary separately needs its real-file syncOut capability. + targets.get(f.lease.id)!.runner!.syncIn = async operations => { + for (const operation of operations) for (const file of operation.files) + await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); + return { filesTransferred: 2, bytesTransferred: 0, operations: [] }; + }; + targets.get(f.lease.id)!.runner!.syncOut = async operations => { + for (const operation of operations) for (const file of operation.files) + await cp(file.sourcePath, file.targetPath, { recursive: true, dereference: false }); + return { filesTransferred: 2, bytesTransferred: 0, operations: [] }; + }; + await db.insert(executionWorkspaces).values({ id: f.issueId, companyId: f.companyId, sourceIssueId: f.issueId, + name: "Existing files", mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd: f.host }); + await db.update(issues).set({ executionWorkspaceId: f.issueId }).where(eq(issues.id, f.issueId)); + vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + try { + await migrateRemoteWorkspace({ db, storage: getStorageService(), companyId: f.companyId, + workspaceId: f.issueId, apply: true, writersStopped: true }); + if (removeHost) await rm(f.host, { recursive: true }); + await requestWorkspaceRecovery({ ...f, idempotencyKey: randomUUID(), actorId: "board" }); + await recoverRetainedWorkspace(f.input); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + expect(await readFile(path.join(f.host, "saved.txt")).catch(() => null)).toBeNull(); + expect(await run(f.runId)).toMatchObject({ status: "failed", runnerProfileJson: { nativeWorkspaceSync: { state: "prepared" } } }); + const restored = path.join(root, randomUUID()); await mkdir(restored); + const copies = await attachPreservedWorkspaceSnapshots({ db, companyId: f.companyId, issueId: f.issueId, + sourceRunIds: [f.runId], cwd: restored }); + expect(copies).toHaveLength(1); + expect(await readFile(path.join(copies[0]!.path, "files", "saved.txt"), "utf8")).toBe("finished work\n"); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.companyId, f.companyId))).toHaveLength(1); + expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0); + } finally { vi.unstubAllEnvs(); } + }, 30_000); + function boardRecoveryApp() { const app = express(); app.use(express.json()); diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index 83a1b1cc674..b9a1bc50cdc 100644 --- a/server/src/services/workspace-recovery.ts +++ b/server/src/services/workspace-recovery.ts @@ -206,7 +206,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ continue; } } - let reference = source.reference; + let reference = source.preservationOnly ? null : source.reference; if (reference) { try { await readNativeWorkspaceSyncLocalCwd({ runId: input.runId, companyId: input.companyId, workspaceId: reference.workspaceId, reference }); }