From a56b648602a15f3b147a30799edf3ce1022a69d3 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Wed, 11 Nov 2026 13:55:47 -0600 Subject: [PATCH] fix(upgrade-readiness): report restartable runs and a handoff verdict that blocks only on lost work (#20910) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A deployer can replace the server while agent runs continue. Before the swap, it reads the upgrade-readiness report of the old server > - The version 1 report is ready only when every active run is adoptable and no wake admission is in flight > - On a busy instance, new runs arrive each minute. Some runs are always in preparation, so the report is almost never ready and the instance does not upgrade > - A run in preparation has done no agent work. Restart recovery starts it again. A swap loses nothing > - This pull request adds a `handoff` object to the report. It separates recoverable, restartable and blocking runs > - The benefit is that a deployer can wait only for work that a swap would lose ## Linked Issues or Issue Description Refs #20852 (reference only; this change does not depend on it and does not use its version 2 shape). **What happened?** A busy instance had 16 running runs and 1 queued run. Fourteen had a live process. Two or three native runs had not launched a process yet. One was a legacy run. `upgradeReadiness.ready` needs `blockingRuns === 0 && pendingWakes === 0` in one probe. That state almost never occurs, so the deployer never swaps the server. **Expected behavior** The report tells the deployer which runs a swap would lose. Runs that only wait for a launch, and wake admissions in flight, do not block a graceful replacement. **Steps to reproduce** 1. Keep a steady flow of new runs on an instance. 2. Read `GET /api/instance/task-drain?upgradeReadiness=1` repeatedly. 3. See `ready: false` each time, because one or more runs are in preparation or a wake admission is in flight. ## What Changed - `summarizeUpgradeReadiness` returns an additive `handoff` object: `version`, `ready`, `recoverableRuns`, `restartableRuns`, `blockingRuns`, `pendingWakes`, `unsavedWakes`. - `handoff.ready` is true when the instance is not draining, the census is complete, and `handoff.blockingRuns` and `handoff.unsavedWakes` are 0. Saved pending wakes do not change it. - The wake queue tracks each public wake admission until its wake request and queued run commit. `unsavedWakes` is the count of admissions before that commit. Without a valid count, every pending wake is unsaved. - The top-level version 1 fields keep their values and meaning. `ready` and `blockingRuns` are unchanged, so a current consumer sees no looser verdict. - The census query marks a run `restartable` only with all of these: `paperclip_runner` adapter; no process identity; no provider session before or after; no checkpoint header; no dispatched runner task; no execution-start event. It must also be a native run in the `observed` phase with preparation fencing and an unsettled finalization record, or a legacy run in the `preparing` stage under a recorded controller. - All other runs without adoption evidence stay blocking. This includes local adapters, legacy runs with a child process, follow-up runs that carry a provider session, and runs the census could not observe. - The OpenAPI description and `doc/testing-native-remote-continuity.md` describe the new fields and their limits. Why these classes are correct on master: - Restartable native runs: restart recovery claims a run with no checkpoint and no provider evidence as `bootstrap_incomplete` and executes the same run again (`native-restart-recovery.ts` `classifyNativeRunnerRecoveryEvidence`; `heartbeat/recovery.ts` executes each claim). A graceful shutdown suspends the run and notifies recovery. Existing tests: "recovers an unclaimed remote startup once without requiring a provider checkpoint" and "recovers an unclaimed remote startup that a graceful shutdown suspended during preparation". - Restartable legacy runs: process-loss recovery retries a `preparing` run with `providerWorkStarted: false`. Existing test: "continues a chain whose preparations were killed twice and then interrupted again, without a Retry". - Pending wakes: the count is wake admissions in flight in the process, not queue rows. A wake commits its wake request and its queued run in one transaction. After the commit, the next server dispatches the queued run at startup, so a saved wake does not block. Before the commit, nothing is saved. Shutdown waits at most 30 seconds and then exits, so an unsaved wake can be lost and blocks. ## Verification - `cd server && npx vitest run src/services/heartbeat/upgrade-readiness.test.ts src/services/heartbeat/upgrade-readiness-post-result.test.ts`: 28 passed. - Red/green: before the implementation, the 7 new tests failed (6 summary tests, 1 PostgreSQL census test). After it, all pass. - Red/green for unsaved wakes: a test holds the agent row lock while a wake admission waits. Without the fix, `handoff.ready` was true with no saved wake. With the fix, it is false with `unsavedWakes: 1`, and the wake saves after the lock releases. - New cases: only recoverable runs; recoverable plus unlaunched runs plus pending wakes (`handoff.ready` true, top-level `ready` false and `blockingRuns` unchanged); a started run without recovery evidence; an unobserved execution; draining; truncated census. - The PostgreSQL test changes one fact at a time (process identity, provider session, phase, fencing, checkpoint, adapter, each execution-start event) and expects the run to become blocking. - `cd server && npx vitest run src/__tests__/openapi-routes.test.ts src/__tests__/instance-settings-routes.test.ts`: 157 passed (187 with the two readiness files). - `cd server && npx tsc --noEmit`: passed. - `pnpm verify` exited 2 (a documentation path is a global CI input). Required CI supplies the broad coverage. - Not run: a live server replacement. The classification relies on the existing recovery tests named above. ## Risks - Low risk for current consumers. The change is additive and the version 1 fields are unchanged. - `handoff.ready` is an observation at one instant. A wake admission that starts after the probe is unsaved until it commits. - Restart recovery applies its own proofs and a bounded attempt budget before it starts a restartable run again. The report is an observation, not a guarantee for each run. - The probe adds one indexed `not exists` on run events per running row, inside the existing two-second statement timeout. > For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and discuss it in `#dev` before opening the PR. Feature PRs that overlap with planned core work may need to be redirected — check the roadmap first. See `CONTRIBUTING.md`. ## Model Used - Anthropic Claude Fable 5.2 (`claude-fable-5-2`), with tool use and code execution in Claude Code. ## 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 - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5.2 --- doc/testing-native-remote-continuity.md | 35 +++++ server/src/routes/openapi.ts | 2 +- server/src/services/heartbeat.ts | 4 +- server/src/services/heartbeat/queue.ts | 17 +- .../heartbeat/upgrade-readiness.test.ts | 147 +++++++++++++++++- .../services/heartbeat/upgrade-readiness.ts | 50 +++++- 6 files changed, 245 insertions(+), 10 deletions(-) diff --git a/doc/testing-native-remote-continuity.md b/doc/testing-native-remote-continuity.md index f4a64d19ad8..2e98d99556f 100644 --- a/doc/testing-native-remote-continuity.md +++ b/doc/testing-native-remote-continuity.md @@ -173,6 +173,41 @@ inspection retains its SQL statement timeout. Header reads are capped at 256 KiB oversized, malformed, legacy-format or local evidence falls back to the ordinary drain. This bounded header probe does not validate every journal record or prove a live sandbox. This is an observation; the existing shutdown and startup owners still perform the actual handoff and enforce Stop and fencing. + +The report also carries an additive `handoff` object, which a deployer can adopt +in place of the top-level `ready`. The top-level version 1 fields keep their +meaning: `blockingRuns` counts every execution without adoption evidence, and +`ready` is false while any exists or an admission is in flight. `handoff` +(its own `version: 1`) separates work a graceful server replacement would lose +from work that is only in flight: + +| Field | Meaning | +| --- | --- | +| `recoverableRuns` | Same count as the top level: a successor adopts these running executions. | +| `restartableRuns` | Runs marked `running` that have launched no agent work. Restart recovery starts the same work again without a Retry. | +| `blockingRuns` | Executions that are neither. A swap would interrupt started agent work that no successor can adopt. | +| `pendingWakes` | Same count as the top level, reported for information. It does not affect `handoff.ready`. | +| `unsavedWakes` | The pending wakes that have not committed their wake request and queued run. They exist only in this process, so a swap can lose them. | +| `ready` | `true` when the instance is not draining, the census is complete, and `handoff.blockingRuns` and `handoff.unsavedWakes` are 0. | + +A run is restartable only when all of this holds: the agent uses the +`paperclip_runner` adapter; the run has no process identity, no provider session +before or after, no checkpoint header and no dispatched runner task; no +execution-start event exists (`native.process_start_requested`, +`native.process_identity_recorded`, `adapter.invoke`, `tool.execution.started` or +any runner-sourced event); and it is either a native run in the `observed` phase +with preparation fencing and an unsettled finalization record, or a legacy run +still in the `preparing` stage under a recorded controller. Follow-up runs that +carry a previous provider session, local adapters and legacy runs with a child +process stay blocking. An execution the census could not observe also blocks. + +`pendingWakes` counts wake admissions in flight inside this process, not queue +rows. An admission commits its wake request and its queued run in one +transaction; after that it only dispatches saved rows, and the next server +dispatches queued runs at startup. Before that commit, for example while it +waits on the agent row lock, nothing is saved. Shutdown waits at most 30 seconds +for in-flight admissions and then exits, so an unsaved admission can be lost. +`handoff.unsavedWakes` counts those and keeps `handoff.ready` false. A positive report alone does not prove hosted upgrade continuity. Retain Postgres and the instance secret encryption key across deployments. The diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 87eb0032948..e9e9a8c92c6 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -7345,7 +7345,7 @@ registry.registerPath({ summary: "Get process task-drain status, with optional bounded upgrade-readiness or idle-safety reports for instance operators", description: "With idleSleepSafety=1, the exact current idle owner can retire idle native sessions before a fresh safety inspection. Busy or failed retirement prevents sleep. Native retirement is skipped for diagnostic inspection. scheduledWake=domain-deadlines-v1 requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a mandatory deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate the report before stopping the instance.", - request: { query: z.object({ upgradeReadiness: z.literal("1").optional().describe("Instance operators can request a bounded aggregate report of current remote-runner recovery eligibility. This read does not hold admission or detach runners."), ownerId: z.string().uuid().optional(), idleSleepSafety: z.literal("1").optional().describe("Instance admins can request a conservative durable-work report while admission and ingress are held. Unknown or present work must prevent automatic sleep."), scheduledWake: z.enum(["1", "2", "domain-deadlines-v1"]).optional().describe("Cloud control callers may request routine deadlines (1), routines and issue monitors (2), or the domain deadline catalog (domain-deadlines-v1). The domain catalog requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate this report before stopping the instance.") }) }, + request: { query: z.object({ upgradeReadiness: z.literal("1").optional().describe("Instance operators can request a bounded aggregate report of current remote-runner recovery eligibility. This read does not hold admission or detach runners. The version 1 report also carries an additive handoff object (version, ready, recoverableRuns, restartableRuns, blockingRuns, pendingWakes, unsavedWakes) for a graceful server replacement: restartable runs have launched no agent work and are started again by restart recovery, and unsavedWakes counts wake admissions that have not committed their queued run. Only handoff.blockingRuns, handoff.unsavedWakes, draining or an incomplete census make handoff.ready false. The top-level version 1 fields keep their meaning."), ownerId: z.string().uuid().optional(), idleSleepSafety: z.literal("1").optional().describe("Instance admins can request a conservative durable-work report while admission and ingress are held. Unknown or present work must prevent automatic sleep."), scheduledWake: z.enum(["1", "2", "domain-deadlines-v1"]).optional().describe("Cloud control callers may request routine deadlines (1), routines and issue monitors (2), or the domain deadline catalog (domain-deadlines-v1). The domain catalog requires signed Cloud authority with idleCapabilities.deadlineCatalog=domain_deadlines_v1 and returns a deadlineProof even when no wake is scheduled. Cloud must acknowledge any returned durable alarm and revalidate this report before stopping the instance.") }) }, responses: { 200: r.ok(), 401: r.unauthorized, 403: r.forbidden }, }); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 954104fe5fb..7dd4ab69b46 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -105,6 +105,7 @@ import { shouldQueueFollowupForRunningIssueWake, DEFERRED_WAKE_CONTEXT_KEY, createHeartbeatQueue, + unsavedWakeAdmissionCount, } from "./heartbeat/queue.js"; export { shouldDeferFollowupWakeForSameIssue, @@ -4521,7 +4522,8 @@ export function heartbeatService( stopTaskDrain, getTaskDrainStatus, prepareIdleWarmNativeSessionsForSleep, - readUpgradeReadiness: () => readUpgradeReadiness(db, { activeRunIds: activeRunExecutions, counters: getTaskDrainStatus }), + readUpgradeReadiness: () => readUpgradeReadiness(db, { activeRunIds: activeRunExecutions, + counters: () => ({ ...getTaskDrainStatus(), unsavedWakes: unsavedWakeAdmissionCount() }) }), computeTaskDrain, applyTaskDrain, diff --git a/server/src/services/heartbeat/queue.ts b/server/src/services/heartbeat/queue.ts index c949cfd5384..18f9c8b4b28 100644 --- a/server/src/services/heartbeat/queue.ts +++ b/server/src/services/heartbeat/queue.ts @@ -402,6 +402,11 @@ export async function claimDueTimerHeartbeatForAgent( /** Queue admission and dispatch share the service lifecycle callbacks and execution ownership. */ +// Wake admissions that have not committed their wake and run yet. Process-wide, like +// the pending-wake set: a server replacement loses these, and only these, pending wakes. +const unsavedWakeAdmissions = new Set(); +export function unsavedWakeAdmissionCount() { return unsavedWakeAdmissions.size; } + export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDependencies) { const { setRunStatus, @@ -2341,17 +2346,22 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende agentId: string, opts: WakeupOptions = {}, ): ReturnType { - const promise = enqueueWakeup(agentId, opts); + // Until its wake and run commit, this admission exists only in this process. + const unsaved = {}; + unsavedWakeAdmissions.add(unsaved); + const promise = enqueueWakeup(agentId, opts, undefined, undefined, { saved: () => { unsavedWakeAdmissions.delete(unsaved); } }); activeWakeupPromises.add(promise); void promise .catch(() => {}) .finally(() => { + unsavedWakeAdmissions.delete(unsaved); activeWakeupPromises.delete(promise); }); return promise; } - async function enqueueWakeup(agentId: string, opts: WakeupOptions = {}, executionWaitRequestId?: string, savedInteractionOrigin?: unknown) { + async function enqueueWakeup(agentId: string, opts: WakeupOptions = {}, executionWaitRequestId?: string, savedInteractionOrigin?: unknown, + admission?: { saved(): void }) { const source = opts.source ?? "on_demand"; const triggerDetail = opts.triggerDetail ?? null; const contextSnapshot: Record = { @@ -4768,6 +4778,8 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende throw error; }); + // The admission has committed. What follows is dispatch of saved rows. + if (outcome.kind !== "deferred" && outcome.kind !== "skipped") admission?.saved(); for (const publication of issueActivityPublications) publishActivity(publication); await executeIssuePostCommitActions(db, issuePostCommitActions); @@ -5009,6 +5021,7 @@ export function createHeartbeatQueue(db: Db, dependencies: HeartbeatQueueDepende }); if (queueOutcome.kind === "skipped") return null; + admission?.saved(); const newRun = queueOutcome.run; publishLiveEvent({ diff --git a/server/src/services/heartbeat/upgrade-readiness.test.ts b/server/src/services/heartbeat/upgrade-readiness.test.ts index e5f92444e3a..f7c515274ae 100644 --- a/server/src/services/heartbeat/upgrade-readiness.test.ts +++ b/server/src/services/heartbeat/upgrade-readiness.test.ts @@ -3,7 +3,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { randomUUID } from "node:crypto"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agents, applyPendingMigrations, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRuns, issues, environmentLeases, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db"; +import { agents, agentWakeupRequests, applyPendingMigrations, companies, createDb, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, heartbeatRunEvents, heartbeatRuns, issues, environmentLeases, nativeRunFinalizations, nativeRunnerCheckpoints } from "@paperclipai/db"; import type { Db } from "@paperclipai/db"; import { eq, sql } from "drizzle-orm"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; @@ -48,6 +48,60 @@ describe("upgrade readiness", () => { expect(report({ rows, before }).ready).toBe(false); } }); + describe("handoff verdict", () => { + const busy = (rows: Parameters[0]["rows"], pendingWakes = 0, unsavedWakes = 0) => report({ rows, + activeRunIds: rows.map(row => row.id), before: { ...idle, activeRuns: rows.length, pendingWakes, unsavedWakes }, + after: { ...idle, activeRuns: rows.length, pendingWakes, unsavedWakes } }); + it("agrees with the version 1 verdict when every execution is recoverable", () => { + expect(busy([{ id: "a", recoverable: true }, { id: "b", recoverable: true }])).toMatchObject({ version: 1, ready: true, + blockingRuns: 0, handoff: { version: 1, ready: true, recoverableRuns: 2, restartableRuns: 0, blockingRuns: 0, pendingWakes: 0, unsavedWakes: 0 } }); + }); + it("is ready with unlaunched runs and saved pending wakes, and leaves the version 1 fields unchanged", () => { + const rows = [{ id: "live", recoverable: true }, { id: "preparing", recoverable: false, restartable: true }, + { id: "launching", recoverable: false, restartable: true }]; + expect(busy(rows, 2)).toMatchObject({ version: 1, ready: false, activeRuns: 3, recoverableRuns: 1, blockingRuns: 2, + pendingWakes: 2, handoff: { version: 1, ready: true, recoverableRuns: 1, restartableRuns: 2, blockingRuns: 0, pendingWakes: 2, unsavedWakes: 0 } }); + }); + it("is not ready while a pending wake has not saved its wake and run", () => { + const rows = [{ id: "live", recoverable: true }]; + expect(busy(rows, 2, 1).handoff).toMatchObject({ ready: false, blockingRuns: 0, pendingWakes: 2, unsavedWakes: 1 }); + // Seen in either read of the counters. + expect(report({ rows, activeRunIds: ["live"], before: { ...idle, activeRuns: 1, pendingWakes: 1, unsavedWakes: 1 }, + after: { ...idle, activeRuns: 1 } }).handoff).toMatchObject({ ready: false, unsavedWakes: 1 }); + }); + it("treats every pending wake as unsaved without a valid saved count", () => { + const rows = [{ id: "live", recoverable: true }]; + for (const unsavedWakes of [undefined, -1, 3, 0.5, Number.NaN]) { + expect(report({ rows, activeRunIds: ["live"], before: { ...idle, activeRuns: 1, pendingWakes: 2, unsavedWakes }, + after: { ...idle, activeRuns: 1 } }).handoff, String(unsavedWakes)).toMatchObject({ ready: false, pendingWakes: 2, unsavedWakes: 2 }); + } + }); + it("is not ready while a started execution has no recovery evidence", () => { + const rows = [{ id: "live", recoverable: true }, { id: "preparing", recoverable: false, restartable: true }, + { id: "legacy-child", recoverable: false }]; + expect(busy(rows)).toMatchObject({ ready: false, blockingRuns: 2, + handoff: { ready: false, recoverableRuns: 1, restartableRuns: 1, blockingRuns: 1 } }); + }); + it("counts an execution this report could not observe as a blocker", () => { + const rows = [{ id: "preparing", recoverable: false, restartable: true }]; + expect(report({ rows, activeRunIds: ["preparing", "unseen"], after: { ...idle, activeRuns: 2 } }).handoff) + .toMatchObject({ ready: false, restartableRuns: 1, blockingRuns: 1 }); + expect(report({ rows, activeRunIds: ["preparing"], after: { ...idle, activeRuns: 2 } }).handoff) + .toMatchObject({ ready: false, restartableRuns: 1, blockingRuns: 1 }); + }); + it("never counts one execution in two classes", () => { + expect(busy([{ id: "a", recoverable: true, restartable: true }]).handoff) + .toMatchObject({ ready: true, recoverableRuns: 1, restartableRuns: 0, blockingRuns: 0 }); + }); + it("stays not ready while draining or truncated", () => { + const rows = [{ id: "preparing", recoverable: false, restartable: true }]; + expect(report({ rows, activeRunIds: ["preparing"], before: { ...idle, activeRuns: 1, draining: true } }).handoff) + .toMatchObject({ ready: false, blockingRuns: 0 }); + const many = Array.from({ length: 1_001 }, (_, i) => ({ id: String(i), recoverable: false, restartable: true })); + expect(report({ rows: many, activeRunIds: many.map(row => row.id), before: { ...idle, activeRuns: many.length }, + after: { ...idle, activeRuns: many.length } })).toMatchObject({ truncated: true, handoff: { ready: false } }); + }); + }); }); const externalUrl = resolveExternalTestDatabaseUrl("PAPERCLIP_UPGRADE_READINESS_TEST_DATABASE_URL"); @@ -259,6 +313,97 @@ postgresTests("upgrade readiness persisted execution evidence", () => { expect((await readUpgradeReadiness(db, input)).ready).toBe(false); // the in-flight local tail is still unknown expect(await readUpgradeReadiness(db, { activeRunIds: new Set(), counters: () => idle })).toMatchObject({ ready: true, activeRuns: 0 }); }); + it("classifies a run with no launched agent work as restartable, and any started work as a blocker", async () => { + const companyId = randomUUID(), agentId = randomUUID(), issueId = randomUUID(), runId = randomUUID(); + const owner = await currentNativeControllerIdentity(); + await db.insert(companies).values({ id: companyId, name: "Restartable", issuePrefix: `S${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Restartable", role: "engineer", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Preparing", status: "in_progress", assigneeAgentId: agentId }); + const unlaunched = { runtimeMode: "native", nativePhase: "observed", processPid: null, processGroupId: null, + processStartedAt: null, sessionIdBefore: null, sessionIdAfter: null, executionStage: null, controllerBootId: null, + runnerProfileJson: { nativePreparationFencingVersion: 1 } }; + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, issueId, nativeIssueId: issueId, status: "running", + runnerInstanceId: randomUUID(), nativeSessionId: randomUUID(), ...unlaunched }); + const claimed = { controllerBootId: owner.bootId, leaseOwner: "controller-lease", controllerGeneration: 1, + leaseExpiresAt: new Date(Date.now() + 60_000), phase: "observed", failureCode: null }; + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, ...claimed }); + const input = { activeRunIds: new Set([runId]), counters: () => ({ ...idle, activeRuns: 1, pendingWakes: 1, unsavedWakes: 0 }) }; + const restartable = { ready: false, activeRuns: 1, recoverableRuns: 0, blockingRuns: 1, pendingWakes: 1, + handoff: { version: 1, ready: true, recoverableRuns: 0, restartableRuns: 1, blockingRuns: 0, pendingWakes: 1 } }; + const blocking = { ready: false, blockingRuns: 1, handoff: { ready: false, restartableRuns: 0, blockingRuns: 1 } }; + try { + expect(await readUpgradeReadiness(db, input)).toMatchObject(restartable); + for (const patch of [{ processPid: process.pid }, { processGroupId: process.pid }, { processStartedAt: new Date() }, + { sessionIdBefore: "provider-session" }, { sessionIdAfter: "provider-session" }, { nativePhase: "result_accepted" }, + { runnerProfileJson: {} }, { runtimeMode: "legacy" }]) { + await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, runId)); + expect(await readUpgradeReadiness(db, input), JSON.stringify(patch)).toMatchObject(blocking); + await db.update(heartbeatRuns).set(unlaunched).where(eq(heartbeatRuns.id, runId)); + } + for (const patch of [{ failureCode: "uncertain" }, { phase: "workspace_finalizing" }]) { + await db.update(nativeRunFinalizations).set(patch).where(eq(nativeRunFinalizations.runId, runId)); + expect(await readUpgradeReadiness(db, input), JSON.stringify(patch)).toMatchObject(blocking); + await db.update(nativeRunFinalizations).set(claimed).where(eq(nativeRunFinalizations.runId, runId)); + } + await db.update(agents).set({ adapterType: "process" }).where(eq(agents.id, agentId)); + expect(await readUpgradeReadiness(db, input)).toMatchObject(blocking); + await db.update(agents).set({ adapterType: "paperclip_runner" }).where(eq(agents.id, agentId)); + await db.insert(nativeRunnerCheckpoints).values({ runId, companyId, issueId, part: "header", material: {} }); + expect(await readUpgradeReadiness(db, input)).toMatchObject(blocking); + await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, runId)); + // The executor records this request before any backend may start a runner. + for (const event of [{ eventType: "native.process_start_requested" }, { eventType: "adapter.invoke" }, + { eventType: "lifecycle", sourceEventId: "runner-event-1" }]) { + const [{ id }] = await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1, ...event }) + .returning({ id: heartbeatRunEvents.id }); + expect(await readUpgradeReadiness(db, input), event.eventType).toMatchObject(blocking); + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.id, id)); + } + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1, eventType: "lifecycle" }); + expect(await readUpgradeReadiness(db, input)).toMatchObject(restartable); + // A direct-adapter preparation that has not reached dispatch is requeued by process-loss recovery. + const legacyPreparing = { runtimeMode: "legacy", nativePhase: null, executionStage: "preparing", controllerBootId: randomUUID() }; + await db.update(heartbeatRuns).set(legacyPreparing).where(eq(heartbeatRuns.id, runId)); + expect(await readUpgradeReadiness(db, input)).toMatchObject(restartable); + for (const patch of [{ executionStage: "dispatching" }, { controllerBootId: null }, { processPid: process.pid }]) { + await db.update(heartbeatRuns).set(patch).where(eq(heartbeatRuns.id, runId)); + expect(await readUpgradeReadiness(db, input), JSON.stringify(patch)).toMatchObject(blocking); + await db.update(heartbeatRuns).set({ ...legacyPreparing, processPid: null }).where(eq(heartbeatRuns.id, runId)); + } + } finally { + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); + await db.update(heartbeatRuns).set({ status: "failed" }).where(eq(heartbeatRuns.id, runId)); + } + }); + it("blocks the handoff while a wake admission waits on the agent lock with nothing saved", async () => { + const companyId = randomUUID(), agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Unsaved wake", issuePrefix: `W${companyId.slice(0, 6)}` }); + await db.insert(agents).values({ id: agentId, companyId, name: "Unsaved wake", role: "engineer", adapterType: "process", + status: "idle", adapterConfig: { command: "true" } }); + const { heartbeatService } = await import("../heartbeat.js"); + const heartbeat = heartbeatService(db); + const saved = () => db.select({ id: agentWakeupRequests.id }).from(agentWakeupRequests).where(eq(agentWakeupRequests.agentId, agentId)); + const lock = await db.$client.reserve(); + let wake: Promise | undefined; + try { + await lock.unsafe("begin"); + await lock.unsafe(`select id from agents where id = '${agentId}' for update`); + wake = heartbeat.wakeup(agentId, { source: "on_demand", triggerDetail: "manual", + requestedByActorType: "user", requestedByActorId: "readiness-user" }); + await vi.waitFor(async () => { + const locks = await db.execute(sql`select 1 from pg_stat_activity where wait_event_type = 'Lock' and query like '%from agents where id =%for update%'`); + expect(locks.length).toBeGreaterThan(0); + }, { timeout: 20_000, interval: 100 }); + expect(await saved()).toEqual([]); + expect((await heartbeat.readUpgradeReadiness()).handoff).toMatchObject({ ready: false, pendingWakes: 1, unsavedWakes: 1 }); + } finally { + await lock.unsafe("rollback"); lock.release(); + await wake; + await heartbeat.drainActiveRunExecutions(); + } + expect(await saved()).toHaveLength(1); + expect((await heartbeat.readUpgradeReadiness()).handoff).toMatchObject({ pendingWakes: 0, unsavedWakes: 0 }); + }, 60_000); it("cancels a locked probe within its SQL budget and releases its connection", async () => { const probeDb = createDb(database.connectionString, { maxConnections: 1 }); const lock = await db.$client.reserve(); diff --git a/server/src/services/heartbeat/upgrade-readiness.ts b/server/src/services/heartbeat/upgrade-readiness.ts index 23952159227..92ee60ee807 100644 --- a/server/src/services/heartbeat/upgrade-readiness.ts +++ b/server/src/services/heartbeat/upgrade-readiness.ts @@ -8,11 +8,17 @@ import { REMOTE_RUNNER_RECONNECT_GRACE_MS } from "../native-runtime/native-runne import { currentNativeControllerIdentity } from "../native-runtime/native-restart-recovery.js"; import { readNativeControllerRecoveryInventory } from "../native-runtime/native-controller-recovery-inventory.js"; import { readNativeFinalizationUpgradeReadiness } from "../native-runtime/native-finalization-upgrade-readiness.js"; +import { NATIVE_PREPARATION_FENCING_VERSION } from "../native-runtime/native-active-runs.js"; +import { PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED } from "../native-local-process-stop.js"; 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; retainedUnowned?: boolean }; +/** `unsavedWakes`: pending wakes whose wake and run have not committed. A caller + * that does not report it leaves every pending wake unsaved. */ +type Counters = { activeRuns: number; pendingWakes: number; unsavedWakes?: number; draining: boolean }; +/** `restartable`: the row and its events show no launched agent work, so a replacement + * server starts the same work again. It is weaker than `recoverable` and never implies it. */ +type Candidate = { id: string; recoverable: boolean; restartable?: 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. @@ -52,7 +58,23 @@ export async function readUpgradeReadiness(db: Db, input: { and f.lease_owner is not null and f.controller_generation > 0 and f.phase = 'observed' and f.result_id is null and f.failure_code is null and c.run_id is not null - ) is true as recoverable + ) is true as recoverable, ( + -- Nothing has reached a runner or a provider: no process identity, no provider + -- session, no checkpoint, no dispatched runner task and no execution-start event + -- (the executor records the start request before any backend may spawn a runner). + -- Restart recovery proves the same facts before it starts such a run again. + a.adapter_type = 'paperclip_runner' + and r.process_pid is null and r.process_group_id is null and r.process_started_at is null + and r.session_id_before is null and r.session_id_after is null + and c.run_id is null and p.id is null + and ((r.runtime_mode = 'native' and r.native_phase = 'observed' + and r.runner_profile_json->>'nativePreparationFencingVersion' = ${String(NATIVE_PREPARATION_FENCING_VERSION)} + and f.phase = 'observed' and f.result_id is null and f.failure_code is null) + or (r.runtime_mode = 'legacy' and r.execution_stage = 'preparing' and r.controller_boot_id is not null)) + and not exists (select 1 from heartbeat_run_events e + where e.company_id = r.company_id and e.run_id = r.id and (e.source_event_id is not null + or e.event_type in ('adapter.invoke', ${PROCESS_START_REQUESTED}, ${PROCESS_IDENTITY_RECORDED}, 'tool.execution.started'))) + ) is true as restartable from heartbeat_runs r join agents a on a.id = r.agent_id and a.company_id = r.company_id left join native_run_finalizations f on f.run_id = r.id and f.company_id = r.company_id left join native_runner_checkpoints c on c.run_id = r.id and c.company_id = r.company_id @@ -86,7 +108,8 @@ 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), retainedUnowned: row.retainedUnowned }); + candidates.push({ id: row.id, recoverable: bound && (row.recoverable || finalization), + restartable: row.restartable === true, retainedUnowned: row.retainedUnowned }); } // Additive evidence does not strengthen or redefine the existing ready flag. const retained = await readNativeControllerRecoveryInventory(db).catch(() => ({ @@ -117,8 +140,25 @@ export function summarizeUpgradeReadiness(input: { const pendingWakes = Math.max(input.before.pendingWakes, input.after.pendingWakes); const draining = input.before.draining || input.after.draining; const truncated = input.rows.length > MAX_UPGRADE_RUNS; + // Additive verdict for a graceful server replacement. It separates work a swap would + // lose from work that is only in flight, and leaves the version 1 fields above as they were. + // - Restartable runs have launched no agent work; restart recovery starts them again. + // - Pending wakes are admissions in flight in this process. A saved one has committed + // its queued run, which the next server dispatches at startup. An unsaved one exists + // only in memory: shutdown waits a bounded time for it, so it can be lost and blocks. + // - Everything else without recovery evidence, including an execution this report + // could not observe, would lose work and blocks. + const restartableRuns = executions.filter(row => row.restartable === true && !eligible.has(row.id)).length; + const handoffBlockingRuns = blockingRuns - restartableRuns; + const unsavedWakes = Math.max(...[input.before, input.after].map(counters => { + const unsaved = counters.unsavedWakes ?? counters.pendingWakes; + return Number.isSafeInteger(unsaved) && unsaved >= 0 && unsaved <= counters.pendingWakes ? unsaved : counters.pendingWakes; + })); + const handoff = { version: 1 as const, + ready: !draining && !truncated && handoffBlockingRuns === 0 && unsavedWakes === 0, + recoverableRuns, restartableRuns, blockingRuns: handoffBlockingRuns, pendingWakes, unsavedWakes }; return { version: 1 as const, observedAt: new Date().toISOString(), controllerBootId: input.controllerBootId, ready: !draining && !truncated && blockingRuns === 0 && pendingWakes === 0, 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, handoff }; }