From fc514283fd446150038112169b98b88fb7a2f7ae Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 4 Nov 2026 21:54:17 -0600 Subject: [PATCH] fix(workspaces): preserve finalized recovery holds during offline import Co-Authored-By: Paperclip --- ...lized-workspace-recovery-migration.test.ts | 174 ++++++++++++++++++ .../finalized-workspace-recovery-migration.ts | 101 ++++++++++ .../legacy-workspace-recovery-migration.ts | 23 ++- .../services/remote-workspace-migration.ts | 58 ++++-- .../services/workspace-recovery-receipt.ts | 16 +- 5 files changed, 348 insertions(+), 24 deletions(-) create mode 100644 server/src/services/finalized-workspace-recovery-migration.test.ts create mode 100644 server/src/services/finalized-workspace-recovery-migration.ts diff --git a/server/src/services/finalized-workspace-recovery-migration.test.ts b/server/src/services/finalized-workspace-recovery-migration.test.ts new file mode 100644 index 00000000000..b799183aa70 --- /dev/null +++ b/server/src/services/finalized-workspace-recovery-migration.test.ts @@ -0,0 +1,174 @@ +import { randomUUID } from "node:crypto"; +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { eq } from "drizzle-orm"; +import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; +import { agents, closeRegisteredClients, companies, completionContracts, createDb, environmentLeases, environments, + executionWorkspaces, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults, workspaceOperations } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../__tests__/helpers/embedded-postgres.js"; +import { createStorageService } from "../storage/service.js"; +import { createLocalDiskStorageProvider } from "../storage/local-disk-provider.js"; +import { inventoryRemoteWorkspaceMigration, migrateRemoteWorkspace } from "./remote-workspace-migration.js"; +import { inspectWorkspaceRecoverySource } from "./workspace-recovery-source.js"; +import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; +import { verifiedWorkspaceRepairReceipt } from "./workspace-recovery-receipt.js"; +import { createSystemRecovery } from "./system-recovery-state.js"; +import { attemptNativeWorkspaceFinalization } from "./native-runtime/native-workspace-finalizer.js"; +import { getWorkspaceSnapshot, loadWorkspaceSnapshotArchive } from "./remote-workspace-snapshot-storage.js"; +import { restoreRemoteWorkspaceArchive } from "./remote-workspace-snapshot-archive.js"; + +const support = await getEmbeddedPostgresTestSupport(); +describe.skipIf(!support.supported)("finalized workspace recovery preservation", () => { + let database: Awaited>, db: ReturnType, root: string; + beforeAll(async () => { + database = await startEmbeddedPostgresTestDatabase("finalized-recovery-migration-"); db = createDb(database.connectionString); + root = await fs.mkdtemp(path.join(os.tmpdir(), "finalized-recovery-migration-")); + vi.stubEnv("PAPERCLIP_SECRETS_MASTER_KEY", "63".repeat(32)); vi.stubEnv("PAPERCLIP_EPHEMERAL_MIGRATION", "true"); + }, 90_000); + afterAll(async () => { + if (database) await closeRegisteredClients(database.connectionString); + await database?.cleanup(); if (root) await fs.rm(root, { recursive: true, force: true }); vi.unstubAllEnvs(); + }); + async function fixture(files = false) { + const companyId = randomUUID(), workspaceId = randomUUID(), runId = randomUUID(), cwd = path.join(root, workspaceId); + await fs.mkdir(cwd); await fs.writeFile(path.join(cwd, "saved.txt"), "original held files\n"); + await db.insert(companies).values({ id: companyId, name: "Finalized recovery", issuePrefix: `F${companyId.slice(0, 6)}` }); + const [agent] = await db.insert(agents).values({ companyId, name: "Historical worker", adapterType: "paperclip_runner" }).returning(); + const [issue] = await db.insert(issues).values({ companyId, title: "Held task", status: "blocked" }).returning(); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId, sourceIssueId: issue.id, + mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd, name: "Held files" }); + await db.update(issues).set({ executionWorkspaceId: workspaceId }).where(eq(issues.id, issue.id)); + const reference = { schema: "paperclip.native-workspace-sync/v3" as const, state: "finalized" as const, workspaceId, + leaseId: randomUUID(), providerLeaseId: randomUUID(), remoteCwd: "/workspaces/task", descriptorSha256: "12".repeat(32), + baselineSha256: "34".repeat(32), finalHostSha256: "56".repeat(32), resourceDisposition: "destroy" as const }; + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId: agent.id, nativeIssueId: issue.id, + status: "failed", runtimeMode: "native", runnerProfileJson: { nativeWorkspaceSync: reference }, resultJson: { retained: "result" } }); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: issue.id, runId, phase: "terminal_failure" }); + if (files) { + const [environment] = await db.insert(environments).values({ name: `Historical environment ${workspaceId}`, driver: "daytona" }).returning(); + await db.insert(environmentLeases).values({ id: reference.leaseId, companyId, environmentId: environment.id, + issueId: issue.id, heartbeatRunId: runId, executionWorkspaceId: workspaceId, provider: "daytona", + providerLeaseId: reference.providerLeaseId, leasePolicy: "ephemeral", status: "expired", metadata: { remoteCwd: reference.remoteCwd } }); + } + const [action] = await db.insert(issueRecoveryActions).values({ companyId, sourceIssueId: issue.id, + kind: "active_run_watchdog", cause: "native_continuation_requires_reconciliation", fingerprint: randomUUID(), + nextAction: "Preserve task hold", evidence: { runId, nativeFailureBlock: { retained: "hold" }, ...(files ? { + systemRecovery: { ...createSystemRecovery(runId), state: "exhausted" }, + workspaceRecovery: { schema: "paperclip.workspace-recovery.v1", runId, status: "blocked", stage: "verifying_stop", attempt: 4, requestIds: [] }, + historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId, leaseIds: [reference.leaseId] }, + } : {}) } }).returning(); + const storage = createStorageService(createLocalDiskStorageProvider(path.join(root, `${workspaceId}-objects`))); + return { companyId, workspaceId, runId, issueId: issue.id, cwd, reference, action, storage }; + } + async function preserved(f: Awaited>) { + const [issue] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)); + const [owner] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, f.action.id)); + return { issue, run, owner, action }; + } + it.each([false, true])("saves held files while preserving task/coordinator/result bytes (file source: %s)", async files => { + const f = await fixture(files), before = await preserved(f); + await expect(migrateRemoteWorkspace({ db, ...f, writersStopped: false, apply: true })).rejects.toThrow("workspace_recovery_unresolved"); + await migrateRemoteWorkspace({ db, ...f, writersStopped: true, apply: true }); + const after = await preserved(f); + expect(after.issue).toEqual(before.issue); expect(after.owner).toEqual(before.owner); expect(after.action).toEqual(before.action); + if (files) { + expect(after.run).toEqual({ ...before.run, updatedAt: expect.any(Date), runnerProfileJson: { ...before.run.runnerProfileJson, + finalizedWorkspaceRecoveryMigration: { workspaceId: f.workspaceId, operationId: expect.any(String), schema: "paperclip.legacy-workspace-recovery-migration.v1" } } }); + const inspected = await inspectWorkspaceRecoverySource(db, { ...f, actionId: f.action.id }); + expect(inspected.sources).toEqual([expect.objectContaining({ preservationOnly: true, reference: f.reference })]); + expect(inspected.sources[0].lease.metadata).toEqual({ remoteCwd: f.reference.remoteCwd }); + } else expect(after.run).toEqual(before.run); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + const authority = readRemoteWorkspaceAuthority(workspace.metadata)!; + const ref = (await getWorkspaceSnapshot(db, { ...f, snapshotId: authority.snapshotId! }))!; + await fs.rm(f.cwd, { recursive: true }); + const archivePath = path.join(root, `${f.workspaceId}.pcws`), targetDir = path.join(root, `${f.workspaceId}-restored`); + await loadWorkspaceSnapshotArchive({ storage: f.storage, reference: ref, targetPath: archivePath }); + await restoreRemoteWorkspaceArchive({ archivePath, targetDir, expectedManifestDigest: ref.manifestDigest }); + expect(await fs.readFile(path.join(targetDir, "saved.txt"), "utf8")).toBe("original held files\n"); + }); + it("retains accepted-result finalization retries with a legacy receipt that has no workspace FK", async () => { + const f = await fixture(), contractId = randomUUID(), resultId = randomUUID(), operationId = randomUUID(); + await db.insert(completionContracts).values({ id: contractId, companyId: f.companyId, issueId: f.issueId, revision: 1, + schemaVersion: "v1", policyVersion: "v1", risk: "low", completionAuthority: "agent", incompleteCriteriaPolicy: "block", + contractJson: {}, canonicalSha256: "12".repeat(32), createdByActorType: "system", createdByActorId: "test" }); + await db.update(heartbeatRuns).set({ status: "succeeded", completionContractId: contractId }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(nativeRunResults).values({ id: resultId, companyId: f.companyId, issueId: f.issueId, runId: f.runId, + completionContractId: contractId, serverFingerprint: "saved-result", schemaStatus: "accepted", resultJson: { saved: true }, canonicalSha256: "34".repeat(32) }); + await db.update(nativeRunFinalizations).set({ phase: "retryable_failure", resultId }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.insert(workspaceOperations).values({ id: operationId, companyId: f.companyId, executionWorkspaceId: null, + issueId: f.issueId, heartbeatRunId: f.runId, phase: "workspace_finalize", status: "succeeded" }); + const before = await preserved(f); + await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true }); + expect(await preserved(f)).toEqual(before); + expect(await attemptNativeWorkspaceFinalization({ db, runId: f.runId })).toMatchObject({ kind: "completed", operation: { id: operationId } }); + }); + it.each(["source-run", "controller", "owner", "phase", "descriptor", "file-marker", "allocation", "running-recovery", "unknown-kind"])("keeps an unproven %s hold blocked", async mode => { + const f = await fixture(true); + if (mode === "source-run") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, runId: randomUUID() } }).where(eq(issueRecoveryActions.id, f.action.id)); + if (mode === "controller") await db.update(heartbeatRuns).set({ controllerLeaseExpiresAt: new Date(Date.now() + 60_000) }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "owner") await db.update(nativeRunFinalizations).set({ leaseOwner: "expired-owner", leaseExpiresAt: new Date(0) }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "phase") await db.update(nativeRunFinalizations).set({ phase: "retryable_failure" }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (mode === "descriptor") await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { ...f.reference, finalHostSha256: null } } }).where(eq(heartbeatRuns.id, f.runId)); + if (mode === "file-marker") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, historicalWorkspaceSource: { schema: "unverified" } } }).where(eq(issueRecoveryActions.id, f.action.id)); + if (mode === "allocation") await db.update(environmentLeases).set({ heartbeatRunId: null }).where(eq(environmentLeases.id, f.reference.leaseId)); + if (mode === "running-recovery") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, + systemRecovery: { ...createSystemRecovery(f.runId), state: "running" } } }).where(eq(issueRecoveryActions.id, f.action.id)); + if (mode === "unknown-kind") await db.update(issueRecoveryActions).set({ kind: "unknown_kind" }).where(eq(issueRecoveryActions.id, f.action.id)); + expect(await inventoryRemoteWorkspaceMigration(db, { ...f, writersStopped: true })).toMatchObject({ eligible: false, reasons: expect.arrayContaining(["workspace_recovery_unresolved"]) }); + }); + it.each(["action", "allocation", "coordinator"])("rechecks the exact %s after upload", async mode => { + const f = await fixture(true), put = f.storage.putFile.bind(f.storage); + f.storage.putFile = async input => { const result = await put(input); + if (mode === "action") await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, retainedChange: true } }).where(eq(issueRecoveryActions.id, f.action.id)); + if (mode === "allocation") await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.reference.leaseId)); + if (mode === "coordinator") await db.update(nativeRunFinalizations).set({ failureCode: "changed" }).where(eq(nativeRunFinalizations.runId, f.runId)); + return result; + }; + await expect(migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true })).rejects.toThrow(); + const [workspace] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, f.workspaceId)); + expect(readRemoteWorkspaceAuthority(workspace.metadata)).toBeNull(); + }); + it("keeps a historical folder separate from its real recovery source", async () => { + const f = await fixture(true), workspaceId = randomUUID(), cwd = path.join(root, workspaceId); await fs.mkdir(cwd); + await fs.writeFile(path.join(cwd, "history.txt"), "historical files\n"); + await db.insert(executionWorkspaces).values({ id: workspaceId, companyId: f.companyId, sourceIssueId: f.issueId, + mode: "isolated_workspace", strategyType: "project_primary", providerType: "local_fs", cwd, name: "Other historical folder" }); + await migrateRemoteWorkspace({ db, ...f, workspaceId, apply: true, writersStopped: true }); + const [historical] = await db.select().from(executionWorkspaces).where(eq(executionWorkspaces.id, workspaceId)); + expect(historical.metadata).not.toHaveProperty("legacyWorkspaceRecoveryMigration"); + expect((await preserved(f)).run.runnerProfileJson).not.toHaveProperty("finalizedWorkspaceRecoveryMigration"); + await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true }); + expect(await readMigratedLegacyWorkspaceSource(db, { ...f, reference: f.reference })).toMatchObject({ reference: f.reference }); + }); + it.each(["old-copyback", "new-snapshot"])("retains verified repair through migration (%s)", async mode => { + const f = await fixture(true), operationId = randomUUID(); + await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, + systemRecovery: { ...createSystemRecovery(f.runId), state: "complete" }, + workspaceRecovery: { ...(f.action.evidence.workspaceRecovery as object), status: "recovered" }, + } }).where(eq(issueRecoveryActions.id, f.action.id)); + const file = (suffix: string) => ({ objectKey: `${f.companyId}/workspace-recovery/${f.runId}/${f.reference.leaseId}/${"78".repeat(32)}.${suffix}`, + sha256: "78".repeat(32), byteSize: 1 }); + await db.insert(workspaceOperations).values({ id: operationId, companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, phase: mode === "old-copyback" ? "workspace_recovery" : "workspace_recovery_snapshot", status: "succeeded", + metadata: { schema: "paperclip.workspace-recovery-snapshot.v1", recoveryActionId: f.action.id, + leaseId: f.reference.leaseId, providerLeaseId: f.reference.providerLeaseId, + descriptorSha256: f.reference.descriptorSha256, finalHostSha256: f.reference.finalHostSha256, + archive: file("tar"), manifest: file("jsonl") } }); + await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true }); + expect(await verifiedWorkspaceRepairReceipt(db, (await preserved(f)).run)).toBe(operationId); + }); + it("requires the exact receipt and leaves ordinary finalized references alone", async () => { + const f = await fixture(true); + expect(await readMigratedLegacyWorkspaceSource(db, { ...f, reference: f.reference })).toBeNull(); + await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true }); + await expect(readMigratedLegacyWorkspaceSource(db, { ...f, reference: { ...f.reference, descriptorSha256: "78".repeat(32) } })) + .rejects.toThrow("legacy_workspace_migration_receipt_invalid"); + await db.update(environmentLeases).set({ providerLeaseId: randomUUID() }).where(eq(environmentLeases.id, f.reference.leaseId)); + await expect(readMigratedLegacyWorkspaceSource(db, { ...f, reference: f.reference })).rejects.toThrow("legacy_workspace_migration_receipt_invalid"); + }); +}); diff --git a/server/src/services/finalized-workspace-recovery-migration.ts b/server/src/services/finalized-workspace-recovery-migration.ts new file mode 100644 index 00000000000..f01e555063a --- /dev/null +++ b/server/src/services/finalized-workspace-recovery-migration.ts @@ -0,0 +1,101 @@ +import { createHash } from "node:crypto"; +import { isDeepStrictEqual } from "node:util"; +import { and, eq } from "drizzle-orm"; +import { executionWorkspaces, heartbeatRuns, issueRecoveryActions, nativeRunFinalizations, nativeRunResults, + workspaceOperations, type Db } from "@paperclipai/db"; +import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; +import { adapterExecutionControls } from "./adapter-execution-control.js"; +import { isLegacyWorkspaceTaskRelated, type LegacyWorkspaceRecoverySource } from "./legacy-workspace-recovery-migration.js"; +import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; +import { readSystemRecovery, SystemRecoveryFailure } from "./system-recovery-state.js"; +import { readWorkspaceRecovery } from "./workspace-recovery-receipt.js"; +import { inspectWorkspaceRecoverySource, terminalWorkspaceRunStates } from "./workspace-recovery-source.js"; + +type ReadDb = Pick; +type Recovery = Pick; +export type FinalizedWorkspaceRecoveryHold = { + actionId: string; runId: string; sourceWorkspaceId: string; fingerprint: string; + source: LegacyWorkspaceRecoverySource | null; +}; + +function canonical(value: unknown): string { + if (value instanceof Date) return JSON.stringify(value.toISOString()); + if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`; + if (value && typeof value === "object") return `{${Object.entries(value).sort(([a], [b]) => a.localeCompare(b)) + .map(([key, item]) => `${JSON.stringify(key)}:${canonical(item)}`).join(",")}}`; + return JSON.stringify(value) ?? "null"; +} + +/** A task hold can outlive completed file finalization. Preserve that hold and + * its exact source; terminal run status alone never releases a file authority. + * This is offline admission, not stop verification or recovery completion. */ +export async function inspectFinalizedWorkspaceRecoveryHold(db: ReadDb, input: { + companyId: string; workspaceId: string; recovery: Recovery; +}): Promise { + const action = input.recovery; + if (action.kind !== "active_run_watchdog" || typeof action.evidence.runId !== "string" + || !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(action.evidence.runId)) return null; + const [run] = await db.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, action.evidence.runId))); + const reference = readNativeWorkspaceSyncReference(run?.runnerProfileJson?.nativeWorkspaceSync); + if (!run || run.runtimeMode !== "native" || run.nativeIssueId !== action.sourceIssueId + || !terminalWorkspaceRunStates.includes(run.status) || !reference || reference.state !== "finalized" + || !reference.finalHostSha256 || run.runnerProfileJson?.remoteWorkspace != null + || adapterExecutionControls.has(run.id) || run.controllerLeaseExpiresAt && run.controllerLeaseExpiresAt > new Date()) return null; + const [owner] = await db.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, run.id), + eq(nativeRunFinalizations.issueId, action.sourceIssueId))); + if (!owner || owner.leaseOwner) return null; + const [workspace] = await db.select().from(executionWorkspaces).where(and( + eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, reference.workspaceId))); + if (!workspace || !await isLegacyWorkspaceTaskRelated(db, { companyId: input.companyId, + issueId: run.nativeIssueId, sourceIssueId: workspace.sourceIssueId })) return null; + // Historical folders never borrow the source receipt. Their actual source + // remains a separate local record (or its already-published remote authority). + if (reference.workspaceId !== input.workspaceId && !readRemoteWorkspaceAuthority(workspace.metadata) + && (!workspace.cwd || !["local_fs", "git_worktree"].includes(workspace.providerType))) return null; + let finalizeReceipt: string | null = null; + if (!["committed", "terminal_failure"].includes(owner.phase)) { + if (owner.phase !== "retryable_failure" || !owner.resultId || !run.completionContractId) return null; + const [result] = await db.select({ id: nativeRunResults.id }).from(nativeRunResults).where(and( + eq(nativeRunResults.companyId, input.companyId), eq(nativeRunResults.issueId, action.sourceIssueId), + eq(nativeRunResults.runId, run.id), eq(nativeRunResults.id, owner.resultId), + eq(nativeRunResults.completionContractId, run.completionContractId), eq(nativeRunResults.schemaStatus, "accepted"))); + // Historical finalization receipts may have no workspace FK. The owning + // finalizer consumes this exact company/run/task receipt before inspecting + // workspace authority; the saved run reference supplies the workspace. + const [receipt] = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and( + eq(workspaceOperations.companyId, input.companyId), eq(workspaceOperations.issueId, action.sourceIssueId), + eq(workspaceOperations.heartbeatRunId, run.id), eq(workspaceOperations.phase, "workspace_finalize"), + eq(workspaceOperations.status, "succeeded"))); + if (!result || !receipt) return null; + finalizeReceipt = receipt.id; + } + const fileKeys = ["systemRecovery", "workspaceRecovery", "workspaceRestoreRecovery", "historicalWorkspaceSource"]; + let source: LegacyWorkspaceRecoverySource | null = null; + if (fileKeys.some(key => action.evidence[key] != null)) { + const system = readSystemRecovery(action.evidence), recovery = readWorkspaceRecovery(action.evidence); + if (!system || system.sourceRunId !== run.id || system.state === "running" || !recovery || recovery.runId !== run.id + || ["queued", "recovering"].includes(recovery.status) + || action.evidence.workspaceRestoreRecovery != null + && !isDeepStrictEqual(action.evidence.workspaceRestoreRecovery, run.resultJson?.workspaceRestoreRecovery)) return null; + try { + const inspected = await inspectWorkspaceRecoverySource(db as Db, { + companyId: input.companyId, issueId: action.sourceIssueId, runId: run.id, actionId: action.id, + }); + source = { runId: run.id, issueId: action.sourceIssueId, reference, + recoveryAllocations: inspected.sources.map(item => ({ leaseId: item.lease.id, + providerLeaseId: item.lease.providerLeaseId!, remoteCwd: item.remoteCwd })).sort((a, b) => a.leaseId.localeCompare(b.leaseId)) }; + } catch (error) { + if (error instanceof SystemRecoveryFailure) return null; + throw error; + } + } + return { actionId: action.id, runId: run.id, sourceWorkspaceId: reference.workspaceId, + source: reference.workspaceId === input.workspaceId ? source : null, + fingerprint: createHash("sha256").update(canonical({ action, reference, + status: run.status, controllerLeaseExpiresAt: run.controllerLeaseExpiresAt?.toISOString() ?? null, + completionContractId: run.completionContractId, restoreRecovery: run.resultJson?.workspaceRestoreRecovery, + owner, finalizeReceipt, source })).digest("hex") }; +} diff --git a/server/src/services/legacy-workspace-recovery-migration.ts b/server/src/services/legacy-workspace-recovery-migration.ts index 8d014facfe4..63917ddb149 100644 --- a/server/src/services/legacy-workspace-recovery-migration.ts +++ b/server/src/services/legacy-workspace-recovery-migration.ts @@ -63,24 +63,31 @@ export async function readLegacyWorkspaceRecoveryAllocations(db: Pick, input: { companyId: string; runId: string; reference: NativeWorkspaceSyncReference | null; }): Promise { - if (input.reference && input.reference.state !== "prepared") return null; const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, nativeIssueId: heartbeatRuns.nativeIssueId, resultJson: sql>`jsonb_build_object('workspaceRestoreRecovery', ${heartbeatRuns.resultJson}->'workspaceRestoreRecovery')`, runnerProfileJson: sql>`jsonb_build_object( 'legacyWorkspaceRecoveryMigration', ${heartbeatRuns.runnerProfileJson}->'legacyWorkspaceRecoveryMigration', + 'finalizedWorkspaceRecoveryMigration', ${heartbeatRuns.runnerProfileJson}->'finalizedWorkspaceRecoveryMigration', 'nativeWorkspaceSync', ${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync', 'remoteWorkspace', ${heartbeatRuns.runnerProfileJson}->'remoteWorkspace')`, }).from(heartbeatRuns).where(and( eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, input.runId), )); - const runMarker = run?.runnerProfileJson?.legacyWorkspaceRecoveryMigration as { workspaceId?: unknown } | undefined; + const runMarker = run?.runnerProfileJson?.[input.reference?.state === "finalized" + ? "finalizedWorkspaceRecoveryMigration" : "legacyWorkspaceRecoveryMigration"] as { workspaceId?: unknown } | undefined; + if (input.reference?.state === "finalized" && runMarker == null) return null; + if (input.reference?.state === "finalized" && !isDeepStrictEqual(input.reference, run?.runnerProfileJson?.nativeWorkspaceSync)) + throw new Error("legacy_workspace_migration_receipt_invalid"); const workspaceId = input.reference?.workspaceId ?? runMarker?.workspaceId; if (typeof workspaceId !== "string") return null; const [workspace] = await db.select({ metadata: executionWorkspaces.metadata }).from(executionWorkspaces).where(and( eq(executionWorkspaces.companyId, input.companyId), eq(executionWorkspaces.id, workspaceId), )); - if (!readRemoteWorkspaceAuthority(workspace?.metadata)) return null; + if (!readRemoteWorkspaceAuthority(workspace?.metadata)) { + if (input.reference?.state === "finalized" && runMarker != null) throw new Error("legacy_workspace_migration_receipt_invalid"); + return null; + } 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"); @@ -100,6 +107,16 @@ export async function readMigratedLegacyWorkspaceSource(db: Pick, || run?.runnerProfileJson?.remoteWorkspace != null || !isDeepStrictEqual(source.recoveryAllocations ?? null, run ? await readLegacyWorkspaceRecoveryAllocations(db, run) : null))) throw new Error("legacy_workspace_migration_receipt_invalid"); + if (source?.recoveryAllocations) { + for (const allocation of source.recoveryAllocations) { + const [lease] = await db.select({ providerLeaseId: environmentLeases.providerLeaseId, metadata: environmentLeases.metadata }) + .from(environmentLeases).where(and(eq(environmentLeases.companyId, input.companyId), + eq(environmentLeases.id, allocation.leaseId), eq(environmentLeases.heartbeatRunId, input.runId), + eq(environmentLeases.issueId, source.issueId))); + if (!lease || lease.providerLeaseId !== allocation.providerLeaseId || lease.metadata?.remoteCwd !== allocation.remoteCwd) + throw new Error("legacy_workspace_migration_receipt_invalid"); + } + } if (!source) throw new Error("legacy_workspace_migration_receipt_invalid"); return source as LegacyWorkspaceRecoverySource; } diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index a2f579466c8..1c01f88f83a 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/server/src/services/remote-workspace-migration.ts @@ -9,6 +9,7 @@ import { } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority, remoteWorkspaceSourceSchema, type RemoteWorkspaceAuthority, type RemoteWorkspaceSource } from "@paperclipai/shared"; import { offlineMigrationExecutionWhere } from "./offline-migration-execution.js"; +import { inspectFinalizedWorkspaceRecoveryHold, type FinalizedWorkspaceRecoveryHold } from "./finalized-workspace-recovery-migration.js"; import { workspaceExportCauses } from "./workspace-recovery-source.js"; import type { StorageService } from "../storage/types.js"; import { persistActivity, publishActivity } from "./activity-log.js"; @@ -58,7 +59,7 @@ 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, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[] }; + if (authority) return { workspace, reasons, authority, marker, source: authority.source, issueBinding: null, retainedDirectory: null, archivedGitSource: null, legacyRecoverySources: [] as LegacyWorkspaceRecoverySource[], finalizedRecoveryHolds: [] as FinalizedWorkspaceRecoveryHold[] }; 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"); @@ -203,22 +204,34 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri )).limit(1); if (service) reasons.push("workspace_runtime_service_unsettled"); 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, kind: issueRecoveryActions.kind, + const recoveries = recoveryIssueIds.length ? await db.select({ id: issueRecoveryActions.id, sourceIssueId: issueRecoveryActions.sourceIssueId, + status: issueRecoveryActions.status, updatedAt: issueRecoveryActions.updatedAt, kind: issueRecoveryActions.kind, cause: issueRecoveryActions.cause, evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( eq(issueRecoveryActions.companyId, workspace.companyId), inArray(issueRecoveryActions.sourceIssueId, recoveryIssueIds), inArray(issueRecoveryActions.status, ["active", "escalated"]), )) : []; - if (recoveries.some(recovery => { - if (!input.writersStopped || object(recovery.evidence?.systemRecovery).state === "running") return true; - if (legacyRecoverySources.some(source => source.runId === recovery.evidence?.runId)) return false; + const finalizedRecoveryHolds: FinalizedWorkspaceRecoveryHold[] = []; + for (const recovery of recoveries.sort((a, b) => a.id.localeCompare(b.id))) { + if (!input.writersStopped || object(recovery.evidence?.systemRecovery).state === "running") { + reasons.push("workspace_recovery_unresolved"); continue; + } + if (legacyRecoverySources.some(source => source.runId === recovery.evidence?.runId && source.reference?.state !== "finalized")) continue; + const held = await inspectFinalizedWorkspaceRecoveryHold(db, { companyId: workspace.companyId, + workspaceId: workspace.id, recovery }); + if (held) { + finalizedRecoveryHolds.push(held); + if (held.source && !legacyRecoverySources.some(source => source.runId === held.source!.runId)) legacyRecoverySources.push(held.source); + continue; + } // These warnings govern task disposition, not file authority. Keep them // unchanged while the offline owner saves the workspace. Unknown recovery // kinds and any declared file source still require explicit preservation. - return !["stranded_assigned_issue", "configuration_validation", "missing_disposition", "deliberate_wait_without_target"].includes(recovery.kind) + if (!["stranded_assigned_issue", "configuration_validation", "missing_disposition", "deliberate_wait_without_target"].includes(recovery.kind) || workspaceExportCauses.includes(recovery.cause) || ["workspaceRestoreRecovery", "workspaceRecovery", "historicalWorkspaceSource", "systemRecovery", "nativeFailureBlock"] - .some(key => recovery.evidence?.[key] != null); - })) reasons.push("workspace_recovery_unresolved"); + .some(key => recovery.evidence?.[key] != null)) reasons.push("workspace_recovery_unresolved"); + } + legacyRecoverySources.sort((a, b) => a.runId.localeCompare(b.runId)); 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"); @@ -226,7 +239,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, legacyRecoverySources }; + return { workspace, reasons: [...new Set(reasons)], authority, marker, source, issueBinding: issue?.executionWorkspaceId ?? null, retainedDirectory, archivedGitSource, legacyRecoverySources, finalizedRecoveryHolds }; } export async function inventoryRemoteWorkspaceMigration(db: Db, input: { companyId: string; workspaceId: string; writersStopped?: boolean }) { @@ -253,7 +266,7 @@ export async function migrateRemoteWorkspace(input: { 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); - 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.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[] }; 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); @@ -272,14 +285,16 @@ 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"); + 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"); - return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources }; + return { workspace, authority: null, operation, source, sourceFingerprint, issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds }; } 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.archivedGitSource ? { archivedGitSource: gitSourceProof(current.archivedGitSource) } : {}), ...(current.retainedDirectory ? { retainedDirectory: { restorePointId: current.retainedDirectory.restorePoint.id, @@ -288,7 +303,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, legacyRecoverySources: current.legacyRecoverySources }; + return { workspace, authority: null, operation: operation!, source, sourceFingerprint, issueBinding: current.issueBinding, retainedDirectory: current.retainedDirectory, archivedGitSource: current.archivedGitSource, legacyRecoverySources: current.legacyRecoverySources, finalizedRecoveryHolds: current.finalizedRecoveryHolds }; }); if (reserved.authority) return { ...await inventoryRemoteWorkspaceMigration(input.db, input), applied: true, replayed: true }; let operation = reserved.operation!; @@ -340,11 +355,17 @@ export async function migrateRemoteWorkspace(input: { await lock.execute(sql`select 1`); const publication = await input.db.transaction(async tx => { await lockMigrationParents(tx, input.companyId, reserved.workspace.sourceIssueId); - for (const source of reserved.legacyRecoverySources) { + for (const held of reserved.finalizedRecoveryHolds) { + await tx.select({ id: issueRecoveryActions.id }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.id, held.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, source.runId))).for("update", { noWait: true }); + .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, source.runId))).for("update", { noWait: true }); + .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"); @@ -373,6 +394,7 @@ 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.finalizedRecoveryHolds) !== canonical(reserved.finalizedRecoveryHolds)) throw new Error("remote_workspace_migration_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"); @@ -383,6 +405,10 @@ export async function migrateRemoteWorkspace(input: { if (reserved.legacyRecoverySources.length) { metadata.legacyWorkspaceRecoveryMigration = { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, operationId: operation.id }; for (const source of reserved.legacyRecoverySources) { + // Completed file finalization retains its ordinary retry path. + // Its exact operation receipt redirects only future file recovery. + 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 @@ -398,7 +424,7 @@ export async function migrateRemoteWorkspace(input: { eq(nativeRunFinalizations.runId, source.runId), eq(nativeRunFinalizations.issueId, source.issueId))); await tx.update(heartbeatRuns).set({ runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) || ${JSON.stringify({ - legacyWorkspaceRecoveryMigration: { schema: LEGACY_WORKSPACE_RECOVERY_MIGRATION, + [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))); diff --git a/server/src/services/workspace-recovery-receipt.ts b/server/src/services/workspace-recovery-receipt.ts index 50d3835f11e..01e8ff8861d 100644 --- a/server/src/services/workspace-recovery-receipt.ts +++ b/server/src/services/workspace-recovery-receipt.ts @@ -72,14 +72,15 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick>'{workspaceRecovery,runId}' = ${run.id}`, sql`${issueRecoveryActions.evidence}#>>'{workspaceRecovery,status}' = 'recovered'`, )).orderBy(desc(workspaceOperations.createdAt)).limit(1); - if (!saved || readWorkspaceRecovery(saved.evidence)?.status !== "recovered") return null; + if (!saved || readWorkspaceRecovery(saved.evidence)?.status !== "recovered") { + receiptId = null; break; + } try { readSnapshot(saved.metadata, run.companyId, run.id); } catch { return null; } receiptId = saved.id; } - return receiptId; + if (receiptId) return receiptId; + // A completed legacy copyback is still a valid receipt after its host + // files move. Migration must not erase that independent recovery proof. + if (reference?.state !== "finalized") return null; } } if (typeof issueId !== "string" || (!remote && (!reference || reference.state !== "finalized" || !reference.finalHostSha256))) return null;