fix(server): wake native reassessment when workspace barriers settle

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

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

@@ -1,4 +1,8 @@
 import express from "express";
+import { mkdtemp, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { workspaceOperationService } from "../services/workspace-operations.js";
 import request from "supertest";
 import { LOW_TRUST_REVIEW_PRESET } from "@paperclipai/shared";
 import { issueRoutes } from "../routes/issues.js";
@@ -28,16 +32,25 @@ describe("native reassessment worker lifecycle", () => {
   let db: ReturnType<typeof createDb>;
   let coordinator: ReturnType<typeof createDeliveryWorkCoordinator> | undefined;
   const companyId = randomUUID(), agentId = randomUUID();
+  let logRoot: string;
+  const priorLogRoot = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
   let admitted = true;
   const errors: unknown[] = [];
   beforeAll(async () => {
+    logRoot = await mkdtemp(join(tmpdir(), "reassessment-workspace-logs-"));
+    process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = logRoot;
     const external = resolveExternalTestDatabaseUrl("PAPERCLIP_STATUS_REASSESSMENT_TEST_DATABASE_URL");
     if (!external) temporary = await startEmbeddedPostgresTestDatabase("paperclip-status-reassessment-");
     db = createDb(external ?? temporary!.connectionString);
     await db.insert(companies).values({ id: companyId, name: "Reassessment lifecycle", issuePrefix: `RS${randomUUID().slice(0, 8)}` });
     await db.insert(agents).values({ id: agentId, companyId, name: "Reassessment agent", adapterType: "codex_local", status: "idle" });
   }, 30_000);
-  afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); });
+  afterAll(async () => {
+    await db?.$client.end({ timeout: 1 }); await temporary?.cleanup();
+    await rm(logRoot, { recursive: true, force: true });
+    if (priorLogRoot === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
+    else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = priorLogRoot;
+  });
   beforeEach(() => {
     admitted = true; errors.length = 0;
     vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
@@ -109,6 +122,55 @@ describe("native reassessment worker lifecycle", () => {
     return { issueId, runId, workProductId, decisionId: committed.decision.id };
   }
 
+  it("wakes when a real workspace finalization settles after a quiet failure", async () => {
+    const seeded = await seedReview();
+    await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.heartbeatRunId, seeded.runId));
+    await dismissAutomaticCompletionReviews(db, seeded.issueId);
+    await start(); await flush(); await quiet();
+    const recorder = workspaceOperationService(db).createRecorder({ companyId, issueId: seeded.issueId, heartbeatRunId: seeded.runId });
+    await expect(recorder.recordOperation({ phase: "workspace_finalize", run: async () => { throw new Error("export interrupted"); } }))
+      .rejects.toThrow("export interrupted");
+    await flush(); await quiet();
+    expect(await status(seeded.issueId)).toBe("in_review");
+    let release!: () => void;
+    const gate = new Promise<void>(resolve => { release = resolve; });
+    let started!: () => void;
+    const running = new Promise<void>(resolve => { started = resolve; });
+    const operation = recorder.recordOperation({ phase: "workspace_finalize", run: async () => {
+      started(); await gate; return { status: "succeeded" };
+    } });
+    try {
+      await running; await flush(); await quiet();
+      expect(await status(seeded.issueId)).toBe("in_review");
+    } finally { release(); await operation; }
+    await flush();
+    expect(await status(seeded.issueId)).toBe("done");
+    expect(await db.select().from(statusDecisions).where(eq(statusDecisions.issueId, seeded.issueId))).toHaveLength(2);
+    await quiet();
+  });
+
+  it("defers workspace barrier writes until outer commit and ignores rollback", async () => {
+    const seeded = await seedReview();
+    await db.update(workspaceOperations).set({ status: "running" }).where(eq(workspaceOperations.heartbeatRunId, seeded.runId));
+    await dismissAutomaticCompletionReviews(db, seeded.issueId);
+    await start(); await flush(); await quiet();
+    const write = async (tx: typeof db) => {
+      await workspaceOperationService(tx).createRecorder({ companyId, issueId: seeded.issueId, heartbeatRunId: seeded.runId })
+        .recordOperation({ phase: "workspace_finalize", run: async () => ({ status: "succeeded" }) });
+      await vi.advanceTimersByTimeAsync(1000);
+      expect(await status(seeded.issueId)).toBe("in_review");
+    };
+    await expect(db.transaction(async tx => {
+      await write(tx as unknown as typeof db); throw new Error("rollback workspace barrier");
+    })).rejects.toThrow("rollback workspace barrier");
+    await flush(); await quiet();
+    expect(await status(seeded.issueId)).toBe("in_review");
+    await db.transaction(tx => write(tx as unknown as typeof db));
+    await flush();
+    expect(await status(seeded.issueId)).toBe("done");
+    await quiet();
+  });
+
   it.each(["confirmation", "checkbox", "question", "verdict", "partial verdict"] as const)("wakes dormant evidence from real interaction %s", async mutation => {
     const interactionId = randomUUID();
     const seeded = await seedReview(`interaction:${interactionId}`);

server/src/__tests__/workspace-operations-reconciliation.test.ts

@@ -1,3 +1,4 @@
+import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js";
 import { randomUUID } from "node:crypto";
 import fs from "node:fs/promises";
 import os from "node:os";
@@ -65,7 +66,8 @@ describe("workspace runtime control serialization", () => {
   });
 });
 
-const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
+const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_WORKSPACE_OPERATIONS_TEST_DATABASE_URL");
+const embeddedPostgresSupport = externalUrl ? { supported: true } : await getEmbeddedPostgresTestSupport();
 const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
 
 if (!embeddedPostgresSupport.supported) {
@@ -81,8 +83,8 @@ describeEmbeddedPostgres("workspace operation reconciliation", () => {
   let previousLogRoot: string | undefined;
 
   beforeAll(async () => {
-    tempDb = await startEmbeddedPostgresTestDatabase("paperclip-workspace-operation-reconcile-");
-    db = createDb(tempDb.connectionString);
+    if (!externalUrl) tempDb = await startEmbeddedPostgresTestDatabase("paperclip-workspace-operation-reconcile-");
+    db = createDb(externalUrl ?? tempDb!.connectionString);
     logRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-workspace-operation-logs-"));
     previousLogRoot = process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
     process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = logRoot;
@@ -99,6 +101,7 @@ describeEmbeddedPostgres("workspace operation reconciliation", () => {
     if (previousLogRoot === undefined) delete process.env.WORKSPACE_OPERATION_LOG_BASE_PATH;
     else process.env.WORKSPACE_OPERATION_LOG_BASE_PATH = previousLogRoot;
     await fs.rm(logRoot, { recursive: true, force: true });
+    await db?.$client.end({ timeout: 1 });
     await tempDb?.cleanup();
   });
 

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

@@ -92,11 +92,11 @@ async function recoverCandidate(db: Db, candidate: { runId: string; companyId: s
         await ownership.assertHeld();
         // Publish the omission barrier atomically with re-admission. This is
         // explicitly not a receipt claiming that files were copied to the host.
+        await notifyNativeFinalizationWork(tx, { companyId: run.companyId, runId: run.id });
         await tx.insert(workspaceOperations).values({ companyId: run.companyId, heartbeatRunId: run.id, issueId: issue.id,
           phase: "workspace_finalize", status: "succeeded", exitCode: 0, finishedAt: now,
           metadata: { owningService: "native_workspace_finalizer", workspaceSync: { omitted: true, reason: "restore_unsafe_archive", legacy: true } },
         });
-        await notifyNativeFinalizationWork(tx, { companyId: run.companyId, runId: run.id });
         await tx.update(nativeRunFinalizations).set({ phase: "result_accepted", failureCode: null, nextAttemptAt: null,
           failureDetail: { workspaceFinalizeAttempt: 0, legacyUnsafeExportOmitted: { resultId: result.id, recoveredAt: now.toISOString() } }, updatedAt: now,
         }).where(eq(nativeRunFinalizations.runId, run.id));

server/src/services/workspace-operations.ts

@@ -1,3 +1,4 @@
+import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 import { randomUUID } from "node:crypto";
 import type { Db } from "@paperclipai/db";
 import { workspaceOperations } from "@paperclipai/db";
@@ -476,6 +477,14 @@ export function workspaceOperationService(db: Db) {
         },
 
         async recordOperation(recordInput) {
+          async function writeState<T>(write: (database: Db) => Promise<T>): Promise<T> {
+            if (recordInput.phase !== "workspace_finalize" || !input.heartbeatRunId) return write(db);
+            return db.transaction(async tx => {
+              await notifyNativeStatusReassessment(tx, { companyId: input.companyId,
+                runId: input.heartbeatRunId!, issueId: input.issueId ?? undefined });
+              return write(tx as unknown as Db);
+            });
+          }
           const currentUserRedactionOptions = {
             enabled: (await instanceSettings.getGeneral()).censorUsernameInLogs,
           };
@@ -521,7 +530,7 @@ export function workspaceOperationService(db: Db) {
           if (runtimeControlAction) liveRuntimeControlOperationIds.add(id);
 
           try {
-            await db.insert(workspaceOperations).values({
+            await writeState(database => database.insert(workspaceOperations).values({
               id,
               companyId: input.companyId,
               executionWorkspaceId,
@@ -539,7 +548,7 @@ export function workspaceOperationService(db: Db) {
               ) as Record<string, unknown> | null,
               startedAt,
               updatedAt: startedAt,
-            });
+            }));
           } catch (insertError) {
             liveRuntimeControlOperationIds.delete(id);
             throw insertError;
@@ -630,7 +639,7 @@ export function workspaceOperationService(db: Db) {
             await append("stderr", result.stderr ?? null);
             const finalized = await logStore.finalize(handle);
             const finishedAt = new Date();
-            const row = await db
+            const row = await writeState(database => database
               .update(workspaceOperations)
               .set({
                 executionWorkspaceId,
@@ -650,7 +659,7 @@ export function workspaceOperationService(db: Db) {
               })
               .where(eq(workspaceOperations.id, id))
               .returning()
-              .then((rows) => rows[0] ?? null);
+              .then((rows) => rows[0] ?? null));
             if (!row) throw notFound("Workspace operation not found");
             return toWorkspaceOperation(row);
           } catch (error) {
@@ -660,7 +669,7 @@ export function workspaceOperationService(db: Db) {
             await append("stderr", error instanceof Error ? error.message : String(error)).catch(() => undefined);
             const finalized = await logStore.finalize(handle).catch(() => null);
             const finishedAt = new Date();
-            await db
+            await writeState(database => database
               .update(workspaceOperations)
               .set({
                 executionWorkspaceId,
@@ -687,7 +696,7 @@ export function workspaceOperationService(db: Db) {
                 finishedAt,
                 updatedAt: finishedAt,
               })
-              .where(eq(workspaceOperations.id, id));
+              .where(eq(workspaceOperations.id, id)));
             throw error;
           } finally {
             // Releasing the in-process claim before the request unwinds is what lets the very

server/src/services/workspace-recovery.test.ts

@@ -1,3 +1,5 @@
+import { resolveExternalTestDatabaseUrl } from "../__tests__/helpers/external-test-database.js";
+import { DELIVERY_QUEUES, subscribeDeliveryWork } from "./delivery-work-notifications.js";
 import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js";
 import { createHeartbeatWorkspaceResolver, resolveExecutionWorkspaceReuseRequestForIssue } from "./heartbeat/workspaces.js";
 import { environmentService } from "./environments.js";
@@ -44,14 +46,18 @@ vi.mock("./environment-execution-target.js", async importOriginal => {
 });
 
 describe("automatic retained workspace recovery", () => {
-  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>>;
+  let temporary: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | undefined;
+  let connectionString: string;
   let db: ReturnType<typeof createDb>;
   let root: string;
   const priorHome = process.env.PAPERCLIP_HOME, priorInstance = process.env.PAPERCLIP_INSTANCE_ID;
   beforeAll(async () => {
     root = await mkdtemp(path.join(os.tmpdir(), "workspace-recovery-"));
     process.env.PAPERCLIP_HOME = path.join(root, "home"); process.env.PAPERCLIP_INSTANCE_ID = "recovery";
-    temporary = await startEmbeddedPostgresTestDatabase("workspace-recovery-"); db = createDb(temporary.connectionString);
+    const external = resolveExternalTestDatabaseUrl("PAPERCLIP_WORKSPACE_RECOVERY_TEST_DATABASE_URL");
+    if (!external) temporary = await startEmbeddedPostgresTestDatabase("workspace-recovery-");
+    connectionString = external ?? temporary!.connectionString;
+    db = createDb(connectionString);
   }, 30_000);
   afterAll(async () => {
     await db?.$client.end(); await temporary?.cleanup(); await rm(root, { recursive: true, force: true });
@@ -153,6 +159,21 @@ describe("automatic retained workspace recovery", () => {
       systemRecovery: { ...state, nextAttemptAt: 0 } } }).where(eq(issueRecoveryActions.id, actionId));
   }
 
+  it("publishes the committed finalization barrier after real retained workspace recovery", async () => {
+    const f = await seed();
+    const observed: Array<Promise<Array<{ status: string }>>> = [];
+    const stop = subscribeDeliveryWork(db, DELIVERY_QUEUES.nativeStatusReassessment, () => {
+      observed.push(db.select({ status: workspaceOperations.status }).from(workspaceOperations).where(and(
+        eq(workspaceOperations.companyId, f.companyId), eq(workspaceOperations.heartbeatRunId, f.runId),
+        eq(workspaceOperations.phase, "workspace_finalize"),
+      )).then(rows => rows));
+    });
+    try { await recoverRetainedWorkspace(f.input); } finally { stop(); }
+    expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete");
+    expect(observed.length).toBeGreaterThan(0);
+    expect((await Promise.all(observed)).some(rows => rows.some(row => row.status === "succeeded"))).toBe(true);
+  });
+
   it("re-enters only the existing finalizer for an accepted result, once per intent, preserving failure history", async () => {
     const f = await seed();
     const contractId = randomUUID(), resultId = randomUUID();
@@ -257,7 +278,7 @@ describe("automatic retained workspace recovery", () => {
     expect(pending.status).toBe("running");
     expect(uploads).toBe(2);
     await makeRetryDue(f.actionId);
-    const restarted = createDb(temporary.connectionString);
+    const restarted = createDb(connectionString);
     try { await recoverRetainedWorkspace({ ...f.input, db: restarted }); }
     finally { await restarted.$client.end(); }
     const operations = await db.select().from(workspaceOperations).where(and(
@@ -466,7 +487,7 @@ describe("automatic retained workspace recovery", () => {
     const pending = await action(f.actionId);
     await db.update(issueRecoveryActions).set({ evidence: { ...pending.evidence,
       systemRecovery: { ...readSystemRecovery(pending.evidence), nextAttemptAt: 0 } } }).where(eq(issueRecoveryActions.id, f.actionId));
-    const restarted = createDb(temporary.connectionString);
+    const restarted = createDb(connectionString);
     try {
       await recoverRetainedWorkspace({ ...f.input, db: restarted });
       expect(await hasVerifiedWorkspaceRepair(restarted, await run(f.runId))).toBe(true);
@@ -796,7 +817,7 @@ describe("automatic retained workspace recovery", () => {
     }
     expect(await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, f.companyId))).toHaveLength(0);
     expect((await db.select().from(environmentLeases).where(eq(environmentLeases.id, f.lease.id)))[0].status).toBe("released");
-    const restarted = createDb(temporary.connectionString), onRecovered = vi.fn(async (_source: { id: string }) => {});
+    const restarted = createDb(connectionString), onRecovered = vi.fn(async (_source: { id: string }) => {});
     try {
       await expect.poll(async () => {
         await reconcileRetainedWorkspaces(restarted, f.input.environmentRuntime, onRecovered);

server/src/services/workspace-recovery.ts

@@ -1,3 +1,4 @@
+import { notifyNativeStatusReassessment } from "./native-status-reassessment-notifications.js";
 import { writeEnvironmentCleanup } from "./environment-cleanup-notifications.js";
 import { resumePreservedResultFinalization } from "./preserved-result-finalization.js";
 import { and, asc, eq, gt, inArray, ne, or, sql } from "drizzle-orm";
@@ -243,6 +244,7 @@ export async function recoverRetainedWorkspace(input: Source & { db: Db; environ
       const next = { ...current!, revision: current!.revision + 1, phase: "reconcile_effects" as const,
         state: "complete" as const, nextAttemptAt: null };
       const activity = await db.transaction(async tx => {
+        await notifyNativeStatusReassessment(tx, input);
         await tx.insert(workspaceOperations).values({ companyId: input.companyId, issueId: input.issueId,
           heartbeatRunId: input.runId, phase: "workspace_finalize", status: "succeeded", finishedAt: new Date(),
           metadata: { owningService: "workspace_recovery", preservationOperationIds: next.preservationOperationIds } });