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 index c1a4ca2cb9a..214be049b16 100644 --- a/server/src/services/finalized-workspace-recovery-migration.test.ts +++ b/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(); diff --git a/server/src/services/remote-workspace-migration.ts b/server/src/services/remote-workspace-migration.ts index 1c01f88f83a..d840e82a642 100644 --- a/server/src/services/remote-workspace-migration.ts +++ b/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`${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(