fix(server): wake reassessment after workspace ownership releases

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

server/src/__tests__/native-status-arbiter-corpus.test.ts

@@ -1,3 +1,4 @@
+import { withNativeWorkspaceFinalizationOwnership } from "../services/native-runtime/native-workspace-finalization-ownership.js";
 import { ensureNativeCompletionContract } from "../services/native-runtime/completion-contracts.js";
 import { nativeStatusReassessmentWork } from "../services/native-runtime/native-status-reassessment-work.js";
 import { reassessNativeStatusFromSource } from "../services/native-runtime/native-status-reassessment.js";
@@ -2454,7 +2455,7 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
     return seeded;
   }
 
-  it.each(["future", "claim", "unknown claim", "assessment", "result", "phase", "decision", "effects", "ownership", "run outcome", "task execution", "contract revision", "contract writer", "workspace status", "workspace operation", "workspace writer"] as const)(
+  it.each(["future", "claim", "unknown claim", "assessment", "result", "phase", "decision", "effects", "ownership", "run outcome", "task execution", "contract revision", "contract writer", "workspace status", "workspace operation", "workspace writer", "workspace owner", "workspace owner receipt"] as const)(
     "guards scheduled reassessment against a changed %s at the commit boundary", async changed => {
       const seeded = await seedAutomaticReview();
       await dismissAutomaticCompletionReviews(db, seeded.issueId);
@@ -2515,6 +2516,20 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
           phase: "workspace_finalize", status: "running", cwd: process.cwd(),
           createdAt: new Date(Math.max(Date.now(), originalBarrier.createdAt.getTime() + 1)),
         });
+        if (changed === "workspace owner receipt") await db.update(heartbeatRuns).set({
+          runnerProfileJson: { ...originalRun.runnerProfileJson, nativeWorkspaceFinalizationOwner: {
+            token: randomUUID(), hostname: "unverified-controller", pid: 12345, processStartedAt: new Date().toISOString(),
+          } },
+        }).where(eq(heartbeatRuns.id, seeded.runId));
+        if (changed === "workspace owner") {
+          let entered!: () => void;
+          const ready = new Promise<void>(resolve => { entered = resolve; });
+          const released = new Promise<void>(resolve => { releaseWriter = resolve; });
+          writer = withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async () => {
+            entered(); await released;
+          }).then(() => undefined);
+          await Promise.race([ready, writer]);
+        }
         if (changed === "contract revision") await writeContract(db);
         if (changed === "contract writer" || changed === "workspace writer") {
           let writerReady!: () => void;
@@ -2577,6 +2592,11 @@ describe("P6-31 Section 18.13 executable status-authority corpus", () => {
       }
       if (changed === "workspace writer") expect((await db.select().from(workspaceOperations)
         .where(eq(workspaceOperations.id, originalBarrier.id)))[0].status).toBe(originalBarrier.status);
+      if (changed === "workspace owner receipt") {
+        expect(await reassessNativeStatusFromSource(db, selected, { onlyWhenDue: true })).toEqual({ kind: "deferred" });
+        await db.update(heartbeatRuns).set({ runnerProfileJson: originalRun.runnerProfileJson })
+          .where(eq(heartbeatRuns.id, seeded.runId));
+      }
       await db.update(nativeRunFinalizations).set({ phase: selected.coordinatorPhase,
         resultId: seeded.resultId, assessmentId: selected.assessmentId, decisionId: selected.decisionId,
         nextAttemptAt: null, leaseOwner: null, leaseExpiresAt: null }).where(eq(nativeRunFinalizations.runId, seeded.runId));

server/src/__tests__/native-status-reassessment-work.test.ts

@@ -1,3 +1,4 @@
+import { withNativeWorkspaceFinalizationOwnership } from "../services/native-runtime/native-workspace-finalization-ownership.js";
 import express from "express";
 import { appendHeartbeatRunEvent } from "../services/heartbeat-run-events.js";
 import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../services/delivery-work-notifications.js";
@@ -124,6 +125,40 @@ describe("native reassessment worker lifecycle", () => {
     return { issueId, runId, workProductId, decisionId: committed.decision.id };
   }
 
+  it("waits quietly for physical workspace ownership to release before reassessment", async () => {
+    const seeded = await seedReview();
+    await dismissAutomaticCompletionReviews(db, seeded.issueId);
+    let enter!: () => void, release!: () => void;
+    const entered = new Promise<void>(resolve => { enter = resolve; });
+    const released = new Promise<void>(resolve => { release = resolve; });
+    const owner = withNativeWorkspaceFinalizationOwnership({ db, companyId, runId: seeded.runId }, async () => {
+      enter(); await released;
+    });
+    try {
+      await Promise.race([entered, owner]);
+      await start(); await flush();
+      expect(await status(seeded.issueId)).toBe("in_review");
+      await quiet();
+    } finally { release(); await owner; }
+    await flush();
+    expect(await status(seeded.issueId)).toBe("done");
+    await quiet();
+  });
+
+  it("keeps an unverified workspace owner dormant despite a successful receipt", async () => {
+    const seeded = await seedReview();
+    await dismissAutomaticCompletionReviews(db, seeded.issueId);
+    const owner = { token: randomUUID(), hostname: "unverified-controller", pid: 12345,
+      processStartedAt: new Date().toISOString() };
+    await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeWorkspaceFinalizationOwner: owner } })
+      .where(eq(heartbeatRuns.id, seeded.runId));
+    await start(); await flush();
+    expect(await status(seeded.issueId)).toBe("in_review");
+    expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, seeded.runId)))[0]
+      .runnerProfileJson?.nativeWorkspaceFinalizationOwner).toEqual(owner);
+    await quiet();
+  });
+
   it.each(["ordinary", "native source"] as const)("wakes from a committed control-plane evidence event through the %s append path", async path => {
     const seeded = await seedReview("event:1");
     await vi.advanceTimersByTimeAsync(1000);

server/src/services/native-runtime/native-status-reassessment.ts

@@ -1,3 +1,4 @@
+import { hasNativeWorkspaceFinalizationOwner } from "./native-workspace-finalization-recovery-state.js";
 import { and, desc, eq, gt, or } from "drizzle-orm";
 import { completionContracts, heartbeatRuns, issueWorkProducts, issues, nativeRunResults,
   statusDecisions, workAssessments, workspaceOperations, type Db } from "@paperclipai/db";
@@ -148,6 +149,7 @@ export async function reassessNativeStatusFromSource(
       .where(and(eq(completionContracts.companyId, row.companyId), eq(completionContracts.issueId, row.issueId)))
       .orderBy(desc(completionContracts.revision)).limit(1).then((entries) => entries[0]) : null;
     if (retiredAutomaticReview && (!currentIssue || currentRun?.status !== "succeeded"
+      || (options.onlyWhenDue && hasNativeWorkspaceFinalizationOwner(currentRun.runnerProfileJson))
       || latestContract?.id !== contractRow.id
       || (currentIssue.executionRunId && currentIssue.executionRunId !== row.runId))) return { kind: "deferred" };
     const reviewBarrier = retiredAutomaticReview ? await db.select({ status: workspaceOperations.status })

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

@@ -1,3 +1,4 @@
+import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
 import { beginRunPhase } from "../execution-preparation-timing.js";
 import { logActivity } from "../activity-log.js";
 import { issueRecoveryActionService } from "../issue-recovery-actions.js";
@@ -21,6 +22,12 @@ const OWNER_KEY = "nativeWorkspaceFinalizationOwner";
 export function nativeWorkspaceFinalizationLockKey(companyId: string, runId: string): string {
   return `native-workspace-finalization:${companyId}:${runId}`;
 }
+export async function tryLockNativeWorkspaceFinalization(tx: Db, input: { companyId: string; runId: string }): Promise<boolean> {
+  const [lock] = await tx.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(
+    ${nativeWorkspaceFinalizationLockKey(input.companyId, input.runId)}, 0)) as acquired`) as unknown as Array<{ acquired: boolean }>;
+  return lock.acquired;
+}
+
 const activeOwnerTokens = new Set<string>();
 // Positive same-boot evidence, retained only when durable cleanup could not be
 // confirmed. Absence from the active set alone never proves another token joined.
@@ -258,9 +265,12 @@ export async function withNativeWorkspaceFinalizationOwnership<T>(
       activeOwnerTokens.delete(owner.token);
       joinedOwnerTokens.add(owner.token);
       try {
-        await input.db.update(heartbeatRuns).set({
-          runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`,
-        }).where(owned);
+        await input.db.transaction(async tx => {
+          await notifyNativeStatusReassessment(tx, { companyId: input.companyId, runId: input.runId });
+          await tx.update(heartbeatRuns).set({
+            runnerProfileJson: sql`coalesce(${heartbeatRuns.runnerProfileJson}, '{}'::jsonb) - ${OWNER_KEY}`,
+          }).where(owned);
+        });
         joinedOwnerTokens.delete(owner.token);
       } catch {
         // A reconnect in this exact controller boot can use the positive join

server/src/services/native-runtime/native-workspace-finalization-recovery-state.ts

@@ -5,6 +5,10 @@ import { nativeSha256 } from "./canonical.js";
 
 export const NATIVE_WORKSPACE_OWNER_UNVERIFIED = "native_workspace_finalization_owner_unverified";
 
+export function hasNativeWorkspaceFinalizationOwner(profile: Record<string, unknown> | null | undefined): boolean {
+  return profile?.nativeWorkspaceFinalizationOwner != null;
+}
+
 /** A terminal task does not prove that an older controller's copyback stopped. */
 export async function hasUnverifiedNativeWorkspaceFinalizationOwner(
   db: Db,
@@ -20,5 +24,5 @@ export async function hasUnverifiedNativeWorkspaceFinalizationOwner(
     eq(heartbeatRuns.runtimeMode, "native"),
   )).limit(1);
   const owner = run?.profile?.nativeWorkspaceFinalizationOwner;
-  return owner != null && action.fingerprint === nativeSha256({ runId, owner });
+  return hasNativeWorkspaceFinalizationOwner(run?.profile) && action.fingerprint === nativeSha256({ runId, owner });
 }

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

@@ -1,3 +1,5 @@
+import { DELIVERY_QUEUES, subscribeDeliveryWork } from "../delivery-work-notifications.js";
+import { resolveExternalTestDatabaseUrl } from "../../__tests__/helpers/external-test-database.js";
 import * as pluginWorkspaces from "../plugin-workspaces.js";
 import * as platformProof from "./native-workspace-platform-proof.js";
 import { currentNativeControllerIdentity } from "./native-restart-recovery.js";
@@ -38,7 +40,8 @@ import { resumeNativeWorkspaceFinalization } from "./native-workspace-finalizer.
 import { reconcileNativeFinalizations } from "./native-finalization-reconciler.js";
 
 describe("native workspace finalization recovery", () => {
-  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let connectionString: string;
   let db: ReturnType<typeof createDb>;
   let workspaceRoot: string;
   let priorLogRoot: string | undefined;
@@ -125,10 +128,10 @@ describe("native workspace finalization recovery", () => {
       workspaceRoot,
       "operation-logs",
     );
-    temporary = await startEmbeddedPostgresTestDatabase(
-      "paperclip-native-workspace-finalizer-",
-    );
-    db = createDb(temporary.connectionString);
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_NATIVE_WORKSPACE_FINALIZER_TEST_DATABASE_URL");
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-native-workspace-finalizer-");
+    connectionString = external ?? temporary!.connectionString;
+    db = createDb(connectionString);
 
     await db.insert(companies).values([
       { id: companyId, name: "Native workspace owner", issuePrefix: "NWO" },
@@ -162,7 +165,8 @@ describe("native workspace finalization recovery", () => {
   }, 30_000);
 
   afterAll(async () => {
-    await temporary.cleanup();
+    await db?.$client.end({ timeout: 1 });
+    await temporary?.cleanup();
     await fs.rm(workspaceRoot, { recursive: true, force: true });
     if (priorLogRoot === undefined) {
       delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
@@ -210,7 +214,17 @@ describe("native workspace finalization recovery", () => {
     const seed = await seedAbandonedCopyback();
     const [before] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId));
     expect(await withNativeWorkspaceFinalizationOwnership(seed, async () => "unexpected")).toEqual({ acquired: false });
-    const receipt = await resumeNativeWorkspaceAfterOwnerStop(seed);
+    const observed: Array<Promise<unknown>> = [];
+    const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeStatusReassessment, () => {
+      observed.push(db.select({ profile: heartbeatRuns.runnerProfileJson }).from(heartbeatRuns)
+        .where(eq(heartbeatRuns.id, seed.runId)).then(rows => rows[0].profile));
+    });
+    let receipt: Awaited<ReturnType<typeof resumeNativeWorkspaceAfterOwnerStop>>;
+    try {
+      receipt = await resumeNativeWorkspaceAfterOwnerStop(seed);
+      await vi.waitFor(() => expect(observed).toHaveLength(1));
+      expect(await Promise.all(observed)).toEqual([{ preserved: "workspace descriptor" }]);
+    } finally { stop(); }
     expect(receipt).toEqual({ runId: seed.runId, resultId: before.resultId, status: "queued" });
     const [after] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, seed.runId));
     expect(after).toEqual(before);
@@ -425,7 +439,7 @@ describe("native workspace finalization recovery", () => {
 
   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 smallPool = createDb(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));
@@ -438,7 +452,7 @@ describe("native workspace finalization recovery", () => {
     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 resumeNativeWorkspaceFinalization({ db: createDb(connectionString), runId: seeded.runId })).toBeNull();
       expect(await db.select().from(workspaceOperations).where(eq(workspaceOperations.heartbeatRunId, seeded.runId))).toEqual([]);
       return "copied";
     });
@@ -462,7 +476,7 @@ describe("native workspace finalization recovery", () => {
     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");
+      const second = await withNativeWorkspaceFinalizationOwnership({ db: createDb(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();
@@ -475,7 +489,7 @@ describe("native workspace finalization recovery", () => {
 
   it("recovers its own joined copyback receipt after the application pool disconnects during cleanup", async () => {
     const seeded = await seedRun({ executionWorkspaceId: randomUUID(), title: "Reconnect joined copyback" });
-    const interruptedDb = createDb(temporary.connectionString);
+    const interruptedDb = createDb(connectionString);
     let joined = false;
     await expect(withNativeWorkspaceFinalizationOwnership({ db: interruptedDb, companyId, runId: seeded.runId }, async () => {
       await fs.writeFile(path.join(workspaceRoot, "joined-copyback.txt"), "durable work");
@@ -493,7 +507,7 @@ describe("native workspace finalization recovery", () => {
         .toEqual({ acquired: false });
     }
     await db.update(heartbeatRuns).set({ runnerProfileJson: beforeReconnect.runnerProfileJson }).where(eq(heartbeatRuns.id, seeded.runId));
-    expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(temporary.connectionString), companyId, runId: seeded.runId }, async () => "resumed"))
+    expect(await withNativeWorkspaceFinalizationOwnership({ db: createDb(connectionString), companyId, runId: seeded.runId }, async () => "resumed"))
       .toEqual({ acquired: true, value: "resumed" });
   });
 

server/src/services/native-runtime/native-workspace-owner-recovery.ts

@@ -1,3 +1,5 @@
+import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
+import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js";
 import os from "node:os";
 import { and, eq, sql } from "drizzle-orm";
 import { heartbeatRuns, issues, issueRecoveryActions, nativeRunFinalizations, type Db } from "@paperclipai/db";
@@ -24,8 +26,7 @@ export async function resumeNativeWorkspaceAfterOwnerStop(input: {
   const admitted = await input.db.transaction(async tx => {
     // Share the physical copyback lock. Do not wait behind an active exporter
     // and then apply a confirmation that described an earlier owner.
-    const locks = await tx.execute(sql`select pg_try_advisory_xact_lock(hashtextextended(${`native-workspace-finalization:${input.companyId}:${input.runId}`}, 0)) as acquired`);
-    if (locks[0]?.acquired !== true) throw conflict("Workspace copyback is still active. Wait for it to stop before confirming recovery.");
+    if (!await tryLockNativeWorkspaceFinalization(tx as unknown as Db, input)) throw conflict("Workspace copyback is still active. Wait for it to stop before confirming recovery.");
     // Match the status arbiter's coordinator -> issue -> run lock order.
     const [coordinator] = await tx.select().from(nativeRunFinalizations).where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId), eq(nativeRunFinalizations.runId, input.runId))).for("update");
     const [issue] = await tx.select().from(issues).where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update");
@@ -65,6 +66,7 @@ export async function resumeNativeWorkspaceAfterOwnerStop(input: {
     const now = new Date();
     const proof = { owner: prior, resultId: result.id, actorId: input.actorId,
       stopEvidence: confirmation.stopEvidence, confirmedAt: now.toISOString() };
+    await notifyNativeStatusReassessment(tx, { companyId: input.companyId, issueId: input.issueId, runId: input.runId });
     await tx.update(heartbeatRuns).set({
       runnerProfileJson: sql`${heartbeatRuns.runnerProfileJson} - ${OWNER_KEY}`, updatedAt: now,
     }).where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, run.id),

server/src/services/native-runtime/status-decision-committer.ts

@@ -1,3 +1,5 @@
+import { hasNativeWorkspaceFinalizationOwner } from "./native-workspace-finalization-recovery-state.js";
+import { tryLockNativeWorkspaceFinalization } from "./native-workspace-finalization-ownership.js";
 import { lockCurrentNativeCompletionContract } from "./completion-contracts.js";
 import { isNativeRunnerOwnershipHeld } from "./native-runner-ownership.js";
 import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js";
@@ -1632,6 +1634,7 @@ export async function commitNativeStatusDecision(input: {
     if (retiredReview && !await lockCurrentNativeCompletionContract(tx as unknown as Db, {
       companyId: input.companyId, issueId: input.issueId, contractId: retiredReview.contractId,
     })) throw new NativeStatusRaceError();
+    if (retiredReview && !await tryLockNativeWorkspaceFinalization(tx as unknown as Db, input)) throw new NativeStatusRaceError();
     const coordinator = await tx
       .select()
       .from(nativeRunFinalizations)
@@ -1677,6 +1680,7 @@ export async function commitNativeStatusDecision(input: {
       )).for("update").limit(1).then(rows => rows[0]);
       if (!run || run.runtimeMode !== "native" || isNativeRunnerOwnershipHeld(run)) throw new NativeStatusRaceError();
       if (expected.retiredAutomaticReview && (run.status !== "succeeded" ||
+          hasNativeWorkspaceFinalizationOwner(run.runnerProfileJson) ||
           (issue.executionRunId !== null && issue.executionRunId !== input.runId))) {
         throw new NativeStatusRaceError();
       }