diff --git a/doc/native-workspace-finalization-recovery.md b/doc/native-workspace-finalization-recovery.md new file mode 100644 index 00000000000..45621debae8 --- /dev/null +++ b/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::', 0)); +UPDATE heartbeat_runs +SET runner_profile_json = runner_profile_json - 'nativeWorkspaceFinalizationOwner' +WHERE company_id = ''::uuid + AND id = ''::uuid + AND native_issue_id = ''::uuid + AND runtime_mode = 'native' + AND runner_profile_json->'nativeWorkspaceFinalizationOwner'->>'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. diff --git a/packages/db/src/client.ts b/packages/db/src/client.ts index 70030b051d3..ecfd7954421 100644 --- a/packages/db/src/client.ts +++ b/packages/db/src/client.ts @@ -255,6 +255,19 @@ export function postgresJsOptions(options: DatabaseClientOptions): Record Db>(); + +export async function withDedicatedDbConnection(db: Db, action: (dedicated: Db) => Promise): Promise { + 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 { diff --git a/packages/db/src/index.ts b/packages/db/src/index.ts index fb62bcf9cca..378c9bd37b8 100644 --- a/packages/db/src/index.ts +++ b/packages/db/src/index.ts @@ -1,5 +1,6 @@ export { createDb, + withDedicatedDbConnection, closeRegisteredClients, getPostgresDataDirectory, ensurePostgresDatabase, diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 2217f8338fb..42b6ae9d086 100644 --- a/server/src/services/heartbeat.ts +++ b/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({ diff --git a/server/src/services/native-runtime/native-finalization-reconciler.ts b/server/src/services/native-runtime/native-finalization-reconciler.ts index 6ad2cebfd64..c5a2a3028a6 100644 --- a/server/src/services/native-runtime/native-finalization-reconciler.ts +++ b/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") { diff --git a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts index 721c61546e7..a4f47f143bb 100644 --- a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts +++ b/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 diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 6f4d572c029..b49ab168988 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/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, diff --git a/server/src/services/native-runtime/native-workspace-finalization-ownership.ts b/server/src/services/native-runtime/native-workspace-finalization-ownership.ts new file mode 100644 index 00000000000..f0e0d78de87 --- /dev/null +++ b/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; +}; +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; + 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( + input: { db: Db; companyId: string; runId: string }, + action: (ownership: NativeWorkspaceFinalizationOwnership) => Promise, +): 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 | 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); + } + } +} diff --git a/server/src/services/native-runtime/native-workspace-finalizer.test.ts b/server/src/services/native-runtime/native-workspace-finalizer.test.ts index 7221370023a..c24624100f0 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.test.ts +++ b/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((resolve) => { enter = resolve; }); + let finish!: () => void; + const finished = new Promise((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((resolve) => { entered = resolve; }); + let release!: () => void; + const held = new Promise((resolve) => { release = resolve; }); + const stat = fs.stat.bind(fs); + let calls = 0; + const spy = vi.spyOn(fs, "stat").mockImplementation(async (...args: Parameters) => { + 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), ), diff --git a/server/src/services/native-runtime/native-workspace-finalizer.ts b/server/src/services/native-runtime/native-workspace-finalizer.ts index 1e0c71bc9c4..1bdc3ac2c21 100644 --- a/server/src/services/native-runtime/native-workspace-finalizer.ts +++ b/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; } diff --git a/server/src/services/native-runtime/native-workspace-sync.ts b/server/src/services/native-runtime/native-workspace-sync.ts index a618f612ebc..9953d1473f7 100644 --- a/server/src/services/native-runtime/native-workspace-sync.ts +++ b/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; + restoreWorkspace(assertOwnership?: () => Promise): Promise; cleanup(): Promise; } @@ -772,8 +772,10 @@ async function finalizePreparedRuntime(input: { target: Extract; runtime: PreparedAdapterExecutionTargetRuntime; descriptor: NativeWorkspaceSyncDescriptor; + assertOwnership?: () => Promise; }): Promise { 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; }): Promise { 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; }