diff --git a/server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts b/server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts index 6144f316570..b5bd1bf553c 100644 --- a/server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts +++ b/server/src/__tests__/heartbeat-observed-workspace-preservation.test.ts @@ -25,6 +25,9 @@ import { hasVerifiedWorkspaceRepair } from "../services/workspace-recovery-recei import { pinFailedObservedWorkspace, settleStoppedObservedWorkspace, ObservedWorkspacePreservationBusyError } from "../services/native-runtime/native-observed-workspace-preservation.js"; import { claimNativeRestartRecoveries } from "../services/native-runtime/native-restart-recovery.js"; import { hasRemoteTerminationReceipt } from "../services/remote-execution-termination.js"; +import { hasConfirmedSandboxStopAndRetain } from "../services/sandbox-stop-and-retain.js"; +import { createOwnershipReportState, runOwnershipReconciliationPass, settleOwnableKind } from "../services/recovery/ownership-invariant-report.js"; +import { OWNABLE_KINDS } from "../services/recovery/ownership-registry.js"; import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; const execute = promisify(execFile); @@ -326,6 +329,35 @@ describeEmbeddedPostgres("stopped observed native workspace preservation", () => } }); + it("settles the fence from the shared reconciler once the provider stop is proven, never before", async () => { + const f = await seedSource("SHARED-RECONCILER"); + const kind = OWNABLE_KINDS.find(entry => entry.id === "observed_workspace_preservation_fence")!; + const pass = (record = true) => runOwnershipReconciliationPass(f.db, { state: createOwnershipReportState(), kinds: [kind], record }); + const settleThis = () => settleOwnableKind(f.db, kind.id, { runIds: [f.runId] }); + const fenceOf = async () => (await f.db.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.sourceIssueId, f.issueId), + eq(issueRecoveryActions.fingerprint, `native-workspace:${f.runId}`))))[0].evidence.observedWorkspacePreservation as { state: string }; + expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true); + const held = await pass(false); + expect(held.kinds[0]!.reasons.fence_expired_without_provider_proof).toBeGreaterThanOrEqual(1); + expect(await settleThis()).toEqual([]); + expect(await fenceOf()).toMatchObject({ state: "stop_pending" }); + + const heartbeat = heartbeatService(f.db); + try { + await heartbeat.sweepPendingCleanupLeases({ backoffMs: 0 }); + const { lease } = await readSource(f); + expect(hasRemoteTerminationReceipt(lease) && hasConfirmedSandboxStopAndRetain(lease)).toBe(true); + const readOnly = await pass(false); + expect(readOnly.kinds[0]!.reasons.provider_stopped).toBeGreaterThanOrEqual(1); + expect(await fenceOf()).toMatchObject({ state: "stop_pending" }); + + expect(await settleThis()).toMatchObject([{ kind: kind.id, runId: f.runId, reason: "provider_stopped" }]); + expect(await fenceOf()).toMatchObject({ state: "stopped" }); + expect((await readSource(f)).owner).toMatchObject({ phase: "terminal_failure", leaseOwner: null }); + expect(await readFile(path.join(f.remoteCwd, "saved.txt"), "utf8")).toBe("unfinished work with an unknown provider effect\n"); + } finally { await heartbeat.drainActiveRunExecutions(); } + }); + it("rechecks a new physical allocation owner after pin before the actual Stop dispatch", async () => { const f = await seedSource("ARRIVING-OWNER"); expect(await pinFailedObservedWorkspace(f.db, companyId, f.runId)).toBe(true); diff --git a/server/src/__tests__/ownership-reconciliation-report.test.ts b/server/src/__tests__/ownership-reconciliation-report.test.ts index cdfd609147a..c05f83521ee 100644 --- a/server/src/__tests__/ownership-reconciliation-report.test.ts +++ b/server/src/__tests__/ownership-reconciliation-report.test.ts @@ -143,7 +143,7 @@ describe("report-only ownership reconciliation", () => { expect(new Set(names).size).toBe(names.length); expect(names.length).toBeGreaterThanOrEqual(40); expect(OWNABLE_KINDS.filter(kind => kind.settle !== null).map(kind => kind.id)) - .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "copyback_owner_hold", "wake_request_claimed", "legacy_controller_lease", "issue_execution_lock", "issue_checkout_lock"]); + .toEqual(["native_accepted_result_lease", "native_finalizer_lease", "native_cleanup_lease", "sandbox_creation_intent", "environment_run_lease_active", "environment_teardown_claim", "environment_retained_orphan_sandbox", "workspace_runtime_control_operation", "native_workspace_finalize_attempt", "execution_workspace_runtime_lease", "copyback_owner_hold", "observed_workspace_preservation_fence", "wake_request_claimed", "legacy_controller_lease", "issue_execution_lock", "issue_checkout_lock"]); expect(OWNABLE_KINDS.filter(kind => !kind.listInProgress).every(kind => kind.coverageGap)).toBe(true); }); diff --git a/server/src/services/native-runtime/native-observed-workspace-preservation.ts b/server/src/services/native-runtime/native-observed-workspace-preservation.ts index 14e466031fc..fe82a3781c7 100644 --- a/server/src/services/native-runtime/native-observed-workspace-preservation.ts +++ b/server/src/services/native-runtime/native-observed-workspace-preservation.ts @@ -174,6 +174,7 @@ export async function pinFailedObservedWorkspace(db: Db, companyId: string, runI && lease.metadata?.workspaceRecoveryCleanup == null; if (!active) return false; await notifyNativeEnvironmentSettlement(tx, { companyId, runId, issueId: task.id }); + await notifyDeliveryWork(tx, DELIVERY_QUEUES.ownershipInvariantReport, JSON.stringify({ companyId, runId })); const retained = await recordRequiredWorkspaceRestoreFailure(tx, run, { workspaceRestoreFailure: "restore_failed" }, lease); if (!retained || !hasRequiredWorkspaceRecovery(retained.resultJson)) throw new Error("observed_workspace_retention_missing"); const [action] = await tx.select().from(issueRecoveryActions).where(and(eq(issueRecoveryActions.companyId, companyId), diff --git a/server/src/services/recovery/copyback-owner-hold.ts b/server/src/services/recovery/copyback-owner-hold.ts new file mode 100644 index 00000000000..1c19be53c63 --- /dev/null +++ b/server/src/services/recovery/copyback-owner-hold.ts @@ -0,0 +1,32 @@ +import type { Db } from "@paperclipai/db"; +import { copybackOwnerStamp } from "../native-runtime/native-workspace-owner-stop-proof.js"; +import type { OwnerProof } from "../owner-proof.js"; +import type { InProgressRecord, RecordOwner, SettleOutcome } from "./ownership-registry.js"; + +const object = (value: unknown): Record => + value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; +const text = (value: unknown) => typeof value === "string" && value ? value : null; + +export function copybackOwnerHoldOwner(evidence: unknown): RecordOwner { + const owner = object(object(object(evidence).workspaceOwnerRecovery).owner); + const pid = typeof owner.pid === "number" && Number.isInteger(owner.pid) && owner.pid > 0 ? owner.pid : null; + const hostname = text(owner.hostname); + const processStartedAt = text(owner.processStartedAt); + if (!pid || !hostname || !processStartedAt) return { kind: "none" }; + const controllerBootId = text(owner.controllerBootId); + const hostHash = text(owner.hostHash); + return { kind: "process", stamp: { ...copybackOwnerStamp({ hostname, pid, processStartedAt, + ...(controllerBootId ? { controllerBootId } : {}), ...(hostHash ? { hostHash } : {}) }), scope: "copyback_owner_hold" } }; +} + +export async function settleCopybackOwnerHold(db: Db, record: InProgressRecord, + context: { now: Date; proof: OwnerProof }): Promise { + if (record.owner.kind !== "process" || !record.companyId) return { settled: false, reason: "owner_not_provable" }; + const { wakeCopybackOwnerRecheck } = await import("../workspace-recovery.js"); + const stamp = record.owner.stamp; + const woke = await wakeCopybackOwnerRecheck(db, { companyId: record.companyId, actionId: record.recordId, + now: context.now, owner: { pid: stamp.pid, processStartedAt: stamp.processStartedAt, bootId: stamp.bootId, + processSpace: stamp.processSpace } }); + return woke ? { settled: false, reason: "copyback_recheck_woken" } + : { settled: false, reason: "copyback_recheck_not_waiting", watch: false }; +} diff --git a/server/src/services/recovery/observed-workspace-fence.ts b/server/src/services/recovery/observed-workspace-fence.ts new file mode 100644 index 00000000000..f7f8f4b7c11 --- /dev/null +++ b/server/src/services/recovery/observed-workspace-fence.ts @@ -0,0 +1,55 @@ +import { inArray } from "drizzle-orm"; +import { environmentLeases, type Db } from "@paperclipai/db"; +import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; +import { hasConfirmedSandboxStopAndRetain } from "../sandbox-stop-and-retain.js"; +import type { InProgressRecord, RecordOwner, SettleOutcome } from "./ownership-registry.js"; + +export const OBSERVED_WORKSPACE_FENCE_SCOPE = "observed_workspace_preservation_fence"; + +const object = (value: unknown): Record => + value && typeof value === "object" && !Array.isArray(value) ? value as Record : {}; + +export function observedWorkspaceFenceOwner(evidence: unknown): RecordOwner { + const fence = object(object(evidence).observedWorkspacePreservation); + const token = typeof fence.leaseOwner === "string" && fence.leaseOwner ? fence.leaseOwner : null; + const generation = typeof fence.controllerGeneration === "number" && Number.isSafeInteger(fence.controllerGeneration) + ? fence.controllerGeneration : null; + return token ? { kind: "fence", stamp: { kind: "remote", scope: OBSERVED_WORKSPACE_FENCE_SCOPE, generation, fenceToken: token }, + fence: { currentFenceToken: null, currentGeneration: null, leaseExpiresAt: null } } : { kind: "none" }; +} + +export async function providerStoppedObservedRuns(db: Db, runIds: readonly string[]): Promise> { + const stopped = new Set(); + if (runIds.length === 0) return stopped; + const leases = await db.select().from(environmentLeases).where(inArray(environmentLeases.heartbeatRunId, [...runIds])); + const byRun = new Map(); + for (const lease of leases) byRun.set(lease.heartbeatRunId!, [...(byRun.get(lease.heartbeatRunId!) ?? []), lease]); + for (const [runId, runLeases] of byRun) { + if (runLeases.length === 1 && hasRemoteTerminationReceipt(runLeases[0]!) && hasConfirmedSandboxStopAndRetain(runLeases[0]!)) { + stopped.add(runId); + } + } + return stopped; +} + +export async function applyObservedFenceProviderProof(db: Db, records: readonly InProgressRecord[]): Promise { + const observed = records.filter(record => record.owner.kind === "fence" && record.runId + && record.owner.stamp.scope === OBSERVED_WORKSPACE_FENCE_SCOPE); + const stopped = await providerStoppedObservedRuns(db, [...new Set(observed.map(record => record.runId!))]); + for (const record of observed) { + if (record.owner.kind === "fence" && stopped.has(record.runId!)) record.owner.fence = { ...record.owner.fence, provider: "stopped" }; + } +} + +export async function settleObservedWorkspaceFence(db: Db, record: InProgressRecord): Promise { + if (!record.companyId || !record.runId) return { settled: false, reason: "owner_not_provable" }; + const { ObservedWorkspacePreservationBusyError, settleStoppedObservedWorkspace } = + await import("../native-runtime/native-observed-workspace-preservation.js"); + try { + return await settleStoppedObservedWorkspace(db, record.companyId, record.runId) + ? { settled: true, reason: "provider_stopped" } : { settled: false, reason: "fence_not_settleable" }; + } catch (error) { + if (error instanceof ObservedWorkspacePreservationBusyError) return { settled: false, reason: "source_busy" }; + throw error; + } +} diff --git a/server/src/services/recovery/ownership-registry.ts b/server/src/services/recovery/ownership-registry.ts index 53dda933fea..bd9dff77519 100644 --- a/server/src/services/recovery/ownership-registry.ts +++ b/server/src/services/recovery/ownership-registry.ts @@ -34,7 +34,8 @@ import { import { settleClaimedWakeRequest } from "./wake-request-claimed-settlement.js"; import { settleAbandonedWorkspaceFinalizeAttempt } from "./workspace-finalize-attempt-settlement.js"; import { runtimeControlOperationOwnerStamp, settleRuntimeControlOperation } from "../workspace-operations.js"; -import { copybackOwnerStamp } from "../native-runtime/native-workspace-owner-stop-proof.js"; +import { copybackOwnerHoldOwner, settleCopybackOwnerHold } from "./copyback-owner-hold.js"; +import { applyObservedFenceProviderProof, observedWorkspaceFenceOwner, settleObservedWorkspaceFence } from "./observed-workspace-fence.js"; import { nativeControllerOwnerStamp, runProcessOwnerStamp } from "../native-runtime/native-owner-stamps.js"; import { settleIssueLock } from "./issue-lock-settlement.js"; import { legacyControllerOwnerStamp, settleLegacyControllerLease } from "../legacy-controller-lease.js"; @@ -461,14 +462,16 @@ type RecoveryActionRow = Pick RecordOwner, invariantChecks: readonly number[] = [1, 2, 7], - runIdOf: (row: RecoveryActionRow) => string | null = () => null): OwnableKind { + runIdOf: (row: RecoveryActionRow) => string | null = () => null, + options: { includeResolved?: boolean } = {}): OwnableKind { return { id, title, obligation, storage: "database", settle: null, invariantChecks, listInProgress: async (db, limit, offset) => { const rows = await db.select({ id: issueRecoveryActions.id, companyId: issueRecoveryActions.companyId, sourceIssueId: issueRecoveryActions.sourceIssueId, ownerType: issueRecoveryActions.ownerType, evidence: issueRecoveryActions.evidence, updatedAt: issueRecoveryActions.updatedAt }) - .from(issueRecoveryActions).where(and(isNull(issueRecoveryActions.resolvedAt), where)).orderBy(asc(issueRecoveryActions.id)).limit(limit + 1).offset(offset); + .from(issueRecoveryActions).where(and(options.includeResolved ? undefined : isNull(issueRecoveryActions.resolvedAt), where)) + .orderBy(asc(issueRecoveryActions.id)).limit(limit + 1).offset(offset); const result = page(rows, limit); return { truncated: result.truncated, records: result.rows.map(row => ({ recordId: row.id, companyId: row.companyId, runId: runIdOf(row), issueId: row.sourceIssueId, @@ -850,46 +853,18 @@ export const OWNABLE_KINDS: readonly OwnableKind[] = [ sql`${issueRecoveryActions.evidence}->'systemRecovery'->>'state' in ('scheduled', 'running', 'awaiting_evidence')`, () => ({ kind: "none" })), { - // Shared path since Phase 2: receipts record their PID space, so the shared - // prover judges the owner (older receipts name none and stay unprovable). - // A proven-gone owner wakes the waiting recheck; the worker still proves the - // copyback process group before it takes ownership, so this never settles - // the hold itself. ...recoveryActionKind("copyback_owner_hold", "Copyback-owner hold (RC-50)", "ownership", sql`${issueRecoveryActions.evidence}->'workspaceOwnerRecovery'->>'state' = 'operator_required'`, - row => { - const owner = json(json(json(row.evidence).workspaceOwnerRecovery).owner); - const pid = int(owner.pid); - const hostname = str(owner.hostname); - const processStartedAt = str(owner.processStartedAt); - if (!pid || !hostname || !processStartedAt) return { kind: "none" }; - return { kind: "process", stamp: { ...copybackOwnerStamp({ hostname, pid, processStartedAt, - ...(str(owner.controllerBootId) ? { controllerBootId: str(owner.controllerBootId)! } : {}), - ...(str(owner.hostHash) ? { hostHash: str(owner.hostHash)! } : {}) }), scope: "copyback_owner_hold" } }; - }), - settle: async (db, record, context) => { - if (record.owner.kind !== "process" || !record.companyId) return { settled: false, reason: "owner_not_provable" }; - const { wakeCopybackOwnerRecheck } = await import("../workspace-recovery.js"); - const stamp = record.owner.stamp; - const woke = await wakeCopybackOwnerRecheck(db, { companyId: record.companyId, actionId: record.recordId, - now: context.now, owner: { pid: stamp.pid, processStartedAt: stamp.processStartedAt, bootId: stamp.bootId, - processSpace: stamp.processSpace } }); - // A hold whose recheck is not waiting (exhausted, past its deadline or - // moved on) is the board's: keep it visible, but stop watching it. - return woke ? { settled: false, reason: "copyback_recheck_woken" } - : { settled: false, reason: "copyback_recheck_not_waiting", watch: false }; - }, + row => copybackOwnerHoldOwner(row.evidence)), + settle: settleCopybackOwnerHold, + }, + { + ...recoveryActionKind("observed_workspace_preservation_fence", "Observed-run workspace preservation fence", "workspace_preservation", + sql`${issueRecoveryActions.evidence}->'observedWorkspacePreservation'->>'state' = 'stop_pending'`, + row => observedWorkspaceFenceOwner(row.evidence), [1, 2, 7], + row => str(json(json(row.evidence).observedWorkspacePreservation).runId), { includeResolved: true }), + settle: settleObservedWorkspaceFence, }, - recoveryActionKind("observed_workspace_preservation_fence", "Observed-run workspace preservation fence", "workspace_preservation", - sql`${issueRecoveryActions.evidence}->'observedWorkspacePreservation'->>'state' = 'stop_pending'`, - row => { - const fence = json(json(row.evidence).observedWorkspacePreservation); - const token = str(fence.leaseOwner); - // The current holder is read in the report pass (`resolveFenceHolders`). - return token ? { kind: "fence", stamp: { kind: "remote", scope: "observed_workspace_preservation_fence", - generation: int(fence.controllerGeneration), fenceToken: token }, - fence: { currentFenceToken: null, currentGeneration: null, leaseExpiresAt: null } } : { kind: "none" }; - }, [1, 2, 7], row => str(json(json(row.evidence).observedWorkspacePreservation).runId)), // Accounting and budgets runKind("run_accounting_debt", "Run accounting debt", "accounting", @@ -1018,4 +993,5 @@ export async function resolveFenceHolders(db: Db, records: InProgressRecord[]): record.owner.fence = { currentFenceToken: holder?.leaseOwner ?? null, currentGeneration: holder?.generation ?? null, leaseExpiresAt: holder?.leaseExpiresAt ?? null }; } + await applyObservedFenceProviderProof(db, pending); }