diff --git a/doc/DATABASE.md b/doc/DATABASE.md index d680d73e907..b647275bb43 100644 --- a/doc/DATABASE.md +++ b/doc/DATABASE.md @@ -836,6 +836,25 @@ with `NOWAIT`. A real `55P03` aborts the publication transaction and reports archive and local files remain available for a later administrative retry. Other database errors propagate. No automatic retry or lock-strength reduction occurs. +Offline import can preserve an active task recovery hold after its exact native +workspace reference is finalized. It checks the terminal run, released controller +and coordinator ownership, task relationship, and any recorded file allocations. +An accepted result with a retrying coordinator also needs the successful +company/run/task `workspace_finalize` receipt consumed by the normal finalizer. +Historical receipts can have no workspace foreign key. Import retains recovery +rows, task pointers, results and coordinator phases. It does not establish provider +termination or resolve the hold. It locks and rechecks the saved evidence before +publication. An unrelated historical folder keeps the obligation on its real +source workspace. + +Finalized file sources receive a separate `finalizedWorkspaceRecoveryMigration` +run marker and an exact migration-operation receipt. They do not receive the +legacy execution-replay suppression marker. Future recovery exports an immutable +snapshot of the recorded allocation instead of merging into the retired local +folder. The existing source, provider-stop and archive checks still apply. +Completed copyback recovery receipts remain valid. Unknown or unverified recovery +sources continue to block import. + The workspace fingerprint pins the recorded source. An arriving reservation can reach a legacy workspace before refusing it, so publication must follow this order too. The lifecycle advisory lock alone does not fence runtime operations. diff --git a/server/src/services/execution-blocker.ts b/server/src/services/execution-blocker.ts index ba949595d33..53850101e4f 100644 --- a/server/src/services/execution-blocker.ts +++ b/server/src/services/execution-blocker.ts @@ -18,7 +18,7 @@ import { nativeManualAdmissionHoldsExecution } from "./native-runtime/native-run const blockerContextKeys = ["issueId", "taskId", "source", "connectionIntentResolution", "interactionKind", "interactionStatus", "mutation", "interactionId"] as const; -const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "legacyWorkspaceRecoveryMigration"] as const; +const blockerProfileKeys = ["adapterDispatch", "nativeExecutionInput", "nativeWorkspaceSync", "remoteWorkspace", "legacyWorkspaceRecoveryMigration", "finalizedWorkspaceRecoveryMigration"] as const; const blockerContextColumn = sql | null>`case when ${heartbeatRuns.contextSnapshot} is null then null else ( select coalesce(jsonb_object_agg(field.key, field.value), '{}'::jsonb) 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..214be049b16 --- /dev/null +++ b/server/src/services/finalized-workspace-recovery-migration.test.ts @@ -0,0 +1,232 @@ +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 { getExecutionBlocker } from "./execution-blocker.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.each([true, false])("requires a successful receipt for accepted-result retries (receipt: %s)", async saved => { + 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)); + if (saved) 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); + if (!saved) { + await expect(migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true })).rejects.toThrow("workspace_recovery_unresolved"); + expect(await preserved(f)).toEqual(before); + return; + } + 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("discovers a finalized child task's held source when importing its ancestor workspace", async () => { + const f = await fixture(true); + const [parent] = await db.insert(issues).values({ companyId: f.companyId, title: "Workspace owner" }).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)); + const before = await preserved(f); + await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true }); + expect((await preserved(f)).issue).toEqual(before.issue); + expect((await preserved(f)).action).toEqual(before.action); + const inspected = await inspectWorkspaceRecoverySource(db, { ...f, actionId: f.action.id }); + expect(inspected.sources).toEqual([expect.objectContaining({ preservationOnly: true, reference: f.reference })]); + expect((await preserved(f)).run.runnerProfileJson).toHaveProperty("finalizedWorkspaceRecoveryMigration.workspaceId", f.workspaceId); + }); + it.each(["old-copyback", "new-snapshot"])("retains verified repair through migration and the execution blocker (%s)", async mode => { + const f = await fixture(true), operationId = randomUUID(); + await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, workspaceRestoreFailure: "restore_unsafe_archive", + 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); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toMatchObject({ + recoveryActionId: f.action.id, workspaceRepairRequired: false, + }); + }); + it("requires every allocation and preserves mixed completed repair receipts", async () => { + const f = await fixture(true), otherLeaseId = randomUUID(), otherProviderLeaseId = randomUUID(); + const [lease] = await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.reference.leaseId)); + await db.insert(environmentLeases).values({ ...lease, id: otherLeaseId, providerLeaseId: otherProviderLeaseId }); + await db.update(issueRecoveryActions).set({ evidence: { ...f.action.evidence, + systemRecovery: { ...createSystemRecovery(f.runId), state: "complete" }, + workspaceRecovery: { ...(f.action.evidence.workspaceRecovery as object), status: "recovered" }, + historicalWorkspaceSource: { schema: "paperclip.historical-workspace-source.v1", runId: f.runId, + leaseIds: [f.reference.leaseId, otherLeaseId] }, + } }).where(eq(issueRecoveryActions.id, f.action.id)); + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, phase: "workspace_recovery", status: "succeeded", metadata: { + recoveryActionId: f.action.id, leaseId: f.reference.leaseId, providerLeaseId: f.reference.providerLeaseId, + descriptorSha256: f.reference.descriptorSha256, finalHostSha256: f.reference.finalHostSha256 } }); + await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true }); + expect(await verifiedWorkspaceRepairReceipt(db, (await preserved(f)).run)).toBeNull(); + const file = (suffix: string) => ({ objectKey: `${f.companyId}/workspace-recovery/${f.runId}/${otherLeaseId}/${"78".repeat(32)}.${suffix}`, + sha256: "78".repeat(32), byteSize: 1 }); + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, phase: "workspace_recovery_snapshot", status: "succeeded", metadata: { + schema: "paperclip.workspace-recovery-snapshot.v1", recoveryActionId: f.action.id, + leaseId: otherLeaseId, providerLeaseId: otherProviderLeaseId, + archive: file("tar"), manifest: file("jsonl") } }); + expect(await verifiedWorkspaceRepairReceipt(db, (await preserved(f)).run)).not.toBeNull(); + }); + it("preserves an older finalized reference with omitted optional fields", async () => { + const f = await fixture(true); + const { resourceDisposition: _disposition, ...legacy } = f.reference; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceSync: { + ...legacy, schema: "paperclip.native-workspace-sync/v1", + } } }).where(eq(heartbeatRuns.id, f.runId)); + await migrateRemoteWorkspace({ db, ...f, apply: true, writersStopped: true }); + const inspected = await inspectWorkspaceRecoverySource(db, { ...f, actionId: f.action.id }); + expect(inspected.sources).toEqual([expect.objectContaining({ preservationOnly: true, + reference: { ...legacy, schema: "paperclip.native-workspace-sync/v1", resourceDisposition: null } })]); + }); + 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..231a8d0b6d9 --- /dev/null +++ b/server/src/services/finalized-workspace-recovery-migration.ts @@ -0,0 +1,93 @@ +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"; +} + +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; + 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"))); + 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..5fd3426503c 100644 --- a/server/src/services/legacy-workspace-recovery-migration.ts +++ b/server/src/services/legacy-workspace-recovery-migration.ts @@ -2,7 +2,7 @@ import { isDeepStrictEqual } from "node:util"; import { and, eq, sql } from "drizzle-orm"; import { environmentLeases, executionWorkspaces, heartbeatRuns, issues, workspaceOperations, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority } from "@paperclipai/shared"; -import type { NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; +import { readNativeWorkspaceSyncReference, type NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; export type LegacyWorkspaceRecoverySource = { runId: string; issueId: string; reference: NativeWorkspaceSyncReference | null; @@ -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, readNativeWorkspaceSyncReference(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 afa8455e195..f85ce050d0f 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"; @@ -59,7 +60,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"); @@ -110,15 +111,15 @@ 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, status: heartbeatRuns.status, runtimeMode: heartbeatRuns.runtimeMode, + const referencedRuns = 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'`, + sql`${heartbeatRuns.runnerProfileJson}->'nativeWorkspaceSync'->>'state' in ('prepared', 'finalized')`, )); const legacyRecoverySources: LegacyWorkspaceRecoverySource[] = []; - for (const run of unexported.sort((a, b) => a.id.localeCompare(b.id))) { + for (const run of referencedRuns.filter(run => object(run.reference).state === "prepared").sort((a, b) => a.id.localeCompare(b.id))) { const reference = readNativeWorkspaceSyncReference(run.reference); const [owner] = await db.select({ phase: nativeRunFinalizations.phase, leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, resultId: nativeRunFinalizations.resultId }) @@ -203,23 +204,36 @@ async function inspect(db: ReadDb, input: { companyId: string; workspaceId: stri ) : undefined), )).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 recoveryIssueIds = [...new Set([workspace.sourceIssueId, ...referencedRuns.map(run => run.issueId), + ...legacyRecoverySources.map(source => source.issueId)].filter((id): id is string => Boolean(id)))]; + 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"); @@ -227,7 +241,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 }) { @@ -254,7 +268,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); @@ -273,14 +287,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, @@ -289,7 +305,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!; @@ -341,11 +357,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"); @@ -374,6 +396,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"); @@ -384,6 +407,8 @@ 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) { + 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 @@ -399,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..be0057ce357 100644 --- a/server/src/services/workspace-recovery-receipt.ts +++ b/server/src/services/workspace-recovery-receipt.ts @@ -2,7 +2,7 @@ import { and, desc, eq, sql, type AnyColumn } from "drizzle-orm"; import { executionWorkspaces, executionWorkspaceSnapshots, heartbeatRuns, issueRecoveryActions, workspaceOperations, type Db } from "@paperclipai/db"; import { readRemoteWorkspaceAuthority, remoteWorkspaceRunReferenceSchema, type ExecutionBlocker } from "@paperclipai/shared"; import { z } from "zod"; -import { readNativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; +import { readNativeWorkspaceSyncReference, type NativeWorkspaceSyncReference } from "./native-runtime/native-workspace-sync.js"; import { readMigratedLegacyWorkspaceSource } from "./legacy-workspace-recovery-migration.js"; import { readSnapshot } from "./workspace-recovery-snapshot.js"; @@ -62,6 +62,28 @@ export function workspaceRecoveryDisplay(evidence: Record, runI }; } +async function verifiedNativeCopybackReceipt(db: Db, input: { + companyId: string; issueId: string; runId: string; reference: NativeWorkspaceSyncReference; +}) { + const { reference } = input; + if (reference.state !== "finalized" || !reference.finalHostSha256) return null; + const [receipt] = await db.select({ id: workspaceOperations.id, metadata: workspaceOperations.metadata }) + .from(workspaceOperations).where(and(eq(workspaceOperations.companyId, input.companyId), + eq(workspaceOperations.issueId, input.issueId), eq(workspaceOperations.heartbeatRunId, input.runId), + eq(workspaceOperations.phase, "workspace_recovery"), eq(workspaceOperations.status, "succeeded"), + sql`${workspaceOperations.metadata}->>'leaseId' = ${reference.leaseId}`, + sql`${workspaceOperations.metadata}->>'providerLeaseId' = ${reference.providerLeaseId}`, + sql`${workspaceOperations.metadata}->>'descriptorSha256' = ${reference.descriptorSha256}`, + sql`${workspaceOperations.metadata}->>'finalHostSha256' = ${reference.finalHostSha256}`, + )).limit(1); + if (!receipt || typeof receipt.metadata?.recoveryActionId !== "string") return null; + const [action] = await db.select({ evidence: issueRecoveryActions.evidence }).from(issueRecoveryActions).where(and( + eq(issueRecoveryActions.companyId, input.companyId), eq(issueRecoveryActions.sourceIssueId, input.issueId), + eq(issueRecoveryActions.id, receipt.metadata.recoveryActionId))); + const state = readWorkspaceRecovery(action?.evidence ?? null); + return state?.runId === input.runId && state.status === "recovered" ? receipt.id : null; +} + /** Only the host writes workspace operations. Adapter output and a user-supplied * repair note cannot substitute for this exact copyback receipt. Stop/authority * are checked separately at every continuation admission. */ @@ -72,20 +94,18 @@ 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") { + const copyback = reference && allocation.leaseId === reference.leaseId + && allocation.providerLeaseId === reference.providerLeaseId + ? await verifiedNativeCopybackReceipt(db, { companyId: run.companyId, issueId, runId: run.id, reference }) : null; + if (!copyback) return null; + receiptId = copyback; + continue; + } try { readSnapshot(saved.metadata, run.companyId, run.id); } catch { return null; } receiptId = saved.id; } @@ -106,6 +133,7 @@ export async function verifiedWorkspaceRepairReceipt(db: Db, run: Pick