fix(recovery): preserve migrated sources through real blocker paths

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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<Record<string, unknown> | null>`case
   when ${heartbeatRuns.contextSnapshot} is null then null else (
     select coalesce(jsonb_object_agg(field.key, field.value), '{}'::jsonb)

server/src/services/finalized-workspace-recovery-migration.test.ts

@@ -10,6 +10,7 @@ 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";
@@ -91,7 +92,7 @@ describe.skipIf(!support.supported)("finalized workspace recovery preservation",
     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 () => {
+  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",
@@ -100,9 +101,14 @@ describe.skipIf(!support.supported)("finalized workspace recovery preservation",
     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,
+    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 } });
@@ -145,9 +151,22 @@ describe.skipIf(!support.supported)("finalized workspace recovery preservation",
     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 => {
+  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,
+    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));
@@ -161,6 +180,9 @@ describe.skipIf(!support.supported)("finalized workspace recovery preservation",
         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();

server/src/services/remote-workspace-migration.ts

@@ -110,15 +110,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<unknown>`${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,7 +203,11 @@ 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)))];
+  // A descendant task can use this workspace without becoming its source task.
+  // Discover holds from the exact recorded workspace references as well; the
+  // recovery owner still validates their task relationship and source binding.
+  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(