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 }; }