From 1349eacaabc98f0c4d42a2f71c3f6fbe3ebb80c7 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 26 Oct 2026 03:15:03 -0500 Subject: [PATCH] fix(server): wake native reassessment when workspace barriers settle Co-Authored-By: Paperclip --- ...-native-status-reassessment-sleep-audit.md | 53 +++++++++++++++ .../native-status-reassessment-work.test.ts | 64 ++++++++++++++++++- ...orkspace-operations-reconciliation.test.ts | 9 ++- .../native-workspace-export-recovery.ts | 2 +- server/src/services/workspace-operations.ts | 21 ++++-- .../src/services/workspace-recovery.test.ts | 31 +++++++-- server/src/services/workspace-recovery.ts | 2 + 7 files changed, 166 insertions(+), 16 deletions(-) diff --git a/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md b/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md index 285f9e8d23f..5f8d871973e 100644 --- a/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md +++ b/doc/plans/2026-10-26-native-status-reassessment-sleep-audit.md @@ -359,3 +359,56 @@ proof that all native ownership/run-event writers are covered. Workspace barrier producers, remaining run/contract/case-handoff facts, concurrent authority guards, current-history/backlog query cost, startup registration and remote PR acceptance remain incomplete. Periodic reassessment is deliberately still enabled. + + +## Workspace finalization barrier producer checkpoint + +The retired-review path defers when the latest `workspace_finalize` record has +not succeeded. It deliberately leaves no recovery timer for an in-progress or +failed barrier, so the actual producer must reactivate it when that fact changes. +Two PostgreSQL regressions reproduced missing reactivation through the normal +workspace-operation recorder: failure followed by a later successful export, +and a successful write inside a caller-owned transaction. + +The recorder now routes its initial state insert, successful settlement and failed +settlement through one local state-write boundary. Only `workspace_finalize` +operations with a run binding register reassessment intent; other phases retain +their existing writes. The added transaction contains only the state write and +notification. Log I/O, the provider callback, and progress/lease updates remain +outside it. Existing caller-owned transactions retain outer commit semantics. +The success test keeps the real callback suspended while the worker goes SQL +quiet, then completes it and observes task completion. The transaction test +proves no decision before outer commit, no decision after rollback, and successful +activation after a later commit. + +The separate retained-workspace recovery success receipt now signals in its +existing transaction before the receipt insert and generation update. A test +runs real retained recovery with temporary files and verifies that the delivered +notification's reader observes a committed `workspace_finalize: succeeded` row. +The historical unsafe-export path already had the common finalization signal; +it now registers that intent before its barrier insert rather than afterward. +Snapshot and copy-only records use other phases and do not satisfy this barrier. +Runtime-control orphan reconciliation operates on action-stamped controls; normal +native workspace finalization uses its own ownership metadata instead. + +All 24 worker lifecycle cases pass, and server typecheck passes. The workspace +operation reconciliation, retained workspace recovery and native status arbiter +corpus suites pass 170 cases across three fresh PostgreSQL databases. The two +workspace suites now accept dedicated, validated external test database URLs; +restart tests reuse that URL, and clients close before fixture teardown. This +uses the known-working local PostgreSQL installation because embedded bootstrap +failed in earlier acceptance attempts; no product database is used. + +Evidence: `/tmp/sleep50-workspace-before.log` and `.status` (two failures), +`/tmp/sleep50-workspace-after.log` and `.status` (24 passes), +`/tmp/sleep50-workspace-consumers.log` and `.status` (170 passes), and +`/tmp/sleep50-workspace-final-types.log` and `.status` (pass). All launched +commands at this checkpoint reached terminal status. These are local checks, +not remote PR acceptance or evidence of host suspension. + +Next: finish run-event/case-handoff, completion-contract, ownership and remaining +interaction producer audit; then verify commit-time business authority, discovery +cost/backlog behavior, lifecycle wiring and remote PR gates. Current scheduled +commit guards fence coordinator/run ownership and source selection, but the +retired-review business-fact checks still run before the commit transaction. +The old periodic reassessment remains enabled until this full slice is ready. diff --git a/server/src/__tests__/native-status-reassessment-work.test.ts b/server/src/__tests__/native-status-reassessment-work.test.ts index 27958e17aa1..4ba8b7e1b47 100644 --- a/server/src/__tests__/native-status-reassessment-work.test.ts +++ b/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; let coordinator: ReturnType | 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(resolve => { release = resolve; }); + let started!: () => void; + const running = new Promise(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}`); diff --git a/server/src/__tests__/workspace-operations-reconciliation.test.ts b/server/src/__tests__/workspace-operations-reconciliation.test.ts index 03d1d80bbeb..4414d511432 100644 --- a/server/src/__tests__/workspace-operations-reconciliation.test.ts +++ b/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(); }); diff --git a/server/src/services/native-runtime/native-workspace-export-recovery.ts b/server/src/services/native-runtime/native-workspace-export-recovery.ts index a86883cd4d3..25314aef082 100644 --- a/server/src/services/native-runtime/native-workspace-export-recovery.ts +++ b/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)); diff --git a/server/src/services/workspace-operations.ts b/server/src/services/workspace-operations.ts index 53fcbf5292c..60103c15fbd 100644 --- a/server/src/services/workspace-operations.ts +++ b/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(write: (database: Db) => Promise): Promise { + 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 | 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 diff --git a/server/src/services/workspace-recovery.test.ts b/server/src/services/workspace-recovery.test.ts index 9edfdf1d391..350cd75ef1f 100644 --- a/server/src/services/workspace-recovery.test.ts +++ b/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>; + let temporary: Awaited> | undefined; + let connectionString: string; let db: ReturnType; 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>> = []; + 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); diff --git a/server/src/services/workspace-recovery.ts b/server/src/services/workspace-recovery.ts index e2fae8179e5..eee81906492 100644 --- a/server/src/services/workspace-recovery.ts +++ b/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 } });