diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index f8f657777a8..7672ca6362f 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2223,7 +2223,7 @@ unknown operations or partial transcripts do not. Replacement uses a fresh session and retires only the exact predecessor's obsolete recovery hold while recording the proof and successor lineage. Retained provider files are not edited. -Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. +Bootstrap retries, exact-checkpoint resumes, and fresh replacement sessions share three total provider attempts, including the original attempt. Linked run IDs, controller restarts, and duplicate wakes do not reset this budget. A resume after a graceful shutdown drain suspended the run for reattachment does not consume an attempt, because the run did not cause that restart. A resume after a crash, which leaves no drain record, still consumes one. Exact-checkpoint resumes retain the 30-second delay. Replacement scheduling and predecessor lineage commit together, with one successor per predecessor and admission through the normal task locks, authorization, pause, approval, and budget gates. A due failed remote retry can retain its provider PID after heartbeat completion releases the task execution lock. Both remote workspace reference formats and diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index 654f152bcf3..3cb7a89231b 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -5,7 +5,8 @@ import { expect, it, vi } from "vitest"; import { resolveLocalRuntimeApiUrl } from "../runtime-api.js"; import { heartbeatService } from "../services/heartbeat.js"; import { instanceSettingsService } from "../services/instance-settings.js"; -import { currentNativeControllerIdentity } from "../services/native-runtime/native-restart-recovery.js"; +import { classifyNativeRunnerRecoveryEvidence, currentNativeControllerIdentity } from "../services/native-runtime/native-restart-recovery.js"; +import type { NativeSessionBackend } from "@paperclipai/paperclip-runner"; import { PaperclipControlPlanePort } from "../services/native-runtime/paperclip-control-plane-port.js"; import { CONTROL_PLANE_CONFORMANCE_OPEN, CONTROL_PLANE_CONFORMANCE_RESULT, CONTROL_PLANE_CONFORMANCE_TERMINAL } from "../vendor/paperclip-runner/testing.js"; import { describeEmbeddedPostgres, nativeRestartFixture } from "./helpers/native-restart.js"; @@ -78,6 +79,87 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut 90_000, ); + it.each([ + // Attempt 0: no execution claimed the run yet, so its first start is counted. + { drainedAttempt: 0, charges: true, ownershipWait: false }, + { drainedAttempt: 1, charges: false, ownershipWait: false }, + // The third attempt may continue after a drain; it does not start a fourth. + { drainedAttempt: 3, charges: false, ownershipWait: false }, + // Scans that wait for the old controller to exit keep the drain record. + { drainedAttempt: 3, charges: false, ownershipWait: true }, + ])("charges a resume after a clean drain only for the first execution (attempt $drainedAttempt, ownership wait $ownershipWait)", async ({ drainedAttempt, charges, ownershipWait }) => { + // RC-22: each graceful restart cost one of the three recovery attempts, so + // infrastructure restarts the run did not cause exhausted its budget. + const f = await seedRecoverableRun(`DRAIN-BUDGET-${drainedAttempt}${ownershipWait ? "-WAIT" : ""}`); + let drained: unknown; + const heartbeat = heartbeatService(f.db, { + nativeSessionBackendFactory: vi.fn(() => { throw new Error("A suspended recovery must not open a provider session"); }), + beforeNativeRuntimeSelection: async () => { + if (!drained) drained = await heartbeat.drainRunningRunsForShutdown("SIGTERM", new Date(), [f.runId]); + }, + }); + try { + // A crash leaves no drain record, so its resume still charges an attempt. + const { claims } = await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ runId: f.runId, chargesProviderAttempt: true })]); + } finally { await heartbeat.drainActiveRunExecutions(); } + expect(drained).toMatchObject({ restartSuspendedRunIds: [f.runId] }); + + await f.db.update(nativeRunFinalizations).set({ attempt: drainedAttempt }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + if (ownershipWait) { + // The drained controller (this test process) is still alive, so the new + // server's scans wait for ownership evidence instead of claiming. + const waiting = heartbeatService(f.db, { nativeSessionBackendFactory: vi.fn(() => { throw new Error("no claim while the owner lives"); }) }); + for (let scan = 0; scan < 2; scan += 1) { + try { expect((await waiting.recoverNativeRunsAfterRestart(new Date(), [f.runId])).claims).toEqual([]); } + finally { await waiting.drainActiveRunExecutions(); } + } + const [owner] = await f.db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(owner.recoveryState).toBe("awaiting_evidence"); + expect((owner.recoveryHistory as Array<{ disposition?: string }>).slice(-2).map(entry => entry.disposition)) + .toEqual(["restart_suspended", "awaiting_evidence"]); + } + await f.db.update(nativeRunFinalizations).set({ controllerPid: 2_000_000_001 }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + // Let the executor take its execution lease, then record the attempt it + // saved when it reaches the provider session. + const savedAttempts: number[] = []; + const reachedProvider = async () => { + const [owner] = await f.db.select({ attempt: nativeRunFinalizations.attempt }).from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, f.runId)); + savedAttempts.push(owner.attempt); + throw new Error("stop at provider dispatch"); + }; + const backend = vi.fn((): NativeSessionBackend => ({ + async descriptor() { + return { kind: "mock", name: "drained-budget", version: "1", capabilities: { + resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }; + }, + openSession: reachedProvider, + recoverSession: reachedProvider, + })); + const next = heartbeatService(f.db, { nativeSessionBackendFactory: backend }); + try { + const { claims } = await next.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ runId: f.runId, chargesProviderAttempt: charges })]); + } finally { await next.drainActiveRunExecutions(); } + expect(backend).toHaveBeenCalledOnce(); + expect(savedAttempts).toEqual([charges ? drainedAttempt + 1 : drainedAttempt]); + }, 90_000); + + it.each([false, true])("still refuses a fourth attempt after a crash (remote bootstrap %s)", (remote) => { + const evidence = remote + ? { remoteSandbox: true, remoteBootstrapProvenIncomplete: true, runnerPidAlive: false, runnerGroupAlive: false, + processStartMatches: false, hasCheckpoint: false, hasProviderEvidence: false, providerAttempt: 3 } + : { runnerPidAlive: false, runnerGroupAlive: false, processStartMatches: false, + hasCheckpoint: true, checkpointIdentityMatches: true, hasProviderEvidence: true, providerAttempt: 3 }; + expect(classifyNativeRunnerRecoveryEvidence(evidence).reason).toBe("execution_recovery_budget_exhausted"); + expect(classifyNativeRunnerRecoveryEvidence({ ...evidence, resumeChargesAttempt: false }).reason) + .not.toBe("execution_recovery_budget_exhausted"); + }); + it("hands a recovery to the next controller when shutdown closed the listener before backend setup", async () => { const f = await seedRecoverableRun("LISTENER-CLOSED"); const listener = fixture.current.server!; diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index d46a0ffe566..b58403e4803 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -96,6 +96,8 @@ export type NativeRestartRecoveryClaim = providerAttempt: number; restartKind: NativeRestartKind; recoveryRequestId: string | null; + /** False when the prior controller suspended this run in a clean drain. */ + chargesProviderAttempt?: boolean; } | { kind: "bootstrap_incomplete"; @@ -105,6 +107,8 @@ export type NativeRestartRecoveryClaim = providerAttempt: number; restartKind: NativeRestartKind; recoveryRequestId: string | null; + /** False when the prior controller suspended this run in a clean drain. */ + chargesProviderAttempt?: boolean; }; export type NativeRestartRecoveryDisposition = @@ -120,13 +124,41 @@ export type NativeRestartRecoveryDisposition = export function nextNativeProviderAttempt( currentAttempt: number, recoveryKind?: NativeRestartRecoveryClaim["kind"], + chargesProviderAttempt = true, ): number { - return recoveryKind === "reattach_existing_runner" || + return !chargesProviderAttempt || + recoveryKind === "reattach_existing_runner" || recoveryKind === "reattach_remote_runner" ? currentAttempt : currentAttempt + 1; } +/** + * The execution recovery budget bounds runs that keep failing their provider + * turn. A graceful drain that suspended the run for reattachment is an + * infrastructure restart the run did not cause, so the next controller's + * resume of that turn does not spend an attempt. A crash (no drain record) + * still charges one. The drain record is consumed by the next claim; waits for + * ownership evidence or a deferred block in between do not erase it. + */ +const RECOVERY_CLAIM_DISPOSITIONS: ReadonlySet = new Set([ + "reattach_existing_runner", "reattach_remote_runner", "resume_dead_runner", "bootstrap_incomplete", +]); +export function suspendedByCleanDrain(coordinator: { + attempt: number; + recoveryHistory: ReadonlyArray<{ disposition?: unknown }>; +}): boolean { + // Attempt 0 means no execution has claimed the run yet. Its first provider + // start must still be counted, whatever interrupted preparation. + if (coordinator.attempt <= 0) return false; + for (let index = coordinator.recoveryHistory.length - 1; index >= 0; index -= 1) { + const disposition = coordinator.recoveryHistory[index]?.disposition; + if (disposition === "restart_suspended") return true; + if (RECOVERY_CLAIM_DISPOSITIONS.has(disposition)) return false; + } + return false; +} + const controllerBootId = randomUUID(); function processIsAlive(pid: number | null | undefined): boolean { @@ -465,6 +497,8 @@ export function classifyNativeRunnerRecoveryEvidence(input: { checkpointFailed?: boolean; stagedAttachmentVerified?: boolean; providerAttempt?: number; + /** False when resuming keeps the current attempt (a clean drain). */ + resumeChargesAttempt?: boolean; }): { claimKind: NativeRestartRecoveryClaim["kind"] | null; reason: string; @@ -486,7 +520,7 @@ export function classifyNativeRunnerRecoveryEvidence(input: { } if (input.remoteBootstrapProvenIncomplete === true && !input.hasCheckpoint && !input.hasProviderEvidence) { // Each same-run bootstrap is another provider attempt; keep restarts bounded. - if ((input.providerAttempt ?? 0) >= 3) + if (input.resumeChargesAttempt !== false && (input.providerAttempt ?? 0) >= 3) return { claimKind: null, reason: "execution_recovery_budget_exhausted" }; return { claimKind: "bootstrap_incomplete", reason: input.remoteBootstrapIncompleteReason ?? "remote_bootstrap_proven_incomplete" }; @@ -526,7 +560,8 @@ export function classifyNativeRunnerRecoveryEvidence(input: { reason: "live_provider_process_identity_unverifiable", }; } - if ((input.providerAttempt ?? 0) >= 3) + // Continuing an attempt already counted needs no budget; a new one does. + if (input.resumeChargesAttempt !== false && (input.providerAttempt ?? 0) >= 3) return { claimKind: null, reason: "execution_recovery_budget_exhausted" }; const checkpointIdentityMatches = input.checkpointIdentityMatches ?? input.hasCheckpoint; @@ -935,6 +970,8 @@ export async function claimNativeRestartRecoveries(input: { input.coordinatedPreviousController ?? null, }); if (!takeover.allowed) { + const previous = row.coordinator.recoveryHistory.at(-1); + const waitAlreadyRecorded = previous?.disposition === "awaiting_evidence" && previous.reason === takeover.reason; const event = historyEntry({ now, restartKind: input.restartKind, @@ -956,7 +993,9 @@ export async function claimNativeRestartRecoveries(input: { .set({ recoveryState: "awaiting_evidence", recoveryRequestId: input.recoveryRequestId ?? null, - recoveryHistory: appendBoundedRecoveryHistory(event), + // Repeated scans of one ownership wait record it once, so the + // bounded history keeps the drain record the next claim reads. + ...(waitAlreadyRecorded ? {} : { recoveryHistory: appendBoundedRecoveryHistory(event) }), updatedAt: now, }) .where( @@ -1112,6 +1151,8 @@ export async function claimNativeRestartRecoveries(input: { companyId: row.run.companyId, issueId: row.coordinator.issueId, runId: row.run.id, }, { runnerInstanceId: row.run.runnerInstanceId!, normalizedSessionId: row.run.nativeSessionId! }); + // Read before this claim appends its own history entry. + const cleanDrainSuspension = suspendedByCleanDrain(row.coordinator); const classification = classifyNativeRunnerRecoveryEvidence({ remoteSandbox: remoteSandbox || Boolean(pluginTask && durableAuthority), unclaimedBootstrap, @@ -1134,6 +1175,7 @@ export async function claimNativeRestartRecoveries(input: { ?.runTerminalState === "failed" && nativeRestartInterruptedTurnId(checkpointRecord as unknown as PersistedNativeSession) === null, providerAttempt: row.coordinator.attempt, + resumeChargesAttempt: !cleanDrainSuspension, }); if (stopCleanup && classification.claimKind !== "reattach_remote_runner") { return { kind: "blocked", runId: row.run.id, reason: "native_stop_requires_existing_remote_session" } as const; @@ -1397,6 +1439,7 @@ export async function claimNativeRestartRecoveries(input: { providerAttempt, restartKind: input.restartKind, recoveryRequestId: input.recoveryRequestId ?? null, + chargesProviderAttempt: !cleanDrainSuspension, }; if (claimKind === "reattach_remote_runner") { return { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 35798b3fa77..3fdd6a830a7 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -9335,6 +9335,7 @@ async function executePaperclipNativeSessionWithinScope( const nextAttempt = nextNativeProviderAttempt( incidentAttempts, recovering?.kind, + !recovering || !("chargesProviderAttempt" in recovering) || recovering.chargesProviderAttempt !== false, ); if (nextAttempt > 3) throw new Error("native_session_retry_exhausted");