From 7a5859172121ed82631ddf3bcfdd4d8d4c73f75a Mon Sep 17 00:00:00 2001 From: Devin Foley Date: Fri, 9 Oct 2026 12:28:16 -0700 Subject: [PATCH] fix(recovery): preserve accepted workspace finalization during reaping (#15718) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - A native run saves its accepted result before it copies workspace changes back. > - A restart can leave that copyback under an unverified physical owner. > - The generic orphan reaper can mistake the exited provider for lost work and release the source lease. > - Recovery must preserve the accepted result and let the existing finalizer settle its remaining work. > - This pull request checks that authority at the conditional process-loss write and preserves the matching recovery hold. ## Linked Issues or Issue Description Related: #15475 added bounded process-loss evidence. The earlier #9687 addresses dead-run finalize barriers with synthetic receipts. This change preserves an existing accepted native result and requires real workspace settlement. **What happened?** After a server restart, the native finalizer required operator verification of unfinished copyback. The generic orphan reaper then marked the run `process_lost` and released its source lease. A new execution path also made the recovery reconciler clear the copyback hold before physical ownership was verified. **Expected behavior** An accepted result bound to the same company, issue, and run keeps its unfinished workspace suffix under native finalization. The generic reaper must not discard its source. The existing finalizer must still complete the work, apply its bounded failure policy, or require operator verification. **Steps to reproduce** 1. Accept a native result and leave its coordinator in `workspace_finalizing`. 2. Stop the controller before copyback completes, leaving its physical owner receipt. 3. Start recovery, then run the generic orphan and lease cleanup sweeps. 4. Before this change, the run can fail with `process_lost` despite its accepted result and explicit copyback hold. **Paperclip version or commit** Reproduced against `011e5669`; the same recovery code remains at base `1602512820`. **Deployment mode** A hosted server with PostgreSQL and native runner execution. The tests use only isolated local PostgreSQL and synthetic fixtures. ## What Changed - Check the exact accepted result and `workspace_finalizing` coordinator before generic process-loss cleanup. - Repeat the check in the conditional status write. Compare the heartbeat tuple version to reject a result publisher that commits while the update waits on its row lock. - Preserve nonterminal copyback holds when another execution path appears. Resolve only the matching run's hold after verified finalization commits. - Add fourteen real PostgreSQL service regressions and document the recovery boundary. ## Verification - `pnpm -r typecheck` passed. - `pnpm exec vitest run server/src/__tests__/heartbeat-process-recovery.test.ts -t 'native finalization'`: 14 passed. The test filter excluded 383 existing tests. - Tests cover startup, repeated reaper and cleanup sweeps, malformed and foreign owners, an expired old controller, exact database bindings, legacy and unsupported states, verified settlement, and unchanged terminal-task dispositions. - The concurrent publisher test uses the real accepted-result transaction and proves the reaper update is blocked by its row lock before the publisher commits. - The local `pnpm test:run` aggregate was stopped after setup timeouts in the chat and tool-access files and a damaged-epoch Runner test failure. It is not a full local pass. - The chat file passed all 1,145 tests when run alone. The tool-access file then passed 419 of 420; one case returned `socket hang up`. That single case passed on its isolated rerun. The cause of the socket failure is not established here. - The damaged-epoch Runner case still fails locally at its existing completion assertion. The same assertion failed on untouched baseline `556daf9a` before this patch; its test and Runner sources are unchanged. Exact-head CI passes this coverage. No workaround for that local fixture is included. - `pnpm build` passed. - Independent review approved this exact commit. All 14 new PostgreSQL regressions and 181 existing recovery/finalizer cases passed. One existing suite needed an isolated rerun after a setup timeout. - Independent negative controls reproduced the original bug. Removing only the tuple-version guard reproduced the blocked-row race. Removing only the hold exclusion cleared the safety hold too early. Restoring this patch passed all three checks. - GitHub CI passed all 53 applicable checks. The two Storybook jobs were skipped as expected. Greptile scored this commit 5/5 with no unresolved review threads. - Local runtime: Node 25.6.1. GitHub CI supplies the repository's Node 24 checks. - `git diff --check`, Gitleaks, and a local diff scan for secrets and private data passed. ## Risks - PostgreSQL tuple versions are used only as an optimistic fence for `process_lost`. Both accepted-result publishers lock and update that heartbeat in their result transaction. - The guard requires the bound accepted result and the durable coordinator state. A heartbeat phase or arbitrary owner JSON alone cannot suppress ordinary process loss. - Existing finalizer retries, ownership verification, explicit cancellation, and terminal-task dispositions remain unchanged. A physical owner can still require operator verification. - This change does not replay provider work or restore a source already destroyed by an older controller. No migration or runtime operation is included. ## Model Used OpenAI Codex, GPT-6 family, with reasoning, repository tools, code execution, and independent agent review. The exact service model ID and context-window size are not exposed in this worker session. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- doc/native-workspace-finalization-recovery.md | 18 ++ .../heartbeat-process-recovery.test.ts | 255 ++++++++++++++++++ server/src/services/heartbeat/recovery.ts | 7 + .../src/services/heartbeat/run-lifecycle.ts | 20 +- .../native-runtime/native-run-finalizer.ts | 5 +- .../native-workspace-finalization-state.ts | 29 ++ server/src/services/recovery/service.ts | 1 + 7 files changed, 332 insertions(+), 3 deletions(-) create mode 100644 server/src/services/native-runtime/native-workspace-finalization-state.ts 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.