diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md index 7db6b4222f6..469514c278e 100644 --- a/doc/native-workspace-finalization-recovery.md +++ b/doc/native-workspace-finalization-recovery.md @@ -28,6 +28,24 @@ start time rather than trusted by PID alone. ## Unverified copyback after controller replacement +Generic orphan recovery does not turn a missing provider PID into `process_lost` +while the native coordinator is `workspace_finalizing` with an accepted result +bound to the same company, issue, and run. The finalizer owns that unfinished +suffix, including the interval before it acquires its physical owner receipt. +The heartbeat's phase or profile alone does not establish this authority. +The conditional orphan write rechecks the durable result and the heartbeat's +PostgreSQL tuple version, so a result published while the write waits on its row +lock cannot be overwritten using an older statement snapshot. Result publishers +lock and update that heartbeat in the same transaction as acceptance. + +The running run keeps its source lease while the existing finalizer either +completes, records its bounded failure policy, or requires operator stop +verification. A new queued run or a healthy child task cannot clear an +unverified copyback hold. Verified finalization clears the matching run's hold +after commitment. Existing terminal-task dispositions and explicit cancellation +remain unchanged; this does not recover a source already destroyed by an older +controller or claim that an unfinished export succeeded. + The controller cannot verify a process on a foreign or unknown host, or orphaned copyback children after an abrupt parent death before the success barrier. It surfaces `native_workspace_finalization_owner_unverified` as board-owned recovery, with no diff --git a/server/src/__tests__/heartbeat-process-recovery.test.ts b/server/src/__tests__/heartbeat-process-recovery.test.ts index f6f986de520..3cdbeb86a49 100644 --- a/server/src/__tests__/heartbeat-process-recovery.test.ts +++ b/server/src/__tests__/heartbeat-process-recovery.test.ts @@ -1,3 +1,5 @@ +import * as conversationContinuation from "../services/conversation-continuation.js"; +import { environmentRuntimeService } from "../services/environment-runtime.js"; import { ensureNativeCompletionContract } from "../services/native-runtime/completion-contracts.js"; import { readNativePlanWait, hasCommittedNativePlanWait } from "../services/native-runtime/native-plan-wait.js"; import { nativeSha256 } from "../services/native-runtime/canonical.js"; @@ -2096,6 +2098,259 @@ describeEmbeddedPostgres("heartbeat orphaned process recovery", () => { expect(result.continuationRequeued).toBe(0); }); + async function seedPendingNativeFinalization(accept = true) { + const fixture = await seedRunFixture({ runtimeMode: "native", adapterType: "paperclip_runner", processPid: 2_000_000_000 }); + const { companyId, agentId, issueId, runId } = fixture; + const contractId = randomUUID(), runnerInstanceId = randomUUID(); + const contractSha = `finalization-contract-${runId}`; + await db.insert(completionContracts).values({ + id: contractId, companyId, issueId, revision: 1, + schemaVersion: "paperclip.completion-contract.v1", policyVersion: "phase6-v1", + risk: "low", completionAuthority: "agent_claim_policy", incompleteCriteriaPolicy: "preserve_non_terminal", + contractJson: { revision: CONTROL_PLANE_CONFORMANCE_RESULT.completionClaim.contractRevision, + objective: "Keep completed work", criteria: [{ id: "objective", requirement: "Keep completed work" }] }, + canonicalSha256: contractSha, createdByActorType: "system", createdByActorId: "test", + }); + await db.update(heartbeatRuns).set({ nativeIssueId: issueId, nativeSessionId: runId, + runnerInstanceId, completionContractId: contractId, completionContractSha256: contractSha, + }).where(eq(heartbeatRuns.id, runId)); + const makePort = (targetDb = db) => new PaperclipControlPlanePort(targetDb, { companyId, issueId, runId, agentId, sessionId: runId, + completionContractId: contractId, completionContractSha256: contractSha, + sourceInstanceId: runnerInstanceId, controlPlaneSourceInstanceId: `finalization-control-${runId}`, + }); + const port = makePort(); + await port.openRun({ ...CONTROL_PLANE_CONFORMANCE_OPEN, + identity: { companyId, issueId, runId, agentId, sessionId: runId }, sourceInstanceId: runnerInstanceId }); + const complete = (targetDb = db) => makePort(targetDb).completeRun({ result: CONTROL_PLANE_CONFORMANCE_RESULT, + terminal: CONTROL_PLANE_CONFORMANCE_TERMINAL, callerResultId: `finalization-result-${runId}` }); + if (accept) await complete(); + const [local] = await db.select().from(environments).where(eq(environments.driver, "local")); + const lease = local + ? { environmentId: local.id, leaseId: (await db.insert(environmentLeases).values({ + companyId, issueId, heartbeatRunId: runId, environmentId: local.id, + status: "active", leasePolicy: "ephemeral", provider: "local", metadata: { driver: "local" }, + updatedAt: new Date(0), + }).returning())[0].id } + : await seedEnvironmentLeaseFixture(fixture); + return { ...fixture, ...lease, complete }; + } + + it.each(["foreign", "malformed", "expired_controller"] as const)( + "preserves native finalization source and board hold after restart with %s owner", + async (kind) => { + const f = await seedPendingNativeFinalization(); + const owner = kind === "malformed" ? { unknown: true } : { + token: randomUUID(), controllerBootId: randomUUID(), + hostname: kind === "foreign" ? "previous-controller.invalid" : os.hostname(), + pid: 2_000_000_000, processStartedAt: new Date(0).toISOString(), + }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }) + .where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ leaseOwner: null, leaseExpiresAt: new Date(0), + controllerPid: 2_000_000_000, controllerProcessStartedAt: new Date(0), + }).where(eq(nativeRunFinalizations.runId, f.runId)); + const accepted = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + const runtime = environmentRuntimeService(db); + const release = vi.spyOn(runtime, "releaseRunLeases"); + const destroy = vi.spyOn(runtime, "destroyRunLease"); + const heartbeat = heartbeatService(db, { environmentRuntime: runtime }); + await heartbeat.recoverNativeRunsAfterRestart(); + const [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(hold).toMatchObject({ status: "active", cause: "native_workspace_finalization_owner_unverified", ownerType: "board", wakePolicy: null }); + for (let tick = 0; tick < 2; tick += 1) { + expect(await heartbeat.reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] }); + // The held run itself is an execution path; it does not settle copyback. + await heartbeat.reconcileStrandedAssignedIssues(); + const actions = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + expect(actions).toHaveLength(1); + expect(actions[0]).toMatchObject({ id: hold.id }); + expect(["active", "escalated"]).toContain(actions[0].status); + } + expect((await heartbeat.getRun(f.runId))).toMatchObject({ status: "running", nativePhase: "workspace_finalizing", errorCode: null, + runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0]) + .toMatchObject({ status: "active", releasedAt: null }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toEqual(accepted); + expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, f.runId))).toEqual([]); + expect(release).not.toHaveBeenCalled(); + expect(destroy).not.toHaveBeenCalled(); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + expect(mockExecutePaperclipNativeSession).not.toHaveBeenCalled(); + await waitForPendingRunFailureReports(); + expect(mockCaptureRunFailure).not.toHaveBeenCalled(); + await heartbeat.drainActiveRunExecutions(); + }, + ); + + it("rejects a stale orphan candidate when native finalization commits before owner acquisition", async () => { + const f = await seedPendingNativeFinalization(false); + // Avoid the unrelated observed-owner gate while the old enumeration still + // lacks a result. Publish through the real result transaction just before + // generic orphan recovery reaches its conditional status write. + await db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing" }).where(eq(nativeRunFinalizations.runId, f.runId)); + const usedConversation = vi.spyOn(conversationContinuation, "runUsedConversationAdapter").mockImplementationOnce(async () => { + await f.complete(); + return false; + }); + const runtime = environmentRuntimeService(db); + const release = vi.spyOn(runtime, "releaseRunLeases"); + try { + expect(await heartbeatService(db, { environmentRuntime: runtime }).reapOrphanedRuns()).toEqual({ reaped: 0, runIds: [] }); + expect(usedConversation).toHaveBeenCalledOnce(); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "running", nativePhase: "workspace_finalizing", errorCode: null }); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0].status).toBe("active"); + expect(release).not.toHaveBeenCalled(); + } finally { usedConversation.mockRestore(); } + }); + + it("fences a native finalization publisher that commits while the orphan UPDATE waits on its row lock", async () => { + const f = await seedPendingNativeFinalization(false); + await db.update(nativeRunFinalizations).set({ phase: "workspace_finalizing" }).where(eq(nativeRunFinalizations.runId, f.runId)); + let release!: () => void; + const gate = new Promise(resolve => { release = resolve; }); + let entered!: () => void; + const writerEntered = new Promise(resolve => { entered = resolve; }); + let publisherPid = 0; + let publisher: Promise | undefined; + const usedConversation = vi.spyOn(conversationContinuation, "runUsedConversationAdapter").mockImplementationOnce(async () => { + publisher = db.transaction(async tx => { + const rows = await tx.execute(sql`select pg_backend_pid() as pid`); + publisherPid = Number(rows[0].pid); + // This is the real publication protocol, not a synthetic version bump: + // completeRun locks heartbeat, accepts the result, and writes its phase. + await f.complete(tx as unknown as typeof db); + entered(); + await gate; + }); + await writerEntered; + return false; + }); + const runtime = environmentRuntimeService(db); + const releaseLease = vi.spyOn(runtime, "releaseRunLeases"); + const heartbeat = heartbeatService(db, { environmentRuntime: runtime }); + const reaping = heartbeat.reapOrphanedRuns(); + try { + await writerEntered; + await vi.waitFor(async () => { + const waiting = await db.execute(sql`select pid from pg_stat_activity + where ${publisherPid} = any(pg_blocking_pids(pid)) and wait_event_type = 'Lock'`); + expect(waiting.length).toBeGreaterThan(0); + }, { timeout: 5_000 }); + release(); + await publisher; + expect(await reaping).toEqual({ reaped: 0, runIds: [] }); + expect((await heartbeat.getRun(f.runId))).toMatchObject({ status: "running", errorCode: null, nativePhase: "workspace_finalizing" }); + expect((await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)))).toHaveLength(1); + expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.leaseId)))[0].status).toBe("active"); + expect(releaseLease).not.toHaveBeenCalled(); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + expect(mockExecutePaperclipNativeSession).not.toHaveBeenCalled(); + await waitForPendingRunFailureReports(); + expect(mockCaptureRunFailure).not.toHaveBeenCalled(); + } finally { + release(); + await publisher?.catch(() => undefined); + await reaping.catch(() => undefined); + usedConversation.mockRestore(); + } + }); + + it.each(["phase_only", "rejected_result", "terminal_failure", "unknown_phase"] as const)( + "does not invent native finalization authority from %s", + async (kind) => { + const f = await seedPendingNativeFinalization(kind !== "phase_only"); + await db.update(heartbeatRuns).set({ nativePhase: "workspace_finalizing", + runnerProfileJson: { nativeWorkspaceFinalizationOwner: { unknown: true } }, + }).where(eq(heartbeatRuns.id, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: kind === "terminal_failure" ? "terminal_failure" + : kind === "unknown_phase" ? "unknown_phase" : "workspace_finalizing", + // Prevent the finalizer from changing this negative control first. + leaseOwner: "unverified-old-owner", leaseExpiresAt: new Date(Date.now() + 60_000), + }).where(eq(nativeRunFinalizations.runId, f.runId)); + if (kind === "rejected_result") await db.update(nativeRunResults).set({ schemaStatus: "rejected" }).where(eq(nativeRunResults.runId, f.runId)); + const result = await heartbeatService(db).reapOrphanedRuns(); + expect(result.runIds).toContain(f.runId); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "failed", errorCode: "process_lost" }); + }, + ); + + it("resolves the exact native finalization hold after verified workspace settlement without replay", async () => { + const f = await seedPendingNativeFinalization(); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: { + token: randomUUID(), hostname: "previous-controller.invalid", pid: 2_000_000_000, processStartedAt: new Date(0).toISOString(), + } } }).where(eq(heartbeatRuns.id, f.runId)); + const heartbeat = heartbeatService(db); + await heartbeat.recoverNativeRunsAfterRestart(); + const [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + const accepted = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + // Model exact operator stop verification and a completed copyback receipt; + // no test or product path infers these facts from a dead provider PID. + await db.update(heartbeatRuns).set({ runnerProfileJson: {} }).where(eq(heartbeatRuns.id, f.runId)); + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, phase: "workspace_finalize", status: "succeeded" }); + await heartbeat.reapOrphanedRuns(); + expect((await heartbeat.getRun(f.runId))).toMatchObject({ status: "succeeded", nativePhase: "committed" }); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id)))[0]) + .toMatchObject({ status: "resolved", outcome: "restored" }); + expect(await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, f.runId))).toEqual(accepted); + expect(mockExecutePaperclipNativeSession).not.toHaveBeenCalled(); + expect(mockAdapterExecute).not.toHaveBeenCalled(); + await heartbeat.drainActiveRunExecutions(); + }); + + it("rejects foreign result, company, and issue bindings for native finalization", async () => { + const source = await seedPendingNativeFinalization(); + const other = await seedPendingNativeFinalization(); + const [otherResult] = await db.select().from(nativeRunResults).where(eq(nativeRunResults.runId, other.runId)); + for (const patch of [{ resultId: otherResult.id }, { companyId: other.companyId }, { issueId: other.issueId }]) { + await expect(db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, source.runId))) + .rejects.toMatchObject({ cause: { code: "23503" } }); + } + // The composite foreign keys are part of the guard's durable authority; + // invalid foreign bindings never become a candidate through a valid write. + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, source.runId)))[0]) + .toMatchObject({ companyId: source.companyId, issueId: source.issueId, phase: "workspace_finalizing" }); + }); + + it("does not extend native finalization authority to a legacy runtime", async () => { + const f = await seedPendingNativeFinalization(); + await db.update(heartbeatRuns).set({ runtimeMode: "legacy" }).where(eq(heartbeatRuns.id, f.runId)); + expect((await heartbeatService(db).reapOrphanedRuns()).runIds).toContain(f.runId); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, f.runId)))[0]) + .toMatchObject({ status: "failed", errorCode: "process_lost" }); + }); + + it("does not clear another run's native finalization hold on the same task", async () => { + const f = await seedPendingNativeFinalization(); + const [otherHold] = await db.insert(issueRecoveryActions).values({ companyId: f.companyId, + sourceIssueId: f.issueId, kind: "active_run_watchdog", status: "active", ownerType: "board", + cause: "native_workspace_finalization_owner_unverified", fingerprint: randomUUID(), + evidence: { runId: randomUUID() }, nextAction: "Verify the other physical owner has stopped.", + }).returning(); + await db.insert(workspaceOperations).values({ companyId: f.companyId, issueId: f.issueId, + heartbeatRunId: f.runId, phase: "workspace_finalize", status: "succeeded" }); + await expect(finalizeNativeRun({ db, runId: f.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true })) + .resolves.toMatchObject({ phase: "committed" }); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, otherHold.id)))[0]) + .toMatchObject({ status: "active", resolvedAt: null }); + }); + + it("keeps existing terminal-task disposition for a native finalization hold", async () => { + const f = await seedPendingNativeFinalization(); + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: { unknown: true } } }) + .where(eq(heartbeatRuns.id, f.runId)); + const heartbeat = heartbeatService(db); + await heartbeat.recoverNativeRunsAfterRestart(); + const [hold] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, f.issueId)); + await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId)); + await heartbeat.reconcileStrandedAssignedIssues(); + expect((await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, hold.id)))[0]) + .toMatchObject({ status: "resolved", resolutionNote: "source_terminal" }); + await heartbeat.drainActiveRunExecutions(); + }); + it.each(["settled", "rejected", "late_callback", "provider_transport_failed"] as const)( "joins startup and reap retained cleanup, keeps recovery live, and drains its %s operation", async (outcome) => { diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index 4798251a1fd..a727f4ad5f4 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,4 +1,5 @@ import { preserveWorkspaceRestoreRecoveryMetadataSql } from "../legacy-workspace-restore-recovery.js"; +import { pendingNativeWorkspaceFinalizationCondition } from "../native-runtime/native-workspace-finalization-state.js"; import { CHAT_COMPLETION_WAKE_REASON } from "../chat-completion-delivery.js"; import { hasStopOnlyCleanup, @@ -2059,6 +2060,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD nativeControllerProcessStartedAt: nativeRunFinalizations.controllerProcessStartedAt, nativeControllerLeaseExpiresAt: nativeRunFinalizations.leaseExpiresAt, + nativeWorkspaceFinalizationPending: pendingNativeWorkspaceFinalizationCondition(db), }) .from(heartbeatRuns) .innerJoin(agents, eq(heartbeatRuns.agentId, agents.id)) @@ -2110,10 +2112,15 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD nativeControllerPid, nativeControllerProcessStartedAt, nativeControllerLeaseExpiresAt, + nativeWorkspaceFinalizationPending, } of activeRuns) { // Authentication timeout requires an explicit ownership resolution, not // repeated reattachment or a process-gone guess on subsequent sweeps. if (isNativeRunnerOwnershipHeld(run)) continue; + // The accepted result, not a missing provider process, owns this suffix. + // The reconciler above either finishes it, records a bounded failure, or + // preserves its explicit physical-copyback hold. Keep its source lease. + if (nativeWorkspaceFinalizationPending) continue; const nativeRun = run.runtimeMode === "native"; const nativeProcessPidAlive = nativeRun && !!run.processPid && isProcessAlive(run.processPid); diff --git a/server/src/services/heartbeat/run-lifecycle.ts b/server/src/services/heartbeat/run-lifecycle.ts index 865aaca7b72..245e36a0ab1 100644 --- a/server/src/services/heartbeat/run-lifecycle.ts +++ b/server/src/services/heartbeat/run-lifecycle.ts @@ -17,6 +17,7 @@ import { CONFIGURATION_INCOMPLETE_FAILURE_CODE } from "./run-preparation.js"; import { WORKSPACE_VALIDATION_FAILURE_CODE } from "./workspaces.js"; import { boundHeartbeatRunEventPayloadForStorage } from "./run-log.js"; import { preserveWorkspaceRestoreRecoveryMetadata } from "../workspace-restore-recovery-state.js"; +import { pendingNativeWorkspaceFinalizationCondition } from "../native-runtime/native-workspace-finalization-state.js"; import { activeIssueInteractionCondition } from "../issue-question-context.js"; import { nativeRetryCancellationCommitCondition, @@ -45,6 +46,7 @@ import { inArray, isNull, ne, + not, notInArray, or, sql, @@ -1092,11 +1094,12 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl // fromStatuses can name a terminal status as its own source (for example, // an idempotent "still failed" patch), so the write below is not always a // genuine transition. Read the pre-write status to tell the two apart. - const previousStatus = await db - .select() + const previous = await db + .select({ run: heartbeatRuns, rowVersion: sql`${heartbeatRuns}.xmin::text` }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, runId)) .then((rows) => rows[0] ?? null); + const previousStatus = previous?.run ?? null; if (previousStatus && (status === "cancelled" || status === "interrupted")) { patch = { ...patch, resultJson: cancellationResultJson(previousStatus, status, patch?.resultJson, patch?.errorCode, patch?.error) }; @@ -1151,6 +1154,19 @@ export function createHeartbeatLifecycle(db: Db, dependencies: HeartbeatLifecycl and( eq(heartbeatRuns.id, runId), inArray(heartbeatRuns.status, fromStatuses), + // A result can commit after the orphan candidate was read. + // Reject that stale process-loss transition before reporting, + // issue promotion, or release of the unfinished source lease. + ...(status === "failed" && patch?.errorCode === "process_lost" + ? [ + not(pendingNativeWorkspaceFinalizationCondition(db)), + // Result publication locks and updates this heartbeat in + // its transaction. If this UPDATE waited behind it, the + // joined tables can still use the older statement snapshot; + // the exact tuple fence also rejects that blocked writer. + sql`${heartbeatRuns}.xmin::text = ${previous?.rowVersion ?? null}`, + ] + : []), ...(cancellationCondition ? [cancellationCondition] : []), ...(isHeartbeatRunTerminalStatus(status) ? [nativeRunnerOwnershipNotHeldCondition()] diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 19d97de8c25..f62e6518947 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -607,7 +607,10 @@ async function resolveCommittedFinalizationRecovery(db: Db, run: typeof heartbea eq(issueRecoveryActions.companyId, run.companyId), eq(issueRecoveryActions.sourceIssueId, issueId), eq(issueRecoveryActions.kind, "active_run_watchdog"), inArray(issueRecoveryActions.status, ["active", "escalated"]), sql`${issueRecoveryActions.evidence}->>'runId' = ${run.id}`, - sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`, + or( + sql`${issueRecoveryActions.wakePolicy}->>'kind' = 'resume_native_run'`, + eq(issueRecoveryActions.cause, "native_workspace_finalization_owner_unverified"), + ), )); for (const action of actions) await issueRecoveryActionService(db).resolveActiveForIssue({ companyId: run.companyId, sourceIssueId: issueId, actionId: action.id, diff --git a/server/src/services/native-runtime/native-workspace-finalization-state.ts b/server/src/services/native-runtime/native-workspace-finalization-state.ts new file mode 100644 index 00000000000..df83b63305e --- /dev/null +++ b/server/src/services/native-runtime/native-workspace-finalization-state.ts @@ -0,0 +1,29 @@ +import { heartbeatRuns, nativeRunFinalizations, nativeRunResults, type Db } from "@paperclipai/db"; +import { and, eq, exists } from "drizzle-orm"; + +/** + * An accepted result transfers recovery to the workspace finalizer. A dead + * provider PID cannot authorize discarding its unfinished workspace suffix, + * including the interval before physical copyback acquires its owner receipt. + * The heartbeat's display phase and arbitrary profile JSON are not evidence. + */ +export function pendingNativeWorkspaceFinalizationCondition(db: Db) { + return and( + eq(heartbeatRuns.runtimeMode, "native"), + exists(db.select({ runId: nativeRunFinalizations.runId }) + .from(nativeRunFinalizations) + .innerJoin(nativeRunResults, and( + eq(nativeRunResults.id, nativeRunFinalizations.resultId), + eq(nativeRunResults.runId, nativeRunFinalizations.runId), + eq(nativeRunResults.companyId, nativeRunFinalizations.companyId), + eq(nativeRunResults.issueId, nativeRunFinalizations.issueId), + eq(nativeRunResults.schemaStatus, "accepted"), + )) + .where(and( + eq(nativeRunFinalizations.runId, heartbeatRuns.id), + eq(nativeRunFinalizations.companyId, heartbeatRuns.companyId), + eq(nativeRunFinalizations.issueId, heartbeatRuns.nativeIssueId), + eq(nativeRunFinalizations.phase, "workspace_finalizing"), + ))), + )!; +} diff --git a/server/src/services/recovery/service.ts b/server/src/services/recovery/service.ts index d27f3e27175..9b1a5c4439d 100644 --- a/server/src/services/recovery/service.ts +++ b/server/src/services/recovery/service.ts @@ -3453,6 +3453,7 @@ export function recoveryService( // provider already did. Only execution reconciliation can clear this hold. if (requiresExecutionReconciliation(action.cause) || isNativeWorkspaceExportRepairCause(action.cause) + || action.cause === "native_workspace_finalization_owner_unverified" || action.cause === "native_workspace_sync_out_unsafe_archive") { // A queued wake or healthy child does not export this accepted result. // Only its native finalizer or an explicit board disposition can settle it.