diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 2478035988a..67b0b47f82c 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1933,10 +1933,32 @@ judged through that boot's row: - no row (a boot from before the lease): the record stays unprovable. A record with a recorded process group stays unprovable across hosts, because -agent processes may outlive their server. The lease is off on a Railway -deployment with an exclusive volume, which proves a replaced deployment from its -own identity. A stateless Railway service keeps it. -`PAPERCLIP_CONTROLLER_INSTANCE_LEASE=on|off` overrides this. +agent processes may outlive their server. The lease is enabled on volume-backed +Railway services too: a process can crash and restart inside the same deployment, +and some saved ownership records do not contain a deployment identity. +`PAPERCLIP_CONTROLLER_INSTANCE_LEASE=off` explicitly disables it. + +Native execution and restart claims register held preparation with this lease before +a runner PID or checkpoint exists. Passive identity reads do not register work. +Restart recovery uses the recorded controller +boot and last write to prove its owner stopped or to fence it after the existing +grace period when the PID space is foreign or unknown. Within the same readable +PID space, the existing process and start-time proof still governs takeover. +A live owner or a grace period keeps the preparation held, with a +scheduled recovery check. Repeating the same evidence wait preserves its last-write +timestamp so the recovery check cannot keep extending its own grace period. +After owner proof, the normal launch, checkpoint, provider-evidence and admission +checks still govern resumption; controller fencing alone does not prove an agent +process stopped or authorize repeating its work. + +Older native controller records with no verifiable boot lease remain held. Recovery +exposes the existing board stop-confirmation action for an unknown or foreign PID +space. It requires `runtime:confirm_owner_stop` and confirmation of the exact run, +boot, PID, process start and recorded host. A changed owner rejects the confirmation. +The receipt only permits normal recovery to inspect and resume the saved work. +Scoped recovery wakes also select fenced preparations with no runner process or +checkpoint yet. They use the existing finite expired-run inventory and normal +restart classifier; a missing checkpoint cannot silently discard that wake. A record saved before its owner was fully recorded (`no_owner_stamp`, `missing_identity`, `process_space_unrecorded`) stays unprovable and is never @@ -4892,6 +4914,19 @@ Upgrade diagnostics retain their existing `ready` meaning and add bounded retain header evidence. That inventory is preliminary, reports partial coverage, and does not certify workspace durability or volume removal. +The upgrade census separates current execution from an old, unowned native +preparation that recovery explicitly retained as `awaiting_evidence`. This requires +an observed preparation with no result, failure, process identity, or checkpoint +header, and an expired, recorded owner from a different controller boot. Such a +run is counted as `retainedBlockedRuns`, rather than `activeRuns` or +`recoverableRuns`. It remains unchanged and held for its normal recovery path; +the observation does not prove that its old controller stopped or authorize +replay, resource release, or volume removal. A run registered in the current +process still blocks unless it has positive recovery evidence. Before/after +execution counters, pending wake operations, drain state, and a truncated census +continue to prevent readiness. Persisted queued wakes are not deleted by an +upgrade; the pending wake counter measures in-process admission operations. + `importNativeControllerRecovery` is an internal one-source operation for the migration owner while old controller files remain available. It requires the admitted configuration digest and exact supervisor snapshot, locks the issue, diff --git a/server/src/__tests__/controller-instance-lease.test.ts b/server/src/__tests__/controller-instance-lease.test.ts index 7df0d46c85d..53a1eecdac5 100644 --- a/server/src/__tests__/controller-instance-lease.test.ts +++ b/server/src/__tests__/controller-instance-lease.test.ts @@ -13,6 +13,7 @@ import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../services/environment-lease- import { proveOwnerGone, type LocalOwnerStamp } from "../services/owner-proof.js"; import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js"; import { beginIdleTrackedWork, readTaskDrain } from "../services/task-admission.js"; +import { currentNativeControllerIdentity, evaluateNativeControllerTakeover } from "../services/native-runtime/native-restart-recovery.js"; const support = await getEmbeddedPostgresTestSupport(); const suite = support.supported ? describe : describe.skip; @@ -93,6 +94,24 @@ suite("controller instance lease: a replaced container's records are provable", expect(await proveOwnerGone(stampOf(live))).toEqual({ verdict: "alive", reason: "controller_lease_held" }); }); + it("tracks native preparation before a runner exists and proves a fenced foreign controller", async () => { + const replaced = await otherBoot({ expiredMsAgo: 600_000 }); + await startHere({}, false); + const owner = await currentNativeControllerIdentity(); + await tickControllerInstanceLease(); + expect(await db.select().from(controllerInstances).where(eq(controllerInstances.bootId, owner.bootId))).toEqual([]); + currentControllerProcessStamp(); + await tickControllerInstanceLease(); + expect(await db.select().from(controllerInstances).where(eq(controllerInstances.bootId, owner.bootId))).toHaveLength(1); + const decision = await evaluateNativeControllerTakeover({ now: new Date(), owner: { + controllerBootId: replaced, controllerPid: 7, controllerHostHash: "old-container", + controllerProcessStartedAt: new Date(minutesAgo(15)), updatedAt: new Date(minutesAgo(10)), + leaseOwner: "old-preparation", leaseExpiresAt: new Date(minutesAgo(5)), + }, isProcessAlive: () => { throw new Error("A foreign PID must not be inspected locally"); } }); + expect(decision).toEqual({ allowed: true, reason: "controller_fenced" }); + expect(await proveControllerInstance(db, replaced, null)).toBe("fenced"); + }); + it("keeps an owner alive (and watched) through the grace, and for work written after an idle lapse", async () => { const justExpired = await otherBoot({ expiredMsAgo: 5_000 }); const resumed = await otherBoot({ expiredMsAgo: 600_000 }); @@ -396,7 +415,7 @@ suite("controller instance lease: a replaced container's records are provable", expect(lost).toHaveBeenCalledWith("fenced"); }); - it("records a graceful stop, and is off on a volume-exclusive Railway deployment or when disabled", async () => { + it("records a graceful stop and keeps native preparation fencing on volume-exclusive deployments", async () => { await startHere(); await stopControllerInstanceLease(db); expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("stopped"); @@ -410,7 +429,7 @@ suite("controller instance lease: a replaced container's records are provable", expect(controllerInstanceLeaseEnabled()).toBe(true); vi.stubEnv("RAILWAY_VOLUME_ID", "v"); vi.stubEnv("RAILWAY_VOLUME_MOUNT_PATH", "/data"); - expect(controllerInstanceLeaseEnabled()).toBe(false); + expect(controllerInstanceLeaseEnabled()).toBe(true); vi.stubEnv("PAPERCLIP_CONTROLLER_INSTANCE_LEASE", "on"); expect(controllerInstanceLeaseEnabled()).toBe(true); vi.unstubAllEnvs(); diff --git a/server/src/services/controller-instance-lease.ts b/server/src/services/controller-instance-lease.ts index 36ff6f52412..48ebd317e54 100644 --- a/server/src/services/controller-instance-lease.ts +++ b/server/src/services/controller-instance-lease.ts @@ -2,7 +2,7 @@ import { and, eq, isNull, sql } from "drizzle-orm"; import { controllerInstances, type Db } from "@paperclipai/db"; import { logger } from "../middleware/logger.js"; import { - currentControllerBootId, currentControllerDeployment, currentProcessSpaceHash, onControllerStamp, + currentControllerBootId, currentProcessSpaceHash, onControllerStamp, } from "./controller-process-identity.js"; import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; import { DELIVERY_QUEUES } from "./delivery-work-notifications.js"; @@ -54,14 +54,8 @@ type LeaseState = { let state: LeaseState | null = null; let registeredDb: Db | null = null; -/** A Railway deployment with an exclusive volume proves a replaced deployment on its own; every - * other host (Docker, Kubernetes, Fly, a stateless Railway service, a rebooted machine) needs the - * lease to prove a replaced process. PAPERCLIP_CONTROLLER_INSTANCE_LEASE=on|off overrides this. */ export function controllerInstanceLeaseEnabled(env: NodeJS.ProcessEnv = process.env): boolean { - const setting = env.PAPERCLIP_CONTROLLER_INSTANCE_LEASE; - if (setting === "off") return false; - if (setting === "on") return true; - return currentControllerDeployment(env)?.exclusiveVolume !== true; + return env.PAPERCLIP_CONTROLLER_INSTANCE_LEASE !== "off"; } function exitOnLostLease(reason: "fenced" | "lease_lost") { diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index dc27c7641a5..f031b88878c 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -211,6 +211,36 @@ describePostgres("heartbeat recovery module ownership", () => { return (await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, invocationSource: "automation", status: "running", runtimeMode: "legacy" }).returning())[0]; } + it.each([false, true])("delivers scoped recovery to a preparation without a runner PID or checkpoint (locally tracked=%s)", async tracked => { + const run = await seedRun(), deps = callbacks(db); + await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, run.agentId)); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, + nativePhase: "observed", runnerProfileJson: { nativePreparationFencingVersion: 1 }, + }).where(eq(heartbeatRuns.id, run.id)); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, + phase: "observed", leaseOwner: "old-preparation", leaseExpiresAt: new Date(0), controllerHostHash: "operator_attested" }); + const claim: NativeRestartRecoveryClaim = { kind: "bootstrap_incomplete", runId: run.id, leaseOwner: "recovered-owner", + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null }; + native.claims.mockResolvedValue([claim]); + if (tracked) deps.activeRunExecutions.add(run.id); + try { + await createHeartbeatRecovery(db, deps).recoverActiveExecutions({ companyId: run.companyId, runId: run.id }); + if (tracked) { + expect(native.claims).not.toHaveBeenCalled(); + expect(deps.executeRun).not.toHaveBeenCalled(); + } else { + expect(native.claims).toHaveBeenCalledWith(expect.objectContaining({ runIds: [run.id] })); + expect(deps.executeRun).toHaveBeenCalledExactlyOnceWith(run.id, { nativeLeaseOwner: claim.leaseOwner, nativeRestartRecovery: claim }); + } + expect(deps.setRunStatusFromLive).not.toHaveBeenCalled(); + expect(deps.terminateHeartbeatRunProcess).not.toHaveBeenCalled(); + } finally { + await Promise.all([...deps.activeRunExecutionPromises]); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, run.id)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.id, run.id)); + } + }); + it("routes an expired pre-capability startup to cleanup before generic orphan ownership holds", async () => { const run = await seedRun(), leaseId = randomUUID(), environmentId = randomUUID(), leaseOwner = randomUUID(); const deadlineAtMs = Date.now() - 1_000, contextId = randomUUID(); diff --git a/server/src/services/heartbeat/upgrade-readiness.test.ts b/server/src/services/heartbeat/upgrade-readiness.test.ts index fb605c53be1..e5f92444e3a 100644 --- a/server/src/services/heartbeat/upgrade-readiness.test.ts +++ b/server/src/services/heartbeat/upgrade-readiness.test.ts @@ -36,6 +36,17 @@ describe("upgrade readiness", () => { it("never treats a truncated census as complete", () => { const rows = Array.from({ length: 1_001 }, (_, i) => ({ id: String(i), recoverable: true })); expect(report({ rows, activeRunIds: [] })).toMatchObject({ ready: false, truncated: true }); + expect(report({ rows: rows.map(row => ({ ...row, recoverable: false, retainedUnowned: true })), + before: idle, after: idle, activeRunIds: [] })).toMatchObject({ ready: false, truncated: true, activeRuns: 0 }); + }); + it("keeps current execution and admission barriers with retained recovery work", () => { + const rows = [{ id: "retained", recoverable: false, retainedUnowned: true }, { id: "remote", recoverable: true }]; + expect(report({ rows })).toMatchObject({ ready: true, activeRuns: 1, recoverableRuns: 1, retainedBlockedRuns: 1 }); + expect(report({ rows, activeRunIds: ["remote", "retained"], after: { ...idle, activeRuns: 2 } })) + .toMatchObject({ ready: false, activeRuns: 2, blockingRuns: 1, retainedBlockedRuns: 0 }); + for (const before of [{ ...idle, activeRuns: 2 }, { ...idle, pendingWakes: 1 }, { ...idle, draining: true }]) { + expect(report({ rows, before }).ready).toBe(false); + } }); }); @@ -54,6 +65,80 @@ postgresTests("upgrade readiness persisted execution evidence", () => { db = createDb(database.connectionString); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); afterAll(async () => { await db?.$client.end(); await database?.cleanup(); vi.unstubAllEnvs(); }); + it("keeps an old unowned preparation visible without making it a live upgrade blocker", async () => { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const owner = await currentNativeControllerIdentity(); + await db.insert(companies).values({ id: companyId, name: "Retained preparation", issuePrefix: `R${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Retained preparation", role: "engineer", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Interrupted preparation", status: "in_progress", assigneeAgentId: agentId }); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, + status: "running", runtimeMode: "native", nativePhase: "observed", runnerInstanceId: randomUUID(), nativeSessionId: randomUUID() }); + const retained = { controllerBootId: "previous-controller", leaseOwner: "previous-controller-lease", controllerGeneration: 1, + leaseExpiresAt: new Date(0), recoveryState: "awaiting_evidence", phase: "observed", failureCode: null }; + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, ...retained }); + const input = { activeRunIds: new Set(), counters: () => idle }; + try { + const savedRun = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)); + const savedOwner = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + expect(await readUpgradeReadiness(db, input)).toMatchObject({ ready: true, activeRuns: 0, + recoverableRuns: 0, blockingRuns: 0, retainedBlockedRuns: 1, + nativeControllerRecovery: { activeExecutionAdoption: { retainedBlockedRuns: 1 } } }); + const [{ default: express }, { default: request }, { instanceSettingsRoutes }, { errorHandler }] = await Promise.all([ + import("express"), import("supertest"), import("../../routes/instance-settings.js"), import("../../middleware/error-handler.js"), + ]); + const app = express(); + let isInstanceAdmin = false; + app.use((req, _res, next) => { + req.actor = { type: "board", userId: "readiness-test", source: "session", isInstanceAdmin }; + next(); + }); + app.use("/api", instanceSettingsRoutes(db)); + app.use(errorHandler); + await request(app).get("/api/instance/task-drain?upgradeReadiness=1").expect(403); + isInstanceAdmin = true; + const response = await request(app).get("/api/instance/task-drain?upgradeReadiness=1").expect(200); + expect(response.body.upgradeReadiness).toMatchObject({ ready: true, activeRuns: 0, blockingRuns: 0, retainedBlockedRuns: 1 }); + expect(await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId))).toEqual(savedRun); + expect(await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId))).toEqual(savedOwner); + expect(await db.select({ status: heartbeatRuns.status }).from(heartbeatRuns).where(eq(heartbeatRuns.id, runId))) + .toEqual([{ status: "running" }]); + expect(await db.select({ state: nativeRunFinalizations.recoveryState }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, runId))).toEqual([{ state: "awaiting_evidence" }]); + for (const patch of [ + { controllerBootId: owner.bootId }, { controllerBootId: null }, { controllerBootId: "" }, + { leaseExpiresAt: new Date(Date.now() + 60_000) }, { leaseExpiresAt: null }, + { leaseOwner: null }, { controllerGeneration: 0 }, { failureCode: "uncertain" }, + { recoveryState: null }, { phase: "workspace_finalizing" }, + ]) { + await db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, runId)); + expect(await readUpgradeReadiness(db, input)).toMatchObject({ ready: false, blockingRuns: 1, retainedBlockedRuns: 0 }); + await db.update(nativeRunFinalizations).set(retained).where(eq(nativeRunFinalizations.runId, runId)); + } + for (const patch of [{ runtimeMode: "legacy" }, { nativePhase: "running" }, { processPid: process.pid }, { processStartedAt: new Date() }]) { + await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, runId)); + expect(await readUpgradeReadiness(db, input)).toMatchObject({ ready: false, blockingRuns: 1, retainedBlockedRuns: 0 }); + await db.update(heartbeatRuns).set({ runtimeMode: "native", nativePhase: "observed", processPid: null, processStartedAt: null }) + .where(eq(heartbeatRuns.id, runId)); + } + await db.insert(nativeRunnerCheckpoints).values({ runId, companyId, issueId, part: "header", material: {} }); + expect(await readUpgradeReadiness(db, input)).toMatchObject({ ready: false, blockingRuns: 1, retainedBlockedRuns: 0 }); + await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, runId)); + await db.update(agents).set({ adapterType: "process" }).where(eq(agents.id, agentId)); + expect(await readUpgradeReadiness(db, input)).toMatchObject({ ready: false, blockingRuns: 1, retainedBlockedRuns: 0 }); + await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, agentId)); + input.activeRunIds.add(runId); + expect(await readUpgradeReadiness(db, { ...input, counters: () => ({ ...idle, activeRuns: 1 }) })) + .toMatchObject({ ready: false, activeRuns: 1, blockingRuns: 1, retainedBlockedRuns: 0 }); + input.activeRunIds.clear(); + let reads = 0; + expect(await readUpgradeReadiness(db, { ...input, counters: () => { + if (reads++ > 0) input.activeRunIds.add(runId); + return { ...idle, activeRuns: input.activeRunIds.size }; + } })).toMatchObject({ ready: false, blockingRuns: 1, retainedBlockedRuns: 0 }); + } finally { + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, runId)); + } + }); it("requires a bound checkpoint and current owner rather than a native label or live process alone", async () => { const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); const owner = await currentNativeControllerIdentity(); diff --git a/server/src/services/heartbeat/upgrade-readiness.ts b/server/src/services/heartbeat/upgrade-readiness.ts index e8a2d5663f3..23952159227 100644 --- a/server/src/services/heartbeat/upgrade-readiness.ts +++ b/server/src/services/heartbeat/upgrade-readiness.ts @@ -12,7 +12,7 @@ import { readNativeFinalizationUpgradeReadiness } from "../native-runtime/native const MAX_UPGRADE_RUNS = 1_000; const POST_RESULT_INSPECTION_BUDGET_MS = 2_000; type Counters = { activeRuns: number; pendingWakes: number; draining: boolean }; -type Candidate = { id: string; recoverable: boolean }; +type Candidate = { id: string; recoverable: boolean; retainedUnowned?: boolean }; /** An observation for Cloud's optional rollout policy, never execution authority. * Only explicit upgrade probes read the database; ordinary drain reads stay local. @@ -35,6 +35,13 @@ export async function readUpgradeReadiness(db: Db, input: { r.runner_profile_json->'remoteWorkspace' as "remoteWorkspace", r.runner_profile_json ? 'remoteWorkspace' as "hasRemoteWorkspace", p.id as "pluginLeaseId", f.result_id is not null as "hasResult", + (r.runtime_mode = 'native' and a.adapter_type = 'paperclip_runner' and r.native_phase = 'observed' + and f.recovery_state = 'awaiting_evidence' and f.phase = 'observed' and f.result_id is null and f.failure_code is null + and f.controller_boot_id is not null and f.controller_boot_id <> '' and f.controller_boot_id <> ${owner.bootId} + and f.controller_generation > 0 and f.lease_owner is not null and f.lease_owner <> '' + and f.lease_expires_at <= clock_timestamp() + and r.process_pid is null and r.process_started_at is null and c.run_id is null + ) is true as "retainedUnowned", case when pg_column_size(c.material) <= ${NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES} and sum(pg_column_size(c.material)) over () <= ${NATIVE_CHECKPOINT_INSPECTION_TOTAL_MAX_BYTES} then c.material else null end as material, ( r.runtime_mode = 'native' and a.adapter_type = 'paperclip_runner' @@ -79,7 +86,7 @@ export async function readUpgradeReadiness(db: Db, input: { && performance.now() < inspectionDeadline ? await readNativeFinalizationUpgradeReadiness(db, { companyId: row.companyId, runId: row.id, controller: owner, supportedOutageMs: 300_000 }).catch(() => false) : false; - candidates.push({ id: row.id, recoverable: bound && (row.recoverable || finalization) }); + candidates.push({ id: row.id, recoverable: bound && (row.recoverable || finalization), retainedUnowned: row.retainedUnowned }); } // Additive evidence does not strengthen or redefine the existing ready flag. const retained = await readNativeControllerRecoveryInventory(db).catch(() => ({ @@ -89,7 +96,8 @@ export async function readUpgradeReadiness(db: Db, input: { const summary = summarizeUpgradeReadiness({ rows: candidates, before, after: input.counters(), activeRunIds: [...new Set([...beforeIds, ...input.activeRunIds])], controllerBootId: owner.bootId }); return { ...summary, nativeControllerRecovery: { activeExecutionAdoption: { - recoverableRuns: summary.recoverableRuns, blockingRuns: summary.blockingRuns, truncated: summary.truncated, + recoverableRuns: summary.recoverableRuns, blockingRuns: summary.blockingRuns, + retainedBlockedRuns: summary.retainedBlockedRuns, truncated: summary.truncated, }, retained } }; } @@ -97,10 +105,13 @@ export function summarizeUpgradeReadiness(input: { rows: readonly Candidate[]; before: Counters; after: Counters; activeRunIds: readonly string[]; controllerBootId: string; }) { - const eligible = new Set(input.rows.filter(row => row.recoverable).map(row => row.id)); - const observedIds = new Set(input.rows.map(row => row.id)); + const activeIds = new Set(input.activeRunIds); + const retained = input.rows.filter(row => row.retainedUnowned === true && !activeIds.has(row.id)); + const executions = input.rows.filter(row => row.retainedUnowned !== true || activeIds.has(row.id)); + const eligible = new Set(executions.filter(row => row.recoverable).map(row => row.id)); + const observedIds = new Set(executions.map(row => row.id)); const unknown = input.activeRunIds.filter(id => !observedIds.has(id)).length; - const activeRuns = Math.max(input.rows.length + unknown, input.before.activeRuns, input.after.activeRuns); + const activeRuns = Math.max(executions.length + unknown, input.before.activeRuns, input.after.activeRuns); const recoverableRuns = eligible.size; const blockingRuns = activeRuns - recoverableRuns; const pendingWakes = Math.max(input.before.pendingWakes, input.after.pendingWakes); @@ -108,6 +119,6 @@ export function summarizeUpgradeReadiness(input: { const truncated = input.rows.length > MAX_UPGRADE_RUNS; return { version: 1 as const, observedAt: new Date().toISOString(), controllerBootId: input.controllerBootId, ready: !draining && !truncated && blockingRuns === 0 && pendingWakes === 0, - activeRuns, recoverableRuns, blockingRuns, pendingWakes, draining, truncated, + activeRuns, recoverableRuns, blockingRuns, retainedBlockedRuns: retained.length, pendingWakes, draining, truncated, supportedOutageMs: 300_000, recoveryGraceMs: REMOTE_RUNNER_RECONNECT_GRACE_MS }; } diff --git a/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts b/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts index 928fc9dcc5c..fad566bd33b 100644 --- a/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts +++ b/server/src/services/native-runtime/native-accepted-result-lease-settlement.ts @@ -103,7 +103,7 @@ export async function requestNativeAcceptedResultLeaseOperatorAction(db: Db, rec if (!held) return; await requestNativeControllerStopConfirmation(db, { runId: record.recordId, companyId: record.companyId, issueId: record.issueId, - controllerBootId: stamp.bootId, controllerPid: stamp.pid, + controllerBootId: stamp.bootId, controllerPid: stamp.pid, controllerHostHash: null, controllerProcessStartedAt: stamp.processStartedAt ? new Date(stamp.processStartedAt) : null, }); } diff --git a/server/src/services/native-runtime/native-controller-host-recovery.test.ts b/server/src/services/native-runtime/native-controller-host-recovery.test.ts index 1d3c4c61e52..dec4b4ffdcf 100644 --- a/server/src/services/native-runtime/native-controller-host-recovery.test.ts +++ b/server/src/services/native-runtime/native-controller-host-recovery.test.ts @@ -77,4 +77,28 @@ describe("native controller rows saved before their host was recorded", () => { await expect(confirmNativeControllerStopped({ db, companyId, issueId, actionId: actions[0]!.id, runId, stopEvidence, actorId: "board-user" })).rejects.toMatchObject({ status: 409 }); }); + + it("binds stop confirmation to an exact foreign-host controller and rejects changed ownership", async () => { + const { companyId, issueId, runId, coordinator } = await seed(); + const owner = { ...coordinator, controllerHostHash: "previous-pid-space" }; + await db.update(nativeRunFinalizations).set({ controllerHostHash: owner.controllerHostHash }) + .where(eq(nativeRunFinalizations.runId, runId)); + await requestNativeControllerStopConfirmation(db, owner); + const [action] = await db.select().from(issueRecoveryActions).where(eq(issueRecoveryActions.sourceIssueId, issueId)); + const input = { db, companyId, issueId, runId, actionId: action!.id, actorId: "board-user", + stopEvidence: "The hosting provider confirms the exact previous container stopped." }; + for (const patch of [{ controllerHostHash: "changed-space" }, { controllerBootId: randomUUID() }, + { controllerPid: 1_999_999_999 }, { controllerProcessStartedAt: new Date(owner.controllerProcessStartedAt!.getTime() + 60_000) }]) { + await db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, runId)); + await expect(confirmNativeControllerStopped(input)).rejects.toMatchObject({ status: 409 }); + await db.update(nativeRunFinalizations).set({ controllerHostHash: owner.controllerHostHash, + controllerBootId: owner.controllerBootId, controllerPid: owner.controllerPid, + controllerProcessStartedAt: owner.controllerProcessStartedAt }).where(eq(nativeRunFinalizations.runId, runId)); + } + await expect(confirmNativeControllerStopped(input)).resolves.toEqual({ runId, status: "queued" }); + const [attested] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + expect(await evaluateNativeControllerTakeover({ owner: attested!, now: new Date() })) + .toEqual({ allowed: true, reason: "operator_attested_controller_stopped" }); + expect((await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, runId)))[0]?.status).toBe("running"); + }); }); diff --git a/server/src/services/native-runtime/native-controller-host-recovery.ts b/server/src/services/native-runtime/native-controller-host-recovery.ts index 84b482dcc83..ae424e1741e 100644 --- a/server/src/services/native-runtime/native-controller-host-recovery.ts +++ b/server/src/services/native-runtime/native-controller-host-recovery.ts @@ -5,12 +5,6 @@ import { conflict } from "../../errors.js"; import { persistActivity, publishActivity } from "../activity-log.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; -/** - * An ephemeral (multi-host) server cannot judge a native controller row saved - * before controller_host_hash existed: its PID may name a process on another - * host. Restart recovery leaves such a row waiting and asks the board, through - * this recovery action, to confirm that the previous server stopped. - */ export const NATIVE_CONTROLLER_HOST_UNVERIFIED = "native_controller_host_unverified"; /** controller_host_hash value written once the board confirms the stop. */ export const OPERATOR_ATTESTED_CONTROLLER_HOST = "operator_attested"; @@ -20,7 +14,7 @@ const changed = () => conflict("The recorded controller is no longer current. Re { code: "native_controller_host_recovery_stale" }); type Coordinator = Pick; + "runId" | "companyId" | "issueId" | "controllerBootId" | "controllerPid" | "controllerProcessStartedAt" | "controllerHostHash">; /** Idempotent. Another active recovery action on the task keeps precedence. */ export async function requestNativeControllerStopConfirmation(tx: Db, coordinator: Coordinator) { @@ -29,9 +23,10 @@ export async function requestNativeControllerStopConfirmation(tx: Db, coordinato status: "active", ownerType: "board", cause: NATIVE_CONTROLLER_HOST_UNVERIFIED, fingerprint: fingerprint(coordinator.runId), attemptCount: 1, evidence: { runId: coordinator.runId, controller: { bootId: coordinator.controllerBootId, pid: coordinator.controllerPid, + hostHash: coordinator.controllerHostHash, processStartedAt: coordinator.controllerProcessStartedAt?.toISOString() ?? null } }, - nextAction: "This run's previous server was recorded before Paperclip stored which host it ran on, so Paperclip " - + "cannot verify that it stopped. Confirm that the previous server process has stopped (for example, its " + nextAction: "Paperclip cannot verify that this run's previous server stopped from the current host. " + + "Confirm that the previous server process has stopped (for example, its " + "container was replaced), and the run resumes from its saved state.", }).onConflictDoNothing(); } @@ -57,13 +52,15 @@ export async function confirmNativeControllerStopped(input: { const recorded = (action?.evidence?.controller ?? {}) as Record; if (!coordinator || !action || !run || action.cause !== NATIVE_CONTROLLER_HOST_UNVERIFIED || action.ownerType !== "board" || !["active", "escalated"].includes(action.status) || action.evidence?.runId !== input.runId - || coordinator.controllerHostHash !== null || coordinator.controllerPid !== recorded.pid - || coordinator.controllerBootId !== recorded.bootId) throw changed(); + || coordinator.controllerHostHash !== (recorded.hostHash ?? null) || coordinator.controllerPid !== recorded.pid + || coordinator.controllerBootId !== recorded.bootId + || (coordinator.controllerProcessStartedAt?.toISOString() ?? null) !== recorded.processStartedAt) throw changed(); // Work intent before the write it announces (sleep-aware background work). await notifyLiveRunRecovery(tx, { companyId: input.companyId, runId: input.runId, agentId: run.agentId }); const now = new Date(); const updated = await tx.update(nativeRunFinalizations).set({ controllerHostHash: OPERATOR_ATTESTED_CONTROLLER_HOST, updatedAt: now }) - .where(and(eq(nativeRunFinalizations.runId, input.runId), isNull(nativeRunFinalizations.controllerHostHash), + .where(and(eq(nativeRunFinalizations.runId, input.runId), coordinator.controllerHostHash === null + ? isNull(nativeRunFinalizations.controllerHostHash) : eq(nativeRunFinalizations.controllerHostHash, coordinator.controllerHostHash), eq(nativeRunFinalizations.controllerPid, coordinator.controllerPid!), eq(nativeRunFinalizations.controllerBootId, coordinator.controllerBootId!))) .returning({ runId: nativeRunFinalizations.runId }); diff --git a/server/src/services/native-runtime/native-restart-recovery.test.ts b/server/src/services/native-runtime/native-restart-recovery.test.ts index 3f88219608a..71406202ae5 100644 --- a/server/src/services/native-runtime/native-restart-recovery.test.ts +++ b/server/src/services/native-runtime/native-restart-recovery.test.ts @@ -1,4 +1,4 @@ -import { currentProcessSpaceHash } from "../controller-process-identity.js"; +import { currentControllerBootId, currentProcessSpaceHash } from "../controller-process-identity.js"; import { afterEach, describe, expect, it, vi } from "vitest"; import { @@ -297,6 +297,14 @@ describe("native controller takeover fencing", () => { now, isProcessAlive: () => true })).resolves.toEqual({ allowed: true, reason: "operator_attested_controller_stopped" }); }); + it("preserves same-space process proof when boot metadata is present", async () => { + const local = { ...owner(), controllerBootId: currentControllerBootId(), controllerHostHash: currentProcessSpaceHash() }; + expect(await evaluateNativeControllerTakeover({ owner: local, now, isProcessAlive: () => false })) + .toEqual({ allowed: true, reason: "controller_process_dead" }); + expect(await evaluateNativeControllerTakeover({ owner: local, now, isProcessAlive: () => true, + readProcessStartedAt: async () => recordedStart })).toEqual({ allowed: false, reason: "controller_still_alive" }); + }); + it.each([ ["another host", "another-host-pid-space"], ["an unreadable PID space", "unknown"], diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 23f9ba7c2e6..14d3954a745 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,4 +1,6 @@ import { readPendingNativeRunStop } from "./native-cancellation-request.js"; +import { CONTROLLER_LEASE_RENEW_MS } from "../controller-instance-lease.js"; +import { currentControllerProcessStamp, UNKNOWN_PROCESS_SPACE } from "../controller-process-identity.js"; import { readNativeSessionResumptionAgent, restoreNativeSessionExecution } from "./native-session-resumption-admission.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { NATIVE_RETAINED_OWNER_WAIT } from "./native-controller-recovery.js"; @@ -401,6 +403,8 @@ export async function evaluateNativeControllerTakeover(input: { * existed: those keep the previous local proof for one release, outside * ephemeral mode only (see LEGACY_HOST_FALLBACK_EVENT). */ controllerHostHash?: string | null; + controllerBootId?: string | null; + updatedAt?: Date; }; now: Date; coordinatedPreviousController?: { @@ -454,15 +458,19 @@ export async function evaluateNativeControllerTakeover(input: { pidExists: input.isProcessAlive ?? base.pidExists, readStartedAt: input.readProcessStartedAt ?? base.readStartedAt, }; + const needsBootProof = owner.controllerHostHash != null + && (owner.controllerHostHash !== probe.processSpace || owner.controllerHostHash === UNKNOWN_PROCESS_SPACE); const proof = await proveOwnerGone({ kind: "local", scope: "native_controller", generation: null, host: null, processSpace: recordsHost ? owner.controllerHostHash ?? null : probe.processSpace, processSpaceUnrecorded: recordsHost && owner.controllerHostHash == null, - bootId: null, pid: priorPid, processStartedAt: owner.controllerProcessStartedAt?.toISOString() ?? null, + bootId: needsBootProof ? owner.controllerBootId ?? null : null, touchedAt: owner.updatedAt?.toISOString() ?? null, + pid: priorPid, processStartedAt: owner.controllerProcessStartedAt?.toISOString() ?? null, processGroupId: null, }, probe); const fallback = proof.legacyHostFallback ? { legacyHostFallback: true as const } : {}; if (proof.verdict === "gone") { + if (proof.reason === "controller_fenced" || proof.reason === "controller_stopped") return { allowed: true, reason: proof.reason }; if (proof.reason === "pid_reused") return { allowed: true, reason: "controller_pid_recycled", ...fallback }; return { allowed: true, @@ -473,6 +481,7 @@ export async function evaluateNativeControllerTakeover(input: { ...fallback, }; } + if (proof.reason === "controller_lease_held" || proof.reason === "controller_lease_grace") return { allowed: false, reason: proof.reason }; if (proof.reason === "process_space_unrecorded") return { allowed: false, reason: "controller_host_unrecorded" }; if (proof.reason === "different_process_space" || proof.reason === "unknown_process_space" || proof.reason === "different_host") { @@ -1187,7 +1196,7 @@ export async function claimNativeRestartRecoveries(input: { input.coordinatedPreviousController ?? null, }); await recordLegacyHostFallbackProof(tx as unknown as Db, row.run, takeover, "native_restart_recovery"); - if (!takeover.allowed && takeover.reason === "controller_host_unrecorded") { + if (!takeover.allowed && ["controller_host_unrecorded", "controller_host_unprovable"].includes(takeover.reason)) { await requestNativeControllerStopConfirmation(tx as unknown as Db, row.coordinator); } if (!takeover.allowed) { @@ -1217,7 +1226,8 @@ export async function claimNativeRestartRecoveries(input: { // Repeated scans of one ownership wait record it once, so the // bounded history keeps the drain record the next claim reads. ...(waitAlreadyRecorded ? {} : { recoveryHistory: appendBoundedRecoveryHistory(event) }), - updatedAt: now, + updatedAt: waitAlreadyRecorded && row.coordinator.recoveryState === "awaiting_evidence" + && row.coordinator.recoveryRequestId === (input.recoveryRequestId ?? null) ? row.coordinator.updatedAt : now, }) .where( and( @@ -1230,7 +1240,9 @@ export async function claimNativeRestartRecoveries(input: { kind: "awaiting_evidence", runId: row.run.id, reason: takeover.reason, - ...(durableAuthority && row.coordinator.leaseExpiresAt ? { retryAt: row.coordinator.leaseExpiresAt } : {}), + ...(durableAuthority && row.coordinator.leaseExpiresAt ? { retryAt: row.coordinator.leaseExpiresAt } + : ["controller_lease_held", "controller_lease_grace"].includes(takeover.reason) + ? { retryAt: new Date(now.getTime() + CONTROLLER_LEASE_RENEW_MS) } : {}), } as const; } @@ -1648,6 +1660,7 @@ export async function claimNativeRestartRecoveries(input: { // The attempt the executor runs this resume as (see its execution claim). const progressAttempt = nextNativeProviderAttempt(providerAttempt, claimKind, !resumeKeepsAttempt); await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId }); + currentControllerProcessStamp(); const claimed = await tx .update(nativeRunFinalizations) .set({ diff --git a/server/src/services/native-runtime/native-runner-checkpoint.ts b/server/src/services/native-runtime/native-runner-checkpoint.ts index f4b47b94e9c..feb6e6832ac 100644 --- a/server/src/services/native-runtime/native-runner-checkpoint.ts +++ b/server/src/services/native-runtime/native-runner-checkpoint.ts @@ -1,4 +1,5 @@ import { readNativeCheckpointRetirement } from "./native-runner-checkpoint-retirement.js"; +import { NATIVE_PREPARATION_FENCING_VERSION } from "./native-active-runs.js"; import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js"; import { and, eq, exists, gt, isNull, lte, inArray, notExists, or, sql } from "drizzle-orm"; import { checkpointDb, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db"; @@ -349,10 +350,12 @@ export async function listExpiredNativeRunnerAuthorities(db: Db, now: Date, runI eq(heartbeatRuns.status, "running"), eq(nativeRunFinalizations.phase, "observed"), isNull(nativeRunFinalizations.resultId), or(isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)), - exists(db.select({ runId: nativeRunnerCheckpoints.runId }).from(nativeRunnerCheckpoints).where(and( + or(exists(db.select({ runId: nativeRunnerCheckpoints.runId }).from(nativeRunnerCheckpoints).where(and( eq(nativeRunnerCheckpoints.runId, nativeRunFinalizations.runId), eq(nativeRunnerCheckpoints.issueId, nativeRunFinalizations.issueId), - eq(nativeRunnerCheckpoints.companyId, nativeRunFinalizations.companyId)))))); + eq(nativeRunnerCheckpoints.companyId, nativeRunFinalizations.companyId)))), + and(sql`${heartbeatRuns.runnerProfileJson}->'nativePreparationFencingVersion' = ${JSON.stringify(NATIVE_PREPARATION_FENCING_VERSION)}::jsonb`, + isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), isNull(heartbeatRuns.processStartedAt))))); return rows.map(row => row.runId); } diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 26b9a82da58..8b8ebc17c71 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -18,6 +18,7 @@ import { agents, authUsers, companies, + controllerInstances, companyMemberships, completionContracts, environmentLeases, @@ -80,6 +81,8 @@ import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority. import { RUNNER_BRIDGE_READ_OPERATIONS, RUNNER_BRIDGE_SCHEMAS } from "./runner-bridge-tools.js"; import { HttpError } from "../../errors.js"; import { legacyControllerBootId } from "../legacy-controller-lease.js"; +import { onControllerStamp } from "../controller-process-identity.js"; +import { registerControllerInstanceDb, resetControllerInstanceLeaseForTests } from "../controller-instance-lease.js"; const mockCaptureRunFailure = vi.hoisted(() => vi.fn()); vi.mock("../../sentry.js", async () => { @@ -2277,6 +2280,42 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); ]); }); + it("recovers a preparation with no runner PID once its foreign controller stops", async () => { + const fixture = await seedRun("REMOTE-PREPARATION-OWNER"); + const oldBoot = randomUUID(); + await fixture.db.update(heartbeatRuns).set({ + runnerProfileJson: { ...fixture.preparedProfile, nativeWorkspaceSync: remotePreparedWorkspace() }, + }).where(eq(heartbeatRuns.id, fixture.runId)); + await fixture.db.update(nativeRunFinalizations).set({ controllerBootId: oldBoot, controllerPid: 7, + controllerProcessStartedAt: new Date(Date.now() - 600_000), controllerHostHash: "previous-pid-space", + leaseOwner: "previous-preparation", leaseExpiresAt: new Date(0), attempt: 1, controllerGeneration: 1, + }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + await fixture.db.insert(controllerInstances).values({ bootId: oldBoot, processSpace: "previous-pid-space", + pid: 7, leaseExpiresAt: new Date(Date.now() + 60_000) }); + registerControllerInstanceDb(fixture.db); + const stamped = vi.fn(), unsubscribe = onControllerStamp(stamped); + try { + const [waiting] = await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, + restartKind: "hard", runIds: [fixture.runId] }); + expect(waiting).toMatchObject({ kind: "awaiting_evidence", reason: "controller_lease_held", retryAt: expect.any(Date) }); + expect(stamped).not.toHaveBeenCalled(); + const [before] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + await claimNativeRestartRecoveries({ db: fixture.db, controller: successor, restartKind: "hard", runIds: [fixture.runId] }); + const [after] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(after!.updatedAt).toEqual(before!.updatedAt); + await fixture.db.update(controllerInstances).set({ stoppedAt: new Date() }).where(eq(controllerInstances.bootId, oldBoot)); + const dispositions = (await Promise.all([successor, { ...successor, bootId: randomUUID() }].map(controller => + claimNativeRestartRecoveries({ db: fixture.db, controller, restartKind: "hard", runIds: [fixture.runId] })))).flat(); + expect(dispositions.filter(row => row.kind === "bootstrap_incomplete")).toHaveLength(1); + expect(dispositions.filter(row => row.kind === "awaiting_evidence")).toHaveLength(1); + expect(stamped).toHaveBeenCalledOnce(); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run).toMatchObject({ status: "running", processPid: null, nativeSessionId: fixture.native.normalizedSessionId }); + const [owner] = await fixture.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(owner).toMatchObject({ attempt: 1, controllerGeneration: 2 }); + } finally { unsubscribe(); resetControllerInstanceLeaseForTests(); } + }); + it("recovers an unclaimed remote startup once without requiring a provider checkpoint", async () => { const fixture = await seedRun("REMOTE-UNCLAIMED"); await fixture.db.update(heartbeatRuns).set({ diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index a7b8873c3a0..ed4ffb7cae8 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -303,7 +303,7 @@ import { noteNativeRunnerEvidence } from "./native-runner-evidence-signal.js"; import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js"; import { registerLiveRunnerGoalController } from "../runner-goal-control-broker.js"; import { applyRunnerGoalPrpEvent } from "../runner-goals.js"; -import { currentProcessSpaceHash } from "../controller-process-identity.js"; +import { currentControllerProcessStamp, currentProcessSpaceHash } from "../controller-process-identity.js"; import { proveOwnerGone } from "../owner-proof.js"; import { NativeRunnerOwnershipUnverifiedError, @@ -9996,6 +9996,7 @@ async function executePaperclipNativeSessionWithinScope( } const leaseNow = new Date(); confirmedLeaseDeadline = performance.now() + executionLeaseTtlMs; + currentControllerProcessStamp(); const claimed = await tx .update(nativeRunFinalizations) .set({