mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-10 12:37:55 +01:00
fix: recover interrupted native preparations and unblock upgrades (#20177)
## Thinking Path > - Paperclip manages AI-agent companies and their work. > - A native run prepares its workspace before a runner PID or checkpoint exists. > - A server crash can interrupt that stage and leave its controller on another host. > - Recovery needs a durable owner proof, a delivered recovery wake, and the saved execution contract. > - The old paths missed controller lease registration and boot proof, skipped checkpoint-free preparations, and counted retained work as live upgrade blockers. > - This PR connects preparation to the existing controller lease and recovery paths, and separates retained work from the upgrade census. > - The result is bounded recovery without a second executor or loss of saved work. ## Linked Issues or Issue Description **What happened?** A native preparation interrupted by a server crash could remain running indefinitely with no runner PID or checkpoint. Native controller stamps did not wake the shared instance lease. That lease was disabled on volume-backed Railway services, and native takeover omitted the boot binding needed for its cross-host proof. Scoped recovery selected checkpoint-backed runs but skipped these preparations. The upgrade probe then counted the retained row as current execution and could block every later upgrade. **Expected behavior** Track the preparation's controller before a runner starts. Recover it through the normal classifier after proving its owner stopped or fencing that boot. Keep live or unprovable owners held. Deliver recovery wakes to preparations without checkpoints. Preserve unresolved historical work and report it separately from current execution during upgrade observations. **Steps to reproduce** 1. Save an observed native preparation with no runner process, result, provider-start evidence or checkpoint. 2. Replace its controller with a process in another PID space. Let its owner lease expire and send a scoped recovery wake. 3. Before this fix, recovery skips the checkpoint-free preparation or cannot prove its foreign controller stopped. Upgrade readiness continues to count that old row as a live blocker. Related work: #18037, #19651, #19868, and #20168. The foreign-PID guard from #19868 remains. The queued first-turn input fix from #20168 is included through the current base. ## What Changed - Register the existing instance lease when native execution and restart claims are written. Keep passive identity reads free of ownership effects. Keep that lease enabled on volume-backed Railway services unless explicitly disabled. - Preserve the existing PID/start-time proof within the same readable PID space. Bind foreign or unknown-space takeover to its saved boot and last write. A live lease or grace period keeps the run held with a scheduled check. Repeated evidence waits preserve their timestamp. - Include explicitly fenced, checkpoint-free preparations in the existing finite recovery candidate inventory. Normal launch, checkpoint, provider evidence and admission checks still decide whether work may resume. - Expose the existing board stop-confirmation operation for historical foreign-host owners. Bind confirmation to the exact run, boot, PID, process start and host. Reject changed ownership. - Report narrowly defined old, unowned preparations as `retainedBlockedRuns` outside the live upgrade census. Local executions, pending admission, drain state, truncation and positive checkpoint proof retain their barriers. - Update the execution contract and add PostgreSQL, real-route, race and preservation regressions. ## Verification - Genuine red/green regressions cover the old upgrade census, missing native lease registration, the volume-backed default, foreign-host confirmation, and scoped recovery skipping checkpoint-free preparation. Test bodies ran against real PostgreSQL before their corresponding production changes. - Combined verification before the final idle correction (head `055be6c84d9`): **459 passed, 22 skipped, 11 files passed**. Covers controller lease/identity, native restart classification and integration, host confirmation, heartbeat recovery, instance route authorization, upgrade readiness, and queued first-turn input. The focused command used `pnpm exec vitest run` with those 11 suites and `--maxWorkers=1 --no-file-parallelism`. The 22 skipped cases require the Rust debug fake-Codex fixture, which was not built locally; the real-process CI coverage must pass. - `pnpm --filter @paperclipai/server typecheck`: passed on head `055be6c84d9b058064daad0af71bb5d893ab9529`. - Actual route proof uses the real heartbeat service and PostgreSQL. Unauthorized access returns 403. The retained run and owner records remain unchanged. Recovery integration admits one bootstrap claim across two replacement controllers, keeps a live owner held, and preserves local execution. - `pnpm verify`: exit 2, requests broad CI because the execution contract changed. Full local suite not run. Current-head CI must supply broad coverage. - Local source checkout on macOS ARM64, Node v25.6.1, pnpm 9.15.4. Tests use isolated PostgreSQL. No live provider restart is claimed by those fixtures. - CI found an idle-query regression in the first version of the lease wiring. Two focused PostgreSQL regressions reproduced the misplaced identity-read notification and the missing claim notification. The fix moves registration to actual ownership writes. `env -u npm_execpath pnpm exec vitest run server/src/__tests__/maintenance-server-routes.test.ts --maxWorkers=1 --no-file-parallelism`: **17 passed**, including 90 real seconds and two qualified hourly windows with no extra idle queries. An initial local setup attempt inherited npm metadata and did not run tests; clearing that metadata selected the correct pnpm helper. Three focused controller/native suites also passed **168 tests, 22 fixture-gated skips**. Server typecheck passed after the correction on head `4a527326dcd5b3a9e286f0e4c3e761d3056af4d9`. - The older CI server shard also exposed a same-space graceful-restart regression. The new boot fallback had overridden local process-exit proof. Seven suspension cases and a focused unit regression reproduced it; the fix limits boot fallback to foreign or unknown PID spaces. Fresh verification on `acf1065d98150210c1e227c7f3b05a1eb489730b`: **251 passed, 22 fixture-gated skips, 4 files passed**, including all **40** graceful-restart suspension tests. Server typecheck passed. The local regression failed before this correction, and the same-space and foreign-space paths pass together after it. - Current-head CI completed successfully on `acf1065d98150210c1e227c7f3b05a1eb489730b` (run 42568109784). Greptile reviewed that exact head at 5/5 with no unresolved review threads. - Secret and private-data scans passed. No workflow, lockfile or migration changes. ## Risks - The instance lease now applies to volume-backed Railway services. Its existing 30-second renewal, quiescence, grace and self-fence behavior applies while work is held. Idle servers retain the existing no-query behavior. An explicit `PAPERCLIP_CONTROLLER_INSTANCE_LEASE=off` remains available. - A controller lease proves only the server owner's state. It does not prove an agent process stopped or authorize replay, resource release or volume removal. The existing classifier and provider/runner checks remain responsible for those decisions. - Previous-release records keep their shape and saved state. A missing boot lease stays unprovable and requires the scoped stop-confirmation path. A changed confirmation target is rejected. - The upgrade observation does not recover a held task by itself. Source tests do not qualify a hosted replacement. Deployment and the affected task's useful result must be verified separately. - This repairs recovery after a crash; it does not claim to identify or fix every source of heap exhaustion. ## Model Used OpenAI `gpt-6.1-sol`, xhigh reasoning effort, through Codex in T3 Code, with repository, shell, and GitHub tools. Context window size is not exposed by the runtime. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge --------- Co-authored-by: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
2d93413085
commit
9b71d03215
14 files changed
+301
-42
No files matched your search
@@ -1933,10 +1933,32 @@ judged through that boot's row:
|
|||||||
- no row (a boot from before the lease): the record stays unprovable.
|
- no row (a boot from before the lease): the record stays unprovable.
|
||||||
|
|
||||||
A record with a recorded process group stays unprovable across hosts, because
|
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
|
agent processes may outlive their server. The lease is enabled on volume-backed
|
||||||
deployment with an exclusive volume, which proves a replaced deployment from its
|
Railway services too: a process can crash and restart inside the same deployment,
|
||||||
own identity. A stateless Railway service keeps it.
|
and some saved ownership records do not contain a deployment identity.
|
||||||
`PAPERCLIP_CONTROLLER_INSTANCE_LEASE=on|off` overrides this.
|
`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`,
|
A record saved before its owner was fully recorded (`no_owner_stamp`,
|
||||||
`missing_identity`, `process_space_unrecorded`) stays unprovable and is never
|
`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
|
header evidence. That inventory is preliminary, reports partial coverage, and
|
||||||
does not certify workspace durability or volume removal.
|
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
|
`importNativeControllerRecovery` is an internal one-source operation for the
|
||||||
migration owner while old controller files remain available. It requires the
|
migration owner while old controller files remain available. It requires the
|
||||||
admitted configuration digest and exact supervisor snapshot, locks the issue,
|
admitted configuration digest and exact supervisor snapshot, locks the issue,
|
||||||
|
|||||||
@@ -13,6 +13,7 @@ import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../services/environment-lease-
|
|||||||
import { proveOwnerGone, type LocalOwnerStamp } from "../services/owner-proof.js";
|
import { proveOwnerGone, type LocalOwnerStamp } from "../services/owner-proof.js";
|
||||||
import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
|
import { createOwnershipReportState, runOwnershipReconciliationPass } from "../services/recovery/ownership-invariant-report.js";
|
||||||
import { beginIdleTrackedWork, readTaskDrain } from "../services/task-admission.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 support = await getEmbeddedPostgresTestSupport();
|
||||||
const suite = support.supported ? describe : describe.skip;
|
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" });
|
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 () => {
|
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 justExpired = await otherBoot({ expiredMsAgo: 5_000 });
|
||||||
const resumed = await otherBoot({ expiredMsAgo: 600_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");
|
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 startHere();
|
||||||
await stopControllerInstanceLease(db);
|
await stopControllerInstanceLease(db);
|
||||||
expect(await proveControllerInstance(db, currentControllerBootId(), null)).toBe("stopped");
|
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);
|
expect(controllerInstanceLeaseEnabled()).toBe(true);
|
||||||
vi.stubEnv("RAILWAY_VOLUME_ID", "v");
|
vi.stubEnv("RAILWAY_VOLUME_ID", "v");
|
||||||
vi.stubEnv("RAILWAY_VOLUME_MOUNT_PATH", "/data");
|
vi.stubEnv("RAILWAY_VOLUME_MOUNT_PATH", "/data");
|
||||||
expect(controllerInstanceLeaseEnabled()).toBe(false);
|
expect(controllerInstanceLeaseEnabled()).toBe(true);
|
||||||
vi.stubEnv("PAPERCLIP_CONTROLLER_INSTANCE_LEASE", "on");
|
vi.stubEnv("PAPERCLIP_CONTROLLER_INSTANCE_LEASE", "on");
|
||||||
expect(controllerInstanceLeaseEnabled()).toBe(true);
|
expect(controllerInstanceLeaseEnabled()).toBe(true);
|
||||||
vi.unstubAllEnvs();
|
vi.unstubAllEnvs();
|
||||||
|
|||||||
@@ -2,7 +2,7 @@ import { and, eq, isNull, sql } from "drizzle-orm";
|
|||||||
import { controllerInstances, type Db } from "@paperclipai/db";
|
import { controllerInstances, type Db } from "@paperclipai/db";
|
||||||
import { logger } from "../middleware/logger.js";
|
import { logger } from "../middleware/logger.js";
|
||||||
import {
|
import {
|
||||||
currentControllerBootId, currentControllerDeployment, currentProcessSpaceHash, onControllerStamp,
|
currentControllerBootId, currentProcessSpaceHash, onControllerStamp,
|
||||||
} from "./controller-process-identity.js";
|
} from "./controller-process-identity.js";
|
||||||
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
import type { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js";
|
||||||
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
import { DELIVERY_QUEUES } from "./delivery-work-notifications.js";
|
||||||
@@ -54,14 +54,8 @@ type LeaseState = {
|
|||||||
let state: LeaseState | null = null;
|
let state: LeaseState | null = null;
|
||||||
let registeredDb: Db | 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 {
|
export function controllerInstanceLeaseEnabled(env: NodeJS.ProcessEnv = process.env): boolean {
|
||||||
const setting = env.PAPERCLIP_CONTROLLER_INSTANCE_LEASE;
|
return env.PAPERCLIP_CONTROLLER_INSTANCE_LEASE !== "off";
|
||||||
if (setting === "off") return false;
|
|
||||||
if (setting === "on") return true;
|
|
||||||
return currentControllerDeployment(env)?.exclusiveVolume !== true;
|
|
||||||
}
|
}
|
||||||
|
|
||||||
function exitOnLostLease(reason: "fenced" | "lease_lost") {
|
function exitOnLostLease(reason: "fenced" | "lease_lost") {
|
||||||
|
|||||||
@@ -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];
|
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 () => {
|
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 run = await seedRun(), leaseId = randomUUID(), environmentId = randomUUID(), leaseOwner = randomUUID();
|
||||||
const deadlineAtMs = Date.now() - 1_000, contextId = randomUUID();
|
const deadlineAtMs = Date.now() - 1_000, contextId = randomUUID();
|
||||||
|
|||||||
@@ -36,6 +36,17 @@ describe("upgrade readiness", () => {
|
|||||||
it("never treats a truncated census as complete", () => {
|
it("never treats a truncated census as complete", () => {
|
||||||
const rows = Array.from({ length: 1_001 }, (_, i) => ({ id: String(i), recoverable: true }));
|
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, 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);
|
db = createDb(database.connectionString);
|
||||||
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
}, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS);
|
||||||
afterAll(async () => { await db?.$client.end(); await database?.cleanup(); vi.unstubAllEnvs(); });
|
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<string>(), 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 () => {
|
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 companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID();
|
||||||
const owner = await currentNativeControllerIdentity();
|
const owner = await currentNativeControllerIdentity();
|
||||||
|
|||||||
@@ -12,7 +12,7 @@ import { readNativeFinalizationUpgradeReadiness } from "../native-runtime/native
|
|||||||
const MAX_UPGRADE_RUNS = 1_000;
|
const MAX_UPGRADE_RUNS = 1_000;
|
||||||
const POST_RESULT_INSPECTION_BUDGET_MS = 2_000;
|
const POST_RESULT_INSPECTION_BUDGET_MS = 2_000;
|
||||||
type Counters = { activeRuns: number; pendingWakes: number; draining: boolean };
|
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.
|
/** An observation for Cloud's optional rollout policy, never execution authority.
|
||||||
* Only explicit upgrade probes read the database; ordinary drain reads stay local.
|
* 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 "remoteWorkspace",
|
||||||
r.runner_profile_json ? 'remoteWorkspace' as "hasRemoteWorkspace", p.id as "pluginLeaseId",
|
r.runner_profile_json ? 'remoteWorkspace' as "hasRemoteWorkspace", p.id as "pluginLeaseId",
|
||||||
f.result_id is not null as "hasResult",
|
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}
|
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, (
|
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'
|
r.runtime_mode = 'native' and a.adapter_type = 'paperclip_runner'
|
||||||
@@ -79,7 +86,7 @@ export async function readUpgradeReadiness(db: Db, input: {
|
|||||||
&& performance.now() < inspectionDeadline
|
&& performance.now() < inspectionDeadline
|
||||||
? await readNativeFinalizationUpgradeReadiness(db, { companyId: row.companyId, runId: row.id,
|
? await readNativeFinalizationUpgradeReadiness(db, { companyId: row.companyId, runId: row.id,
|
||||||
controller: owner, supportedOutageMs: 300_000 }).catch(() => false) : false;
|
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.
|
// Additive evidence does not strengthen or redefine the existing ready flag.
|
||||||
const retained = await readNativeControllerRecoveryInventory(db).catch(() => ({
|
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(),
|
const summary = summarizeUpgradeReadiness({ rows: candidates, before, after: input.counters(),
|
||||||
activeRunIds: [...new Set([...beforeIds, ...input.activeRunIds])], controllerBootId: owner.bootId });
|
activeRunIds: [...new Set([...beforeIds, ...input.activeRunIds])], controllerBootId: owner.bootId });
|
||||||
return { ...summary, nativeControllerRecovery: { activeExecutionAdoption: {
|
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 } };
|
}, retained } };
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -97,10 +105,13 @@ export function summarizeUpgradeReadiness(input: {
|
|||||||
rows: readonly Candidate[]; before: Counters; after: Counters;
|
rows: readonly Candidate[]; before: Counters; after: Counters;
|
||||||
activeRunIds: readonly string[]; controllerBootId: string;
|
activeRunIds: readonly string[]; controllerBootId: string;
|
||||||
}) {
|
}) {
|
||||||
const eligible = new Set(input.rows.filter(row => row.recoverable).map(row => row.id));
|
const activeIds = new Set(input.activeRunIds);
|
||||||
const observedIds = new Set(input.rows.map(row => row.id));
|
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 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 recoverableRuns = eligible.size;
|
||||||
const blockingRuns = activeRuns - recoverableRuns;
|
const blockingRuns = activeRuns - recoverableRuns;
|
||||||
const pendingWakes = Math.max(input.before.pendingWakes, input.after.pendingWakes);
|
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;
|
const truncated = input.rows.length > MAX_UPGRADE_RUNS;
|
||||||
return { version: 1 as const, observedAt: new Date().toISOString(), controllerBootId: input.controllerBootId,
|
return { version: 1 as const, observedAt: new Date().toISOString(), controllerBootId: input.controllerBootId,
|
||||||
ready: !draining && !truncated && blockingRuns === 0 && pendingWakes === 0,
|
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 };
|
supportedOutageMs: 300_000, recoveryGraceMs: REMOTE_RUNNER_RECONNECT_GRACE_MS };
|
||||||
}
|
}
|
||||||
@@ -103,7 +103,7 @@ export async function requestNativeAcceptedResultLeaseOperatorAction(db: Db, rec
|
|||||||
if (!held) return;
|
if (!held) return;
|
||||||
await requestNativeControllerStopConfirmation(db, {
|
await requestNativeControllerStopConfirmation(db, {
|
||||||
runId: record.recordId, companyId: record.companyId, issueId: record.issueId,
|
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,
|
controllerProcessStartedAt: stamp.processStartedAt ? new Date(stamp.processStartedAt) : null,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
@@ -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,
|
await expect(confirmNativeControllerStopped({ db, companyId, issueId, actionId: actions[0]!.id, runId,
|
||||||
stopEvidence, actorId: "board-user" })).rejects.toMatchObject({ status: 409 });
|
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");
|
||||||
|
});
|
||||||
});
|
});
|
||||||
@@ -5,12 +5,6 @@ import { conflict } from "../../errors.js";
|
|||||||
import { persistActivity, publishActivity } from "../activity-log.js";
|
import { persistActivity, publishActivity } from "../activity-log.js";
|
||||||
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.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";
|
export const NATIVE_CONTROLLER_HOST_UNVERIFIED = "native_controller_host_unverified";
|
||||||
/** controller_host_hash value written once the board confirms the stop. */
|
/** controller_host_hash value written once the board confirms the stop. */
|
||||||
export const OPERATOR_ATTESTED_CONTROLLER_HOST = "operator_attested";
|
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" });
|
{ code: "native_controller_host_recovery_stale" });
|
||||||
|
|
||||||
type Coordinator = Pick<typeof nativeRunFinalizations.$inferSelect,
|
type Coordinator = Pick<typeof nativeRunFinalizations.$inferSelect,
|
||||||
"runId" | "companyId" | "issueId" | "controllerBootId" | "controllerPid" | "controllerProcessStartedAt">;
|
"runId" | "companyId" | "issueId" | "controllerBootId" | "controllerPid" | "controllerProcessStartedAt" | "controllerHostHash">;
|
||||||
|
|
||||||
/** Idempotent. Another active recovery action on the task keeps precedence. */
|
/** Idempotent. Another active recovery action on the task keeps precedence. */
|
||||||
export async function requestNativeControllerStopConfirmation(tx: Db, coordinator: Coordinator) {
|
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,
|
status: "active", ownerType: "board", cause: NATIVE_CONTROLLER_HOST_UNVERIFIED,
|
||||||
fingerprint: fingerprint(coordinator.runId), attemptCount: 1,
|
fingerprint: fingerprint(coordinator.runId), attemptCount: 1,
|
||||||
evidence: { runId: coordinator.runId, controller: { bootId: coordinator.controllerBootId, pid: coordinator.controllerPid,
|
evidence: { runId: coordinator.runId, controller: { bootId: coordinator.controllerBootId, pid: coordinator.controllerPid,
|
||||||
|
hostHash: coordinator.controllerHostHash,
|
||||||
processStartedAt: coordinator.controllerProcessStartedAt?.toISOString() ?? null } },
|
processStartedAt: coordinator.controllerProcessStartedAt?.toISOString() ?? null } },
|
||||||
nextAction: "This run's previous server was recorded before Paperclip stored which host it ran on, so Paperclip "
|
nextAction: "Paperclip cannot verify that this run's previous server stopped from the current host. "
|
||||||
+ "cannot verify that it stopped. Confirm that the previous server process has stopped (for example, its "
|
+ "Confirm that the previous server process has stopped (for example, its "
|
||||||
+ "container was replaced), and the run resumes from its saved state.",
|
+ "container was replaced), and the run resumes from its saved state.",
|
||||||
}).onConflictDoNothing();
|
}).onConflictDoNothing();
|
||||||
}
|
}
|
||||||
@@ -57,13 +52,15 @@ export async function confirmNativeControllerStopped(input: {
|
|||||||
const recorded = (action?.evidence?.controller ?? {}) as Record<string, unknown>;
|
const recorded = (action?.evidence?.controller ?? {}) as Record<string, unknown>;
|
||||||
if (!coordinator || !action || !run || action.cause !== NATIVE_CONTROLLER_HOST_UNVERIFIED || action.ownerType !== "board"
|
if (!coordinator || !action || !run || action.cause !== NATIVE_CONTROLLER_HOST_UNVERIFIED || action.ownerType !== "board"
|
||||||
|| !["active", "escalated"].includes(action.status) || action.evidence?.runId !== input.runId
|
|| !["active", "escalated"].includes(action.status) || action.evidence?.runId !== input.runId
|
||||||
|| coordinator.controllerHostHash !== null || coordinator.controllerPid !== recorded.pid
|
|| coordinator.controllerHostHash !== (recorded.hostHash ?? null) || coordinator.controllerPid !== recorded.pid
|
||||||
|| coordinator.controllerBootId !== recorded.bootId) throw changed();
|
|| coordinator.controllerBootId !== recorded.bootId
|
||||||
|
|| (coordinator.controllerProcessStartedAt?.toISOString() ?? null) !== recorded.processStartedAt) throw changed();
|
||||||
// Work intent before the write it announces (sleep-aware background work).
|
// Work intent before the write it announces (sleep-aware background work).
|
||||||
await notifyLiveRunRecovery(tx, { companyId: input.companyId, runId: input.runId, agentId: run.agentId });
|
await notifyLiveRunRecovery(tx, { companyId: input.companyId, runId: input.runId, agentId: run.agentId });
|
||||||
const now = new Date();
|
const now = new Date();
|
||||||
const updated = await tx.update(nativeRunFinalizations).set({ controllerHostHash: OPERATOR_ATTESTED_CONTROLLER_HOST, updatedAt: now })
|
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.controllerPid, coordinator.controllerPid!),
|
||||||
eq(nativeRunFinalizations.controllerBootId, coordinator.controllerBootId!)))
|
eq(nativeRunFinalizations.controllerBootId, coordinator.controllerBootId!)))
|
||||||
.returning({ runId: nativeRunFinalizations.runId });
|
.returning({ runId: nativeRunFinalizations.runId });
|
||||||
|
|||||||
@@ -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 { afterEach, describe, expect, it, vi } from "vitest";
|
||||||
|
|
||||||
import {
|
import {
|
||||||
@@ -297,6 +297,14 @@ describe("native controller takeover fencing", () => {
|
|||||||
now, isProcessAlive: () => true })).resolves.toEqual({ allowed: true, reason: "operator_attested_controller_stopped" });
|
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([
|
it.each([
|
||||||
["another host", "another-host-pid-space"],
|
["another host", "another-host-pid-space"],
|
||||||
["an unreadable PID space", "unknown"],
|
["an unreadable PID space", "unknown"],
|
||||||
|
|||||||
@@ -1,4 +1,6 @@
|
|||||||
import { readPendingNativeRunStop } from "./native-cancellation-request.js";
|
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 { readNativeSessionResumptionAgent, restoreNativeSessionExecution } from "./native-session-resumption-admission.js";
|
||||||
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
|
import { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
|
||||||
import { NATIVE_RETAINED_OWNER_WAIT } from "./native-controller-recovery.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
|
* existed: those keep the previous local proof for one release, outside
|
||||||
* ephemeral mode only (see LEGACY_HOST_FALLBACK_EVENT). */
|
* ephemeral mode only (see LEGACY_HOST_FALLBACK_EVENT). */
|
||||||
controllerHostHash?: string | null;
|
controllerHostHash?: string | null;
|
||||||
|
controllerBootId?: string | null;
|
||||||
|
updatedAt?: Date;
|
||||||
};
|
};
|
||||||
now: Date;
|
now: Date;
|
||||||
coordinatedPreviousController?: {
|
coordinatedPreviousController?: {
|
||||||
@@ -454,15 +458,19 @@ export async function evaluateNativeControllerTakeover(input: {
|
|||||||
pidExists: input.isProcessAlive ?? base.pidExists,
|
pidExists: input.isProcessAlive ?? base.pidExists,
|
||||||
readStartedAt: input.readProcessStartedAt ?? base.readStartedAt,
|
readStartedAt: input.readProcessStartedAt ?? base.readStartedAt,
|
||||||
};
|
};
|
||||||
|
const needsBootProof = owner.controllerHostHash != null
|
||||||
|
&& (owner.controllerHostHash !== probe.processSpace || owner.controllerHostHash === UNKNOWN_PROCESS_SPACE);
|
||||||
const proof = await proveOwnerGone({
|
const proof = await proveOwnerGone({
|
||||||
kind: "local", scope: "native_controller", generation: null, host: null,
|
kind: "local", scope: "native_controller", generation: null, host: null,
|
||||||
processSpace: recordsHost ? owner.controllerHostHash ?? null : probe.processSpace,
|
processSpace: recordsHost ? owner.controllerHostHash ?? null : probe.processSpace,
|
||||||
processSpaceUnrecorded: recordsHost && owner.controllerHostHash == null,
|
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,
|
processGroupId: null,
|
||||||
}, probe);
|
}, probe);
|
||||||
const fallback = proof.legacyHostFallback ? { legacyHostFallback: true as const } : {};
|
const fallback = proof.legacyHostFallback ? { legacyHostFallback: true as const } : {};
|
||||||
if (proof.verdict === "gone") {
|
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 };
|
if (proof.reason === "pid_reused") return { allowed: true, reason: "controller_pid_recycled", ...fallback };
|
||||||
return {
|
return {
|
||||||
allowed: true,
|
allowed: true,
|
||||||
@@ -473,6 +481,7 @@ export async function evaluateNativeControllerTakeover(input: {
|
|||||||
...fallback,
|
...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 === "process_space_unrecorded") return { allowed: false, reason: "controller_host_unrecorded" };
|
||||||
if (proof.reason === "different_process_space" || proof.reason === "unknown_process_space"
|
if (proof.reason === "different_process_space" || proof.reason === "unknown_process_space"
|
||||||
|| proof.reason === "different_host") {
|
|| proof.reason === "different_host") {
|
||||||
@@ -1187,7 +1196,7 @@ export async function claimNativeRestartRecoveries(input: {
|
|||||||
input.coordinatedPreviousController ?? null,
|
input.coordinatedPreviousController ?? null,
|
||||||
});
|
});
|
||||||
await recordLegacyHostFallbackProof(tx as unknown as Db, row.run, takeover, "native_restart_recovery");
|
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);
|
await requestNativeControllerStopConfirmation(tx as unknown as Db, row.coordinator);
|
||||||
}
|
}
|
||||||
if (!takeover.allowed) {
|
if (!takeover.allowed) {
|
||||||
@@ -1217,7 +1226,8 @@ export async function claimNativeRestartRecoveries(input: {
|
|||||||
// Repeated scans of one ownership wait record it once, so the
|
// Repeated scans of one ownership wait record it once, so the
|
||||||
// bounded history keeps the drain record the next claim reads.
|
// bounded history keeps the drain record the next claim reads.
|
||||||
...(waitAlreadyRecorded ? {} : { recoveryHistory: appendBoundedRecoveryHistory(event) }),
|
...(waitAlreadyRecorded ? {} : { recoveryHistory: appendBoundedRecoveryHistory(event) }),
|
||||||
updatedAt: now,
|
updatedAt: waitAlreadyRecorded && row.coordinator.recoveryState === "awaiting_evidence"
|
||||||
|
&& row.coordinator.recoveryRequestId === (input.recoveryRequestId ?? null) ? row.coordinator.updatedAt : now,
|
||||||
})
|
})
|
||||||
.where(
|
.where(
|
||||||
and(
|
and(
|
||||||
@@ -1230,7 +1240,9 @@ export async function claimNativeRestartRecoveries(input: {
|
|||||||
kind: "awaiting_evidence",
|
kind: "awaiting_evidence",
|
||||||
runId: row.run.id,
|
runId: row.run.id,
|
||||||
reason: takeover.reason,
|
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;
|
} as const;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1648,6 +1660,7 @@ export async function claimNativeRestartRecoveries(input: {
|
|||||||
// The attempt the executor runs this resume as (see its execution claim).
|
// The attempt the executor runs this resume as (see its execution claim).
|
||||||
const progressAttempt = nextNativeProviderAttempt(providerAttempt, claimKind, !resumeKeepsAttempt);
|
const progressAttempt = nextNativeProviderAttempt(providerAttempt, claimKind, !resumeKeepsAttempt);
|
||||||
await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId });
|
await notifyLiveRunRecovery(tx, { companyId: row.run.companyId, runId: row.run.id, agentId: row.run.agentId });
|
||||||
|
currentControllerProcessStamp();
|
||||||
const claimed = await tx
|
const claimed = await tx
|
||||||
.update(nativeRunFinalizations)
|
.update(nativeRunFinalizations)
|
||||||
.set({
|
.set({
|
||||||
|
|||||||
@@ -1,4 +1,5 @@
|
|||||||
import { readNativeCheckpointRetirement } from "./native-runner-checkpoint-retirement.js";
|
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 { notifyLiveRunRecovery } from "../live-run-recovery-notifications.js";
|
||||||
import { and, eq, exists, gt, isNull, lte, inArray, notExists, or, sql } from "drizzle-orm";
|
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";
|
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"),
|
eq(heartbeatRuns.status, "running"), eq(nativeRunFinalizations.phase, "observed"),
|
||||||
isNull(nativeRunFinalizations.resultId),
|
isNull(nativeRunFinalizations.resultId),
|
||||||
or(isNull(nativeRunFinalizations.leaseExpiresAt), lte(nativeRunFinalizations.leaseExpiresAt, now)),
|
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.runId, nativeRunFinalizations.runId),
|
||||||
eq(nativeRunnerCheckpoints.issueId, nativeRunFinalizations.issueId),
|
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);
|
return rows.map(row => row.runId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import {
|
|||||||
agents,
|
agents,
|
||||||
authUsers,
|
authUsers,
|
||||||
companies,
|
companies,
|
||||||
|
controllerInstances,
|
||||||
companyMemberships,
|
companyMemberships,
|
||||||
completionContracts,
|
completionContracts,
|
||||||
environmentLeases,
|
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 { RUNNER_BRIDGE_READ_OPERATIONS, RUNNER_BRIDGE_SCHEMAS } from "./runner-bridge-tools.js";
|
||||||
import { HttpError } from "../../errors.js";
|
import { HttpError } from "../../errors.js";
|
||||||
import { legacyControllerBootId } from "../legacy-controller-lease.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());
|
const mockCaptureRunFailure = vi.hoisted(() => vi.fn());
|
||||||
vi.mock("../../sentry.js", async () => {
|
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 () => {
|
it("recovers an unclaimed remote startup once without requiring a provider checkpoint", async () => {
|
||||||
const fixture = await seedRun("REMOTE-UNCLAIMED");
|
const fixture = await seedRun("REMOTE-UNCLAIMED");
|
||||||
await fixture.db.update(heartbeatRuns).set({
|
await fixture.db.update(heartbeatRuns).set({
|
||||||
|
|||||||
@@ -303,7 +303,7 @@ import { noteNativeRunnerEvidence } from "./native-runner-evidence-signal.js";
|
|||||||
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
import { removeNativeHarnessBackup } from "./native-harness-backup-cleanup.js";
|
||||||
import { registerLiveRunnerGoalController } from "../runner-goal-control-broker.js";
|
import { registerLiveRunnerGoalController } from "../runner-goal-control-broker.js";
|
||||||
import { applyRunnerGoalPrpEvent } from "../runner-goals.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 { proveOwnerGone } from "../owner-proof.js";
|
||||||
import {
|
import {
|
||||||
NativeRunnerOwnershipUnverifiedError,
|
NativeRunnerOwnershipUnverifiedError,
|
||||||
@@ -9996,6 +9996,7 @@ async function executePaperclipNativeSessionWithinScope(
|
|||||||
}
|
}
|
||||||
const leaseNow = new Date();
|
const leaseNow = new Date();
|
||||||
confirmedLeaseDeadline = performance.now() + executionLeaseTtlMs;
|
confirmedLeaseDeadline = performance.now() + executionLeaseTtlMs;
|
||||||
|
currentControllerProcessStamp();
|
||||||
const claimed = await tx
|
const claimed = await tx
|
||||||
.update(nativeRunFinalizations)
|
.update(nativeRunFinalizations)
|
||||||
.set({
|
.set({
|
||||||
|
|||||||
Reference in new issue
Block a user