fix: serialize native workspace finalization and preserve committed outcomes

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

doc/native-workspace-finalization-recovery.md

@@ -0,0 +1,60 @@
+# Native workspace finalization ownership and recovery
+
+Native workspace export and merge acquire a PostgreSQL advisory lock scoped to
+company and run before the first physical copyback. The live heartbeat and the
+reconciler share that lock. Recovery skips a busy owner without recording another
+workspace operation or spending a retry. A completed workspace barrier is reread
+under ownership before export, and a committed coordinator cannot be overwritten
+by a late failure receipt.
+
+The lock transaction holds no row locks. Ordinary progress and finalization
+receipts remain visible through the normal database pool. A dedicated connection outside the application pool is
+reserved for the duration of copyback and closed when it settles; even a one-connection application pool remains available for progress writes. Its run profile carries
+`nativeWorkspaceFinalizationOwner`, an exact token, host, PID, and process-start
+receipt. Losing the lock connection does not prove physical copyback stopped:
+a contender still refuses a receipt whose controller is alive. The original
+callback joins before its token is released, and publication checks the lock
+connection and token. Graceful completion clears the receipt. A controller that
+has exited on the same host can be recovered; PID reuse is checked against its
+recorded start time rather than trusted by PID alone.
+
+## Controller replacement on a different host
+
+The controller cannot verify a process on a foreign or unknown host. It surfaces
+`native_workspace_finalization_owner_unverified` as board-owned recovery, with no
+automatic provider wake. This is an intentional limit: elapsed time or a missing
+database connection never proves the old copyback process stopped.
+
+An instance operator must first verify through the deployment platform that the
+exact prior controller **and its copyback subprocesses** have stopped. Retain the
+sandbox, accepted native result, and workspace descriptor. Do not run a new
+provider turn, delete workspace contents, or relax archive confinement.
+
+After that platform verification, use a database maintenance transaction to
+release only the exact receipt shown by the recovery action. Replace the four
+placeholders with the action's company, run, token, and source issue. The advisory
+lock prevents concurrent acquisition during this change; the token comparison
+prevents clearing a newer owner. A zero-row update means ownership changed and
+requires fresh inspection. This maintenance operation is for a full-control
+instance operator, not an agent tool.
+
+```sql
+BEGIN;
+SELECT pg_advisory_xact_lock(hashtextextended(
+  'native-workspace-finalization:<company-id>:<run-id>', 0));
+UPDATE heartbeat_runs
+SET runner_profile_json = runner_profile_json - 'nativeWorkspaceFinalizationOwner'
+WHERE company_id = '<company-id>'::uuid
+  AND id = '<run-id>'::uuid
+  AND native_issue_id = '<issue-id>'::uuid
+  AND runtime_mode = 'native'
+  AND runner_profile_json->'nativeWorkspaceFinalizationOwner'->>'token' = '<owner-token>'
+RETURNING id;
+COMMIT;
+```
+
+Resolve the existing board recovery action with a note containing the platform
+stop evidence. The ordinary reconciliation sweep then resumes workspace
+finalization from the accepted result. Confirm the run's native phase and
+`resultJson.finalizationPhase` are `committed`, there is no `nextAttemptAt`, and
+no workspace operation is still running. The accepted provider result is reused.

packages/db/src/client.ts

@@ -255,6 +255,19 @@ export function postgresJsOptions(options: DatabaseClientOptions): Record<string
   return driverOptions;
 }
 
+// A long advisory-lock transaction must not borrow the normal pool that its
+// work needs for progress writes. Keep connection configuration private to the
+// originating Db lifetime; callers receive neither URLs nor credentials.
+const dedicatedDbFactories = new WeakMap<object, () => Db>();
+
+export async function withDedicatedDbConnection<T>(db: Db, action: (dedicated: Db) => Promise<T>): Promise<T> {
+  const factory = dedicatedDbFactories.get(db);
+  if (!factory) throw new Error("dedicated_connection_requires_create_db");
+  const dedicated = factory();
+  try { return await action(dedicated); }
+  finally { await dedicated.$client.end({ timeout: 1 }); }
+}
+
 export function createDb(url: string, options?: DatabaseClientOptions) {
   const resolved = resolveDatabaseClientOptions(options ?? databaseClientOptionsFromEnv());
   const sql = postgres(url, postgresJsOptions(resolved));
@@ -263,7 +276,11 @@ export function createDb(url: string, options?: DatabaseClientOptions) {
   // The registry keeps the real client (teardown must end the actual pool);
   // drizzle gets the retrying face so a pooler-recycled socket replays the
   // query instead of failing the request that happened to draw it.
-  return drizzlePg(withTransientWriteRetry(sql), { schema });
+  const db = drizzlePg(withTransientWriteRetry(sql), { schema });
+  dedicatedDbFactories.set(db, () => createDb(url, {
+    ...resolved, maxConnections: 1, applicationName: "paperclip-workspace-finalization-lock",
+  }));
+  return db;
 }
 
 export async function getPostgresDataDirectory(url: string): Promise<string | null> {

packages/db/src/index.ts

@@ -1,5 +1,6 @@
 export {
   createDb,
+  withDedicatedDbConnection,
   closeRegisteredClients,
   getPostgresDataDirectory,
   ensurePostgresDatabase,

server/src/services/heartbeat.ts

@@ -1,4 +1,10 @@
 import { classifyNativeWorkspaceFailure, type NativeWorkspaceFailureCode } from "./native-runtime/native-workspace-failure.js";
+import {
+  withNativeWorkspaceFinalizationOwnership,
+  NativeWorkspaceFinalizationBusyError,
+  NativeWorkspaceFinalizationOwnershipLostError,
+  type NativeWorkspaceFinalizationOwnership,
+} from "./native-runtime/native-workspace-finalization-ownership.js";
 import { applyWorkspaceRestoreFailure } from "@paperclipai/adapter-utils/workspace-restore-result";
 import { hasWorkspaceRestoreFailure } from "@paperclipai/shared";
 import { externalConversationStateSql, nonIdleSlackIssueCondition } from "./slack-conversation-state.js";
@@ -24437,44 +24443,75 @@ export function heartbeatService(
           // If recording the barrier itself fails, propagate as a run failure
           // rather than silently leaving dependents stranded behind a missing
           // finalize row.
-          if (nativeWorkspaceSync) {
-            await nativeWorkspaceSync.restoreWorkspace();
-          }
-          await db
-            .update(heartbeatRuns)
-            .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) })
-            .where(
-              and(
-                eq(heartbeatRuns.id, run.id),
-                eq(heartbeatRuns.status, "running"),
-              ),
-            );
-          const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded";
-          await recordWorkspaceFinalize(workspaceFinalizeStatus);
-          if (adapterResult.nativeFinalization) {
-            adapterResult.nativeFinalization.workspaceFinalizeStatus =
-              workspaceFinalizeStatus;
+          const completeWorkspace = async (ownership?: NativeWorkspaceFinalizationOwnership) => {
             try {
-              const finalized = await finalizeNativeRun({
-                db,
-                runId: run.id,
-                workspaceFinalizeStatus,
-                preserveProviderAttempt: Boolean(nativeWorkspaceSync),
-              });
-              await dispatchPendingNativeStatusWakeups({
-                companyId: run.companyId,
-              });
-              if (finalized.phase === "committed") {
-                await nativeWorkspaceSync?.cleanup();
+              if (nativeWorkspaceSync) {
+                const exported = await db.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
+                  eq(workspaceOperations.companyId, run.companyId),
+                  eq(workspaceOperations.heartbeatRunId, run.id),
+                  eq(workspaceOperations.phase, "workspace_finalize"),
+                  eq(workspaceOperations.status, "succeeded"),
+                )).limit(1);
+                if (exported.length) adapterFinalizeOutcome = "succeeded";
+                else await nativeWorkspaceSync.restoreWorkspace(ownership?.assertHeld);
               }
-            } catch (finalizeErr) {
-              logger.warn(
-                { err: finalizeErr, runId: run.id },
-                "native result persisted but finalization did not apply; the reconciliation loop will retry",
-              );
+              await ownership?.assertHeld();
+              await db
+                .update(heartbeatRuns)
+                .set({ executionControlDeadlineAt: new Date(Date.now() + 60_000) })
+                .where(
+                  and(
+                    eq(heartbeatRuns.id, run.id),
+                    eq(heartbeatRuns.status, "running"),
+                  ),
+                );
+              const workspaceFinalizeStatus = hasWorkspaceRestoreFailure(adapterResult.resultJson) ? "failed" : "succeeded";
+              await recordWorkspaceFinalize(workspaceFinalizeStatus);
+              if (adapterResult.nativeFinalization) {
+                adapterResult.nativeFinalization.workspaceFinalizeStatus =
+                  workspaceFinalizeStatus;
+                try {
+                  const finalized = await finalizeNativeRun({
+                    db,
+                    runId: run.id,
+                    workspaceFinalizeStatus,
+                    preserveProviderAttempt: Boolean(nativeWorkspaceSync),
+                  });
+                  await dispatchPendingNativeStatusWakeups({
+                    companyId: run.companyId,
+                  });
+                  if (finalized.phase === "committed") {
+                    await nativeWorkspaceSync?.cleanup();
+                  }
+                } catch (finalizeErr) {
+                  logger.warn(
+                    { err: finalizeErr, runId: run.id },
+                    "native result persisted but finalization did not apply; the reconciliation loop will retry",
+                  );
+                }
+              }
+            } catch (error) {
+              if (ownership) {
+                await ownership.assertHeld();
+                await recordWorkspaceFinalize("failed");
+              }
+              throw error;
             }
+          };
+          if (nativeWorkspaceSync) {
+            const owned = await withNativeWorkspaceFinalizationOwnership({
+              db, companyId: run.companyId, runId: run.id,
+            }, completeWorkspace);
+            if (!owned.acquired) throw new NativeWorkspaceFinalizationBusyError();
+          } else {
+            await completeWorkspace();
           }
         } catch (adapterErr) {
+          if (adapterErr instanceof NativeWorkspaceFinalizationBusyError
+            || adapterErr instanceof NativeWorkspaceFinalizationOwnershipLostError) {
+            nativeWorkspaceFinalizeScheduled = true;
+            throw adapterErr;
+          }
           if (adapterErr instanceof NativeControllerDetachedForRestartError) {
             // Preserve the provider and its run for the new controller. This
             // also keeps generic teardown from terminalizing/releasing its lease.
@@ -25403,6 +25440,14 @@ export function heartbeatService(
           }
           return;
         }
+        if (err instanceof NativeWorkspaceFinalizationBusyError
+          || err instanceof NativeWorkspaceFinalizationOwnershipLostError) {
+          // Another exact owner is finishing copyback, or this owner lost its
+          // lock connection. Preserve the accepted result and let reconciliation
+          // inspect durable ownership; neither case consumes an export retry.
+          logger.info({ runId: run.id, reason: err.message }, "native workspace finalization deferred to its durable owner");
+          return;
+        }
         if (err instanceof NativeWorkspaceFinalizeScheduledError) {
           const coordinator = await db
             .select({

server/src/services/native-runtime/native-finalization-reconciler.ts

@@ -856,6 +856,8 @@ export async function reconcileNativeFinalizations(
         runId: row.runId,
         environmentRuntime: options.environmentRuntime,
       });
+      // Busy is ownership, not another failed export or retry-budget debit.
+      if (!operation) continue;
       const workspaceFinalizeStatus =
         operation.status === "succeeded" ? "succeeded" : "failed";
       if (workspaceFinalizeStatus === "failed") {

server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts

@@ -12,6 +12,7 @@ import {
   nativeRunFinalizations,
   nativeRunResults,
   workAssessments,
+  issueRecoveryActions,
 } from "@paperclipai/db";
 import {
   getEmbeddedPostgresTestSupport,
@@ -509,6 +510,49 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a
     expect(replayed).toEqual(repaired);
   });
 
+  it("retires this run's scheduled finalization retry after successful commit", async () => {
+    const fixture = await seedNativeRun();
+    await driveToCompleteResult(fixture, CONTROL_PLANE_CONFORMANCE_TERMINAL);
+    await recordNativeFinalizationFailure({ db, runId: fixture.runId,
+      error: new Error("native_workspace_sync_out_failed"), failureScope: "workspace", projectRunStatus: true });
+    const [pending] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId));
+    expect(pending).toMatchObject({ status: "active", evidence: { runId: fixture.runId } });
+    await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    const [settled] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.id, pending.id));
+    expect(settled.status).toBe("resolved");
+  });
+
+  it("repairs stale retry metadata on an already committed successful run", async () => {
+    const fixture = await seedNativeRun();
+    await driveToCompleteResult(fixture, CONTROL_PLANE_CONFORMANCE_TERMINAL);
+    await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+    await db.update(heartbeatRuns).set({ resultJson: {
+      ...before.resultJson, finalizationPhase: "retryable_failure", failureCode: "native_finalization_invalid",
+      originalFailureCode: "native_finalization_invalid", nextAttemptAt: new Date().toISOString(),
+    } }).where(eq(heartbeatRuns.id, fixture.runId));
+    await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+    expect(after).toMatchObject({ status: "succeeded", nativePhase: "committed", resultJson: {
+      finalizationPhase: "committed", failureCode: null, originalFailureCode: null, nextAttemptAt: null,
+    } });
+  });
+
+  it("ignores a stale workspace failure after the same run committed successfully", async () => {
+    const fixture = await seedNativeRun();
+    await driveToCompleteResult(fixture, CONTROL_PLANE_CONFORMANCE_TERMINAL);
+    await finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true });
+    const [before] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+    const late = await recordNativeFinalizationFailure({
+      db, runId: fixture.runId, error: new Error("native_workspace_sync_out_failed"),
+      failureScope: "workspace", projectRunStatus: true,
+    });
+    expect(late.phase).toBe("committed");
+    const [after] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
+    expect(after).toEqual(before);
+    expect(await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, fixture.issueId))).toEqual([]);
+  });
+
   it("emits zero events when a retryable-failure write's conditional status spread is omitted", async () => {
     const fixture = await seedNativeRun();
     // recordNativeFinalizationFailure only needs the run and coordinator rows

server/src/services/native-runtime/native-run-finalizer.ts

@@ -12,12 +12,14 @@ import {
   heartbeatRuns,
   heartbeatRunEvents,
   issueApprovals,
+  issueRecoveryActions,
   issueThreadInteractions,
   issues,
   nativeRunFinalizations,
   nativeRunResults,
   statusDecisions,
   workAssessments,
+  workspaceOperations,
 } from "@paperclipai/db";
 import { classifyNativeEvidence } from "./evidence-classifier.js";
 import type { PrpIgnoredAttentionRequest } from "@paperclipai/paperclip-runner";
@@ -282,6 +284,29 @@ async function recordRetryableFailure(input: {
         : ("failed" as const);
   let terminalRunToEmit: typeof heartbeatRuns.$inferSelect | null = null;
   const outcome = await input.db.transaction(async (tx) => {
+    // Admission snapshots cannot authorize a failure write after a different
+    // owner committed success. Match status-committer lock ordering.
+    const current = await tx.select().from(nativeRunFinalizations).where(and(
+      eq(nativeRunFinalizations.runId, input.run.id),
+      eq(nativeRunFinalizations.companyId, input.run.companyId),
+    )).for("update").limit(1).then((rows) => rows[0] ?? null);
+    if (!current) throw new Error("native_finalization_missing");
+    if (current.phase === "committed" || current.phase === "terminal_failure"
+      || current.leaseOwner !== input.coordinator.leaseOwner) return current;
+    const [currentRun] = await tx.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, input.run.id)).limit(1);
+    if (!currentRun) throw new Error("native_finalization_run_missing");
+    if (input.failureScope === "workspace") {
+      const exported = await tx.select({ id: workspaceOperations.id }).from(workspaceOperations).where(and(
+        eq(workspaceOperations.companyId, input.run.companyId),
+        eq(workspaceOperations.heartbeatRunId, input.run.id),
+        eq(workspaceOperations.phase, "workspace_finalize"),
+        eq(workspaceOperations.status, "succeeded"),
+      )).limit(1);
+      // The live exporter may have acquired ownership since the stale error
+      // was raised, or already durably published a successful copyback.
+      if (exported.length || record(currentRun.runnerProfileJson).nativeWorkspaceFinalizationOwner) return current;
+    }
+    input = { ...input, coordinator: current, run: currentRun };
     const issue = await tx
       .select({
         lastStatusDecisionId: issues.lastStatusDecisionId,
@@ -525,6 +550,7 @@ export async function recordNativeFinalizationFailure(input: {
   if (!run || run.runtimeMode !== "native" || !coordinator) throw input.error;
   const message =
     input.error instanceof Error ? input.error.message : String(input.error);
+  if (message === "native_finalization_lease_busy") return coordinator;
   const failureCode = message.startsWith("native_")
     ? message
     : "native_finalization_invalid";
@@ -548,6 +574,19 @@ export async function recordNativeFinalizationFailure(input: {
   });
 }
 
+async function resolveCommittedFinalizationRecovery(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) {
+  const actions = await db.select().from(issueRecoveryActions).where(and(
+    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'`,
+  ));
+  for (const action of actions) await issueRecoveryActionService(db).resolveActiveForIssue({
+    companyId: run.companyId, sourceIssueId: issueId, actionId: action.id,
+    status: "resolved", outcome: "restored", resolutionNote: "The accepted native result and workspace finalization committed successfully; this run needs no further finalization retry.",
+  });
+}
+
 async function projectCommittedRun(input: {
   db: Db;
   run: typeof heartbeatRuns.$inferSelect;
@@ -601,10 +640,10 @@ async function projectCommittedRun(input: {
                   'observedAt', ${heartbeatRuns.updatedAt}
                 )
               )
-              else ${heartbeatRuns.resultJson}
-            end`,
+              else coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb)
+            end || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)`,
           }
-        : {}),
+        : { resultJson: sql`coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) || jsonb_build_object('finalizationPhase', 'committed', 'failureCode', null, 'originalFailureCode', null, 'nextAttemptAt', null)` }),
       updatedAt: now,
     })
     .where(
@@ -620,6 +659,10 @@ async function projectCommittedRun(input: {
               isNotNull(heartbeatRuns.errorCode),
               isNull(heartbeatRuns.finishedAt),
               sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
+              sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
+              sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
+              sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
+              sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
             ),
           ),
         ),
@@ -631,6 +674,10 @@ async function projectCommittedRun(input: {
           sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`,
           isNull(heartbeatRuns.finishedAt),
           sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`,
+          sql`${heartbeatRuns.resultJson}->>'finalizationPhase' is distinct from 'committed'`,
+          sql`${heartbeatRuns.resultJson}->>'nextAttemptAt' is not null`,
+          sql`${heartbeatRuns.resultJson}->>'failureCode' is not null`,
+          sql`${heartbeatRuns.resultJson}->>'originalFailureCode' is not null`,
           ...(terminalState === "succeeded"
             ? [isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode)]
             : []),
@@ -1047,6 +1094,7 @@ export async function finalizeNativeRun(input: {
         logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
       });
     }
+    await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId);
     return coordinator;
   }
   const [resultRow, contractRow] = await Promise.all([
@@ -1321,6 +1369,7 @@ export async function finalizeNativeRun(input: {
           resultJson: {
             ...record(run.resultJson),
             finalizationPhase,
+            ...(finalizationPhase === "committed" ? { failureCode: null, originalFailureCode: null, nextAttemptAt: null } : {}),
             assessmentId: assessmentRow.id,
             decisionId: committed.decision.id,
             authoritativeDecision: decision.toStatus,
@@ -1376,6 +1425,7 @@ export async function finalizeNativeRun(input: {
           logger.warn({ err, runId: run.id }, "Slack conversation settlement deferred to reconciliation");
         });
       }
+      if (finalizationPhase === "committed") await resolveCommittedFinalizationRecovery(input.db, run, coordinator.issueId);
       return {
         ...coordinator,
         phase: finalizationPhase,

server/src/services/native-runtime/native-workspace-finalization-ownership.ts

@@ -0,0 +1,122 @@
+import { issueRecoveryActionService } from "../issue-recovery-actions.js";
+import { nativeSha256 } from "./canonical.js";
+import { randomUUID } from "node:crypto";
+import os from "node:os";
+import { and, eq, sql } from "drizzle-orm";
+import { heartbeatRuns, withDedicatedDbConnection, type Db } from "@paperclipai/db";
+import {
+  currentNativeControllerIdentity,
+  evaluateNativeControllerTakeover,
+} from "./native-restart-recovery.js";
+
+const OWNER_KEY = "nativeWorkspaceFinalizationOwner";
+type Owner = {
+  token: string;
+  hostname: string;
+  pid: number;
+  processStartedAt: string;
+};
+export type NativeWorkspaceFinalizationOwnership = {
+  token: string;
+  /** Recheck both the physical lock connection and the durable owner before publishing. */
+  assertHeld(): Promise<void>;
+};
+export class NativeWorkspaceFinalizationBusyError extends Error {
+  constructor() { super("native_workspace_finalization_busy"); }
+}
+export class NativeWorkspaceFinalizationOwnershipLostError extends Error {
+  constructor(options?: ErrorOptions) {
+    super("native_workspace_finalization_ownership_lost", options);
+  }
+}
+
+function readOwner(value: unknown): Owner | null {
+  if (!value || typeof value !== "object") return null;
+  const owner = value as Partial<Owner>;
+  return typeof owner.token === "string" && typeof owner.hostname === "string"
+    && Number.isInteger(owner.pid) && Number(owner.pid) > 0
+    && typeof owner.processStartedAt === "string" && Number.isFinite(Date.parse(owner.processStartedAt))
+    ? owner as Owner : null;
+}
+
+/**
+ * The advisory lock serializes controllers without holding row locks across I/O.
+ * The durable receipt additionally prevents a disconnected lock session from
+ * authorizing takeover while its controller is still doing physical copyback.
+ * There is deliberately no wall-clock timeout that can steal a slow export.
+ */
+export async function withNativeWorkspaceFinalizationOwnership<T>(
+  input: { db: Db; companyId: string; runId: string },
+  action: (ownership: NativeWorkspaceFinalizationOwnership) => Promise<T>,
+): Promise<{ acquired: false } | { acquired: true; value: T }> {
+  const identity = await currentNativeControllerIdentity();
+  const owner: Owner = {
+    token: randomUUID(), hostname: os.hostname(), pid: identity.pid,
+    processStartedAt: identity.processStartedAt.toISOString(),
+  };
+  const scope = and(eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.runtimeMode, "native"));
+  const owned = and(scope, sql`${heartbeatRuns.runnerProfileJson}->${OWNER_KEY}->>'token' = ${owner.token}`);
+  let claimed = false;
+  let actionPromise: Promise<T> | undefined;
+  try {
+    return await withDedicatedDbConnection(input.db, (dedicated) => dedicated.transaction(async (lock) => {
+      const rows = await lock.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(${`native-workspace-finalization:${input.companyId}:${input.runId}`}, 0)) as acquired`);
+      if (rows[0]?.acquired !== true) return { acquired: false } as const;
+      // Reads and durable progress use the ordinary pool, not the lock transaction.
+      const [run] = await input.db.select({ profile: heartbeatRuns.runnerProfileJson, issueId: heartbeatRuns.nativeIssueId, agentId: heartbeatRuns.agentId }).from(heartbeatRuns).where(scope).limit(1);
+      if (!run) throw new Error("native_workspace_finalization_binding_missing");
+      const rawPrior = run.profile?.[OWNER_KEY];
+      if (rawPrior != null) {
+        const prior = readOwner(rawPrior);
+        // A foreign host or malformed receipt cannot prove the old physical owner stopped.
+        if (!prior || prior.hostname !== owner.hostname) {
+          if (run.issueId) await issueRecoveryActionService(input.db).upsertSourceScoped({
+            companyId: input.companyId, sourceIssueId: run.issueId,
+            kind: "active_run_watchdog", ownerType: "board", returnOwnerAgentId: run.agentId,
+            cause: "native_workspace_finalization_owner_unverified",
+            fingerprint: nativeSha256({ runId: input.runId, owner: rawPrior }),
+            evidence: { runId: input.runId, owner: rawPrior },
+            nextAction: "Verify the previous controller and its workspace-copyback processes have stopped, then release only this exact workspace owner receipt using doc/native-workspace-finalization-recovery.md. Resume workspace finalization without another provider turn.",
+            wakePolicy: null, maxAttempts: 1, supersedeOnIdentityChange: true,
+          });
+          return { acquired: false } as const;
+        }
+        const takeover = await evaluateNativeControllerTakeover({
+          owner: { leaseOwner: prior.token, leaseExpiresAt: new Date(0), controllerPid: prior.pid,
+            controllerProcessStartedAt: new Date(prior.processStartedAt) }, now: new Date(),
+        });
+        if (!takeover.allowed) return { acquired: false } as const;
+      }
+      await input.db.update(heartbeatRuns).set({
+        runnerProfileJson: sql`jsonb_set(coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb), array[${OWNER_KEY}], ${JSON.stringify(owner)}::jsonb)`,
+      }).where(scope);
+      claimed = true;
+      const ownership: NativeWorkspaceFinalizationOwnership = {
+        token: owner.token,
+        assertHeld: async () => {
+          try {
+            // A failed transaction is never transparently reconnected by the DB client.
+            await lock.execute(sql`select 1`);
+            const [current] = await input.db.select({ id: heartbeatRuns.id }).from(heartbeatRuns).where(owned).limit(1);
+            if (!current) throw new Error("owner_replaced");
+          } catch (cause) {
+            throw new NativeWorkspaceFinalizationOwnershipLostError({ cause });
+          }
+        },
+      };
+      actionPromise = action(ownership);
+      const value = await actionPromise;
+      await ownership.assertHeld();
+      return { acquired: true, value } as const;
+    }));
+  } finally {
+    // postgres.js may reject the transaction when the socket dies before the
+    // callback settles. Join that callback before releasing its durable fence.
+    await actionPromise?.catch(() => undefined);
+    if (claimed) {
+      await input.db.update(heartbeatRuns).set({
+        runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`,
+      }).where(owned).catch(() => undefined);
+    }
+  }
+}

server/src/services/native-runtime/native-workspace-finalizer.test.ts

@@ -1,9 +1,13 @@
+import { spawn } from "node:child_process";
+import { once } from "node:events";
+import { readProcessStartedAt } from "../hot-restart.js";
+import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js";
 import { randomUUID } from "node:crypto";
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
-import { afterAll, beforeAll, describe, expect, it } from "vitest";
-import { and, desc, eq } from "drizzle-orm";
+import { afterAll, beforeAll, describe, expect, it, vi } from "vitest";
+import { and, desc, eq, sql } from "drizzle-orm";
 import {
   agents,
   companies,
@@ -14,6 +18,7 @@ import {
   issues,
   nativeRunFinalizations,
   nativeRunResults,
+  issueRecoveryActions,
   projects,
   workspaceOperations,
 } from "@paperclipai/db";
@@ -154,6 +159,133 @@ describe("native workspace finalization recovery", () => {
     }
   });
 
+  it("reuses a successful export even if an old controller wrote a later failed barrier", async () => {
+    const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Late failed barrier" });
+    const [success] = await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: seeded.runId,
+      issueId: seeded.issueId, phase: "workspace_finalize", status: "succeeded" }).returning();
+    await db.insert(workspaceOperations).values({ companyId, heartbeatRunId: seeded.runId,
+      issueId: seeded.issueId, phase: "workspace_finalize", status: "failed", createdAt: new Date(Date.now() + 1000) });
+    const operation = await resumeNativeWorkspaceFinalization({ db, runId: seeded.runId });
+    expect(operation?.id).toBe(success.id);
+    expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, seeded.runId))).toHaveLength(2);
+  });
+
+  it("keeps ordinary progress writable even when the application pool has one connection", async () => {
+    const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Small application pool" });
+    const smallPool = createDb(temporary.connectionString, { maxConnections: 1 });
+    const result = await withNativeWorkspaceFinalizationOwnership({ db: smallPool, companyId, runId: seeded.runId }, async (ownership) => {
+      await ownership.assertHeld();
+      await smallPool.update(heartbeatRuns).set({ updatedAt: new Date() }).where(eq(heartbeatRuns.id, seeded.runId));
+      return "progress saved";
+    });
+    expect(result).toEqual({ acquired: true, value: "progress saved" });
+  });
+
+  it("skips recovery while the live heartbeat owns slow copyback", async () => {
+    const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Live copyback" });
+    const live = await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async (ownership) => {
+      await ownership.assertHeld();
+      expect(await resumeNativeWorkspaceFinalization({ db: createDb(temporary.connectionString), runId: seeded.runId })).toBeNull();
+      expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, seeded.runId))).toEqual([]);
+      return "copied";
+    });
+    expect(live).toEqual({ acquired: true, value: "copied" });
+    const [run] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
+    expect(run.runnerProfileJson).not.toHaveProperty("nativeWorkspaceFinalizationOwner");
+  });
+
+  it("keeps the physical owner fenced after its lock connection dies until copyback joins", async () => {
+    const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Disconnected copyback" });
+    let enter!: () => void;
+    const entered = new Promise<void>((resolve) => { enter = resolve; });
+    let finish!: () => void;
+    const finished = new Promise<void>((resolve) => { finish = resolve; });
+    let published = false;
+    const first = withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async (ownership) => {
+      enter(); await finished; await ownership.assertHeld(); published = true;
+    });
+    // Observe rejection immediately, while retaining the promise until its callback joins.
+    const outcome = first.then(() => null, (error: unknown) => error);
+    try {
+      await entered;
+      await db.execute(sql`select pg_terminate_backend(pid) from pg_locks where locktype = 'advisory' and database = (select oid from pg_database where datname = current_database()) and pid <> pg_backend_pid()`);
+      const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(temporary.connectionString), companyId, runId: seeded.runId }, async () => "must not run");
+      expect(second).toEqual({ acquired: false });
+      const [held] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
+      expect(held.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toBeTruthy();
+    } finally { finish(); }
+    expect(await outcome).toBeTruthy();
+    expect(published).toBe(false);
+    expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async () => "recovered"))
+      .toEqual({ acquired: true, value: "recovered" });
+  });
+
+  it("recovers a receipt after the exact controller process on this host exited", async () => {
+    const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Dead copyback controller" });
+    const child = spawn(process.execPath, ["-e", "setInterval(() => {}, 1000)"], { stdio: "ignore" });
+    await once(child, "spawn");
+    const startedAt = await readProcessStartedAt(child.pid!);
+    expect(startedAt).toBeTruthy();
+    const exited = once(child, "close");
+    child.kill("SIGTERM"); await exited;
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: {
+      token: randomUUID(), hostname: os.hostname(), pid: child.pid, processStartedAt: startedAt,
+    } } }).where(eq(heartbeatRuns.id, seeded.runId));
+    expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async () => "recovered"))
+      .toEqual({ acquired: true, value: "recovered" });
+  });
+
+  it("surfaces an unverifiable foreign controller as board recovery without physical writes", async () => {
+    const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Foreign copyback controller" });
+    const owner = { token: randomUUID(), hostname: "other-controller.invalid", pid: 42, processStartedAt: new Date().toISOString() };
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } }).where(eq(heartbeatRuns.id, seeded.runId));
+    const work = vi.fn();
+    expect(await withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, work)).toEqual({ acquired: false });
+    expect(work).not.toHaveBeenCalled();
+    const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, seeded.issueId));
+    expect(action).toMatchObject({ ownerType: "board", cause: "native_workspace_finalization_owner_unverified", wakePolicy: null, evidence: { runId: seeded.runId, owner } });
+    expect(action.nextAction).toContain("previous controller");
+    await expect(withNativeWorkspaceFinalizationOwnership({ db, companyId: foreignCompanyId, runId: seeded.runId }, work))
+      .rejects.toThrow("native_workspace_finalization_binding_missing");
+    const [held] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId));
+    expect(held.runnerProfileJson?.nativeWorkspaceFinalizationOwner).toEqual(owner);
+  });
+
+  it("does not start another recovery while workspace finalization is still running", async () => {
+    const cwd = path.join(workspaceRoot, "slow-workspace");
+    await fs.mkdir(cwd);
+    const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Slow workspace export" });
+    await db.insert(workspaceOperations).values({
+      companyId, heartbeatRunId: seeded.runId, issueId: seeded.issueId,
+      phase: "workspace_finalize", cwd, status: "failed",
+    });
+    let entered!: () => void;
+    const started = new Promise<void>((resolve) => { entered = resolve; });
+    let release!: () => void;
+    const held = new Promise<void>((resolve) => { release = resolve; });
+    const stat = fs.stat.bind(fs);
+    let calls = 0;
+    const spy = vi.spyOn(fs, "stat").mockImplementation(async (...args: Parameters<typeof fs.stat>) => {
+      if (args[0] === cwd) { calls += 1; entered(); await held; }
+      return stat(...args);
+    });
+    const first = resumeNativeWorkspaceFinalization({ db, runId: seeded.runId });
+    try {
+      await started;
+      const second = resumeNativeWorkspaceFinalization({ db, runId: seeded.runId });
+      // A second invocation must return busy without waiting for or executing the slow I/O.
+      const outcome = await Promise.race([second, new Promise((resolve) => setTimeout(() => resolve("still-running"), 1_000))]);
+      release();
+      await Promise.all([first, second]);
+      expect(outcome).toBeNull();
+      expect(calls).toBe(1);
+      const operations = await db.select().from(workspaceOperations).where(and(
+        eq(workspaceOperations.heartbeatRunId, seeded.runId), eq(workspaceOperations.status, "succeeded"),
+      ));
+      expect(operations).toHaveLength(1);
+    } finally { release(); await first; spy.mockRestore(); }
+  });
+
   it("records directory-only recovery without inventing an execution-workspace foreign key", async () => {
     const cwd = path.join(workspaceRoot, "directory-only");
     await fs.mkdir(cwd);
@@ -195,7 +327,7 @@ describe("native workspace finalization recovery", () => {
       cwd,
       status: "succeeded",
     });
-    expect(operation.metadata).toMatchObject({
+    expect(operation?.metadata).toMatchObject({
       owningService: "native_workspace_finalizer",
       observation: "workspace_directory",
     });
@@ -301,15 +433,15 @@ describe("native workspace finalization recovery", () => {
       cwd: authorizedCwd,
       status: "succeeded",
     });
-    expect(operation.cwd).not.toBe(foreignCwd);
-    expect(operation.cwd).not.toBe(mismatchedCwd);
+    expect(operation!.cwd).not.toBe(foreignCwd);
+    expect(operation!.cwd).not.toBe(mismatchedCwd);
 
     const persisted = await db
       .select()
       .from(workspaceOperations)
       .where(
         and(
-          eq(workspaceOperations.id, operation.id),
+          eq(workspaceOperations.id, operation!.id),
           eq(workspaceOperations.companyId, companyId),
           eq(workspaceOperations.issueId, seeded.issueId),
         ),

server/src/services/native-runtime/native-workspace-finalizer.ts

@@ -1,3 +1,4 @@
+import { withNativeWorkspaceFinalizationOwnership } from "./native-workspace-finalization-ownership.js";
 import fs from "node:fs/promises";
 import { and, desc, eq } from "drizzle-orm";
 import type { Db } from "@paperclipai/db";
@@ -72,6 +73,18 @@ export async function resumeNativeWorkspaceFinalization(input: {
     throw new Error("native_workspace_finalization_binding_missing");
   }
 
+  const owned = await withNativeWorkspaceFinalizationOwnership({
+    db: input.db, companyId: bound.companyId, runId: input.runId,
+  }, async (ownership) => {
+  const successful = await input.db.select().from(workspaceOperations).where(and(
+    eq(workspaceOperations.companyId, bound.companyId),
+    eq(workspaceOperations.heartbeatRunId, input.runId),
+    eq(workspaceOperations.issueId, bound.issueId),
+    eq(workspaceOperations.phase, "workspace_finalize"),
+    eq(workspaceOperations.status, "succeeded"),
+  )).orderBy(desc(workspaceOperations.createdAt)).limit(1).then((rows) => rows[0] ?? null);
+  // A stale failure from a pre-fencing controller cannot invalidate exported work.
+  if (successful) return successful;
   const previous = await input.db.select().from(workspaceOperations).where(and(
     eq(workspaceOperations.companyId, bound.companyId),
     eq(workspaceOperations.heartbeatRunId, input.runId),
@@ -117,6 +130,7 @@ export async function resumeNativeWorkspaceFinalization(input: {
         : "workspace_directory",
     },
     run: async () => {
+      await ownership.assertHeld();
       if (nativeWorkspaceSync) {
         if (!input.environmentRuntime) {
           return {
@@ -172,7 +186,9 @@ export async function resumeNativeWorkspaceFinalization(input: {
             db: input.db,
             runId: input.runId,
             target,
+            assertOwnership: ownership.assertHeld,
           });
+          await ownership.assertHeld();
           if (!restored) {
             return workspaceSyncFailure("workspace_sync_out_unrecoverable");
           }
@@ -190,6 +206,7 @@ export async function resumeNativeWorkspaceFinalization(input: {
             },
           };
         } catch (error) {
+          await ownership.assertHeld();
           const { code } = classifyNativeWorkspaceFailure(error);
           return workspaceSyncFailure(code);
         }
@@ -202,6 +219,7 @@ export async function resumeNativeWorkspaceFinalization(input: {
         };
       }
       const isDirectory = await fs.stat(cwd).then((stat) => stat.isDirectory()).catch(() => false);
+      await ownership.assertHeld();
       if (!isDirectory) {
         return {
           status: "failed",
@@ -237,4 +255,6 @@ export async function resumeNativeWorkspaceFinalization(input: {
       };
     },
   });
+  });
+  return owned.acquired ? owned.value : null;
 }

server/src/services/native-runtime/native-workspace-sync.ts

@@ -87,7 +87,7 @@ export interface NativeWorkspaceSyncReference {
 export interface PreparedNativeWorkspaceSync {
   mode: WorkspaceInboundMode;
   reference: NativeWorkspaceSyncReference;
-  restoreWorkspace(): Promise<void>;
+  restoreWorkspace(assertOwnership?: () => Promise<void>): Promise<void>;
   cleanup(): Promise<void>;
 }
 
@@ -772,8 +772,10 @@ async function finalizePreparedRuntime(input: {
   target: Extract<AdapterExecutionTarget, { transport: "sandbox" }>;
   runtime: PreparedAdapterExecutionTargetRuntime;
   descriptor: NativeWorkspaceSyncDescriptor;
+  assertOwnership?: () => Promise<void>;
 }): Promise<NativeWorkspaceSyncReference> {
   await input.runtime.restoreWorkspace();
+  await input.assertOwnership?.();
   const finalSnapshot =
     await import("@paperclipai/adapter-utils/workspace-restore-merge").then(
       ({ captureDirectorySnapshot }) =>
@@ -796,6 +798,7 @@ async function finalizePreparedRuntime(input: {
     finalizedAt: new Date().toISOString(),
     finalHostSha256,
   };
+  await input.assertOwnership?.();
   const reference = await writeDescriptor(finalizedDescriptor);
   await persistRunReference(input.db, input.runId, reference);
   await persistLeaseStamp({
@@ -993,7 +996,7 @@ export async function prepareNativeWorkspaceSync(input: {
     get reference() {
       return reference;
     },
-    restoreWorkspace: async () => {
+    restoreWorkspace: async (assertOwnership) => {
       if (!restorePromise) {
         restorePromise = finalizePreparedRuntime({
           db: input.db,
@@ -1001,6 +1004,7 @@ export async function prepareNativeWorkspaceSync(input: {
           target,
           runtime: preparedRuntime,
           descriptor,
+          assertOwnership,
         })
           .then((finalizedReference) => {
             reference = finalizedReference;
@@ -1025,6 +1029,7 @@ export async function resumeNativeWorkspaceSync(input: {
   db: Db;
   runId: string;
   target: AdapterExecutionTarget;
+  assertOwnership?: () => Promise<void>;
 }): Promise<boolean> {
   if (input.target.kind !== "remote" || input.target.transport !== "sandbox") {
     throw new Error("workspace_sync_out_unrecoverable");
@@ -1049,6 +1054,7 @@ export async function resumeNativeWorkspaceSync(input: {
   ) {
     throw new Error("workspace_sync_out_unrecoverable");
   }
+  await input.assertOwnership?.();
   if (
     existing.descriptor.state === "finalized" &&
     existing.descriptor.finalHostSha256
@@ -1083,6 +1089,7 @@ export async function resumeNativeWorkspaceSync(input: {
     target: input.target,
     runtime,
     descriptor: existing.descriptor,
+    assertOwnership: input.assertOwnership,
   });
   return true;
 }