diff --git a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts index 0ce76469c5f..b32671995b8 100644 --- a/server/src/__tests__/heartbeat-native-restart-suspension.test.ts +++ b/server/src/__tests__/heartbeat-native-restart-suspension.test.ts @@ -384,6 +384,55 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut expect(reached).toHaveBeenCalled(); }, 60_000); + // Daytona waves s9330 and s9331 (TM1): stalls of the same server outlived + // the execution lease. Renewal found the lease lapsed and detached the + // executor for "the next controller" with no shutdown handoff, and the + // executor returned with the run still running. The next controller is this + // same process; nothing may leave the run running with no recovery. + it("recovers a run whose executor detached after its own lease lapsed, with no restart", async () => { + const f = await seedRecoverableRun("SELF-DETACHED-LAPSED"); + let opened = 0; + const reachedAgain = vi.fn(); + const backend = vi.fn((): NativeSessionBackend => { + opened += 1; + const first = opened === 1; + const open = async () => { + if (first) { + // What the remote renewal's loss does: the lease lapsed during the + // stall, and the executor is detached for the next controller. + await f.db.update(nativeRunFinalizations).set({ leaseExpiresAt: new Date(Date.now() - 1_000) }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + void detachNativeSessionsForRestart([f.runId]); + throw new Error("runner stream lost while the controller was stopped"); + } + reachedAgain(); + throw new Error("stop at provider dispatch"); + }; + return { + async descriptor() { + return { kind: "mock", name: "self-detached", version: "1", capabilities: { + resume: true, typedEvents: true, steering: false, interruption: true, structuredResult: true }, + runtimeContextCapabilities: { instructions: "native", skills: "native", mcp: "native" } }; + }, + openSession: open, + recoverSession: open, + } as unknown as NativeSessionBackend; + }); + const heartbeat = heartbeatService(f.db, { nativeSessionBackendFactory: backend }); + try { + const { claims } = await heartbeat.recoverNativeRunsAfterRestart(new Date(), [f.runId]); + expect(claims).toEqual([expect.objectContaining({ runId: f.runId })]); + await heartbeat.drainActiveRunExecutions(); + // The same process takes the run back on its own, without a restart. + await waitForCondition("the same process to take the run back", () => reachedAgain.mock.calls.length > 0, 15_000); + } finally { await heartbeat.drainActiveRunExecutions(); } + expect(reachedAgain).toHaveBeenCalled(); + const transitions = await f.db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, f.runId), + eq(heartbeatRunEvents.eventType, "native.recovery.transition"))); + expect(transitions.map(event => (event.payload as Record).reason)) + .not.toContain("controller_still_alive"); + }, 60_000); + async function shutdownDispositions(db: Awaited>["db"], runId: string) { const events = await db.select().from(heartbeatRunEvents).where(and(eq(heartbeatRunEvents.runId, runId), @@ -832,7 +881,9 @@ describeEmbeddedPostgres("heartbeat native recovery interrupted by graceful shut .where(eq(nativeRunFinalizations.runId, f.runId)); const onClaim = vi.fn(); expect(await claimNativeRestartRecoveries({ db: f.db, restartKind: "hard", runIds: [f.runId], onClaim })) - .toEqual([{ kind: "awaiting_evidence", runId: f.runId, reason: "controller_still_alive" }]); + .toEqual([{ kind: "awaiting_evidence", runId: f.runId, reason: "controller_still_alive", + // This process's own live lease lapses by itself; recovery looks again then. + retryAt: expect.any(Date) }]); expect(onClaim).not.toHaveBeenCalled(); }, 60_000); diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index baaaace2ea1..72916c807b0 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -55,7 +55,7 @@ import { prepareHeartbeatTaskContext } from "./heartbeat/task-context.js"; import { selectHeartbeatEnvironment } from "./heartbeat/environment-selection.js"; import { reconcileMuseNativeTerminalRun } from "./native-runtime/muse-terminal-reconciliation.js"; import { prepareHeartbeatWorkspace } from "./heartbeat/workspace-preparation.js"; -import { nativeRecoveryClaimRelinquished, releaseLapsedNativeRestartClaim } from "./native-runtime/native-execution-lease.js"; +import { nativeRecoveryClaimRelinquished, ownNativeExecutionLeaseLapsed, releaseLapsedNativeRestartClaim } from "./native-runtime/native-execution-lease.js"; import { executeHeartbeatRuntime, NativeSessionResumeScheduledError, NativeWorkspaceFinalizeScheduledError } from "./heartbeat/runtime-execution.js"; import { preserveNativeFailedWorkspace } from "./native-runtime/native-failed-workspace-recovery.js"; import { nativeRunHasNoWorkspaceAccess } from "./workspace-restore-recovery-state.js"; @@ -2041,6 +2041,11 @@ export function heartbeatService( Parameters[0] | null = null; let nativeSessionResumeScheduled = false; let restartClaimLapsed = false; + // The executor detached for "the next controller" because its own lease + // lapsed (for example after a stall), with no shutdown or hot-restart + // handoff. The next controller is this process: recover the run here once + // the executor has exited. A shutdown detach leaves it to the next server. + let detachedWithoutHandoff = false; const shutdownHandedOff = watchShutdownHandoff(run.id); let nativeOwnershipHeld = false; let nativeStartupCleanupScheduled = false; @@ -3707,6 +3712,8 @@ export function heartbeatService( } if (err instanceof NativeControllerDetachedForRestartError) { nativeSessionResumeScheduled = true; + detachedWithoutHandoff = !shutdownHandedOff() && await ownNativeExecutionLeaseLapsed(db, run) + .catch(() => false); return; } if (err instanceof NativeRunnerOwnershipUnverifiedError) { @@ -4168,7 +4175,7 @@ export function heartbeatService( } finally { controllerLease.stop(); activeRunExecutions.delete(run.id); - if (restartClaimLapsed) scheduleNativeSessionResumeDispatch(run.id, new Date(), "restart"); + if (restartClaimLapsed || detachedWithoutHandoff) scheduleNativeSessionResumeDispatch(run.id, new Date(), "restart"); // A failed owned Stop remains visible until this exact executor settles, // including a graceful exit result arriving after the cancellation error. // It is never retained beyond the active execution's cleanup. diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index f6ea8aa8dae..88914e28336 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,5 +1,5 @@ import { settleOwnableKind } from "../recovery/ownership-invariant-report.js"; -import { markRunHandedOffForShutdown } from "./shutdown-handoff.js"; +import { isRunHandedOffForShutdown, markRunHandedOffForShutdown } from "./shutdown-handoff.js"; import { ORPHANED_RUN_LEASE_FAILURE_REASON } from "../environment-run-lease-settlement.js"; import { ENVIRONMENT_LEASE_CONTROLLER_KEY } from "../environment-lease-controller.js"; import { readNativeStartupCleanup, readNativeStartupContext } from "../native-runtime/native-startup-cleanup.js"; @@ -1009,6 +1009,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD }, db, liveRunExecutions: activeRunExecutions, + handedOffForShutdown: isRunHandedOffForShutdown, runIds, restartKind, recoveryRequestId: intent?.recoveryRequestId ?? null, diff --git a/server/src/services/heartbeat/shutdown-handoff.ts b/server/src/services/heartbeat/shutdown-handoff.ts index 57df43b2cef..0d84fcdd18c 100644 --- a/server/src/services/heartbeat/shutdown-handoff.ts +++ b/server/src/services/heartbeat/shutdown-handoff.ts @@ -11,3 +11,8 @@ export function watchShutdownHandoff(runId: string): () => boolean { return current !== undefined && current !== atStart; }; } + +/** This process handed the run to the next server (graceful drain or hot restart). */ +export function isRunHandedOffForShutdown(runId: string): boolean { + return runsHandedOffForShutdown.has(runId); +} diff --git a/server/src/services/native-runtime/native-execution-lease.ts b/server/src/services/native-runtime/native-execution-lease.ts index b63b7e1398f..626fe99ed33 100644 --- a/server/src/services/native-runtime/native-execution-lease.ts +++ b/server/src/services/native-runtime/native-execution-lease.ts @@ -220,6 +220,23 @@ export async function nativeRecoveryClaimRelinquished(input: { && row.lastDisposition === "restart_suspended"; } +/** + * The run's controller is this process and its execution lease has lapsed + * (or was released): the executor that held it detached because of that loss, + * not because a shutdown or hot restart handed the run on. + */ +export async function ownNativeExecutionLeaseLapsed(db: Db, input: { companyId: string; id: string }): Promise { + const controller = await currentNativeControllerIdentity(); + const [row] = await db.select({ runId: nativeRunFinalizations.runId }).from(nativeRunFinalizations) + .innerJoin(heartbeatRuns, eq(heartbeatRuns.id, nativeRunFinalizations.runId)) + .where(and(eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.runId, input.id), + eq(heartbeatRuns.status, "running"), isNull(nativeRunFinalizations.resultId), + eq(nativeRunFinalizations.controllerBootId, controller.bootId), eq(nativeRunFinalizations.controllerPid, controller.pid), + or(isNull(nativeRunFinalizations.leaseExpiresAt), sql`${nativeRunFinalizations.leaseExpiresAt} <= clock_timestamp()`))) + .limit(1); + return Boolean(row); +} + export async function releaseLapsedNativeRestartClaim(input: { db: Db; companyId: string; runId: string; leaseOwner: string; controllerGeneration: number; providerAttempt: number; }): Promise { diff --git a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts index e00347330c5..ecc80c19754 100644 --- a/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts +++ b/server/src/services/native-runtime/native-failed-workspace-recovery.test.ts @@ -87,7 +87,9 @@ const support = externalUrl ? { supported: true } : await getEmbeddedPostgresTes expect(await claimNativeRestartRecoveries({ db, restartKind: "hard", runIds: [f.runId] })).toEqual([ { kind: kind === "owned" ? "awaiting_evidence" : "blocked", runId: f.runId, - reason: kind === "owned" ? "controller_still_alive" : "provider_checkpoint_permanently_failed" }, + reason: kind === "owned" ? "controller_still_alive" : "provider_checkpoint_permanently_failed", + // This process's own live lease lapses by itself; recovery looks again then. + ...(kind === "owned" ? { retryAt: expect.any(Date) } : {}) }, ]); const current = await readRun(f.runId); expect(current.status).toBe(kind === "owned" ? "running" : "failed"); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index ec0bce99dbd..bd81b559f24 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -164,9 +164,13 @@ export function suspendedByCleanDrain(coordinator: { }): 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; + return coordinator.attempt > 0 && handedToNextControllerSinceClaim(coordinator.recoveryHistory); +} + +/** The latest record since the last claim hands the run to the next server. */ +function handedToNextControllerSinceClaim(history: ReadonlyArray<{ disposition?: unknown }>): boolean { + for (let index = history.length - 1; index >= 0; index -= 1) { + const disposition = history[index]?.disposition; if (disposition === "restart_suspended") return true; if (RECOVERY_CLAIM_DISPOSITIONS.has(disposition)) return false; } @@ -1029,6 +1033,8 @@ export async function claimNativeRestartRecoveries(input: { recoveryRequestId?: string | null; runIds?: string[]; liveRunExecutions?: Pick, "has">; + /** Runs this process handed to the next server (graceful drain or hot restart). */ + handedOffForShutdown?: (runId: string) => boolean; now?: Date; limit?: number; /** Called after each committed claim, before scanning the next candidate. */ @@ -1306,8 +1312,27 @@ export async function claimNativeRestartRecoveries(input: { && (controllerHostHash === "unknown" || row.coordinator.controllerHostHash !== controllerHostHash); const nonExecutionOwner = row.coordinator.leaseOwner?.startsWith("native-finalizer:") || row.coordinator.leaseOwner?.startsWith("native-cleanup:"); + // The recorded controller is this exact process (same boot and PID), so + // a PID probe can only ever answer "alive" and the wait would never end. + // This process knows whether it still drives the run: its executor is + // registered for the whole execution. While it is, leave the run to it. + // Once it has exited (for example after a stall outlived its lease and + // renewal detached it for "the next controller"), the prior owner is + // provably gone, and the evidence checks below decide as for any + // stopped controller. An unexpired self lease still waits: a claim that + // has not reached its executor yet holds it, and it lapses on its own. + // A run this process handed to the next server (graceful drain or hot + // restart) stays with that server: this process is about to exit. + const ownControllerProcess = !nonExecutionOwner && isCurrentNativeControllerProcess(row.coordinator); + if (ownControllerProcess && activeNativeExecutionRunIds.has(row.run.id)) { + return { kind: "awaiting_evidence", runId: row.run.id, reason: "current_controller_preparing" } as const; + } + const ownLeaseLapsed = ownControllerProcess && row.coordinator.leaseExpiresAt !== null + && row.coordinator.leaseExpiresAt <= now && !input.handedOffForShutdown?.(row.run.id) + && !handedToNextControllerSinceClaim(row.coordinator.recoveryHistory); const evaluatedTakeover = nonExecutionOwner ? { allowed: false, reason: "native_finalization_or_cleanup_owner" } + : ownLeaseLapsed ? { allowed: true, reason: "current_controller_released" } : await evaluateNativeControllerTakeover({ owner: row.coordinator, now, @@ -1379,6 +1404,10 @@ export async function claimNativeRestartRecoveries(input: { runId: row.run.id, reason: takeover.reason, ...(durableAuthority && row.coordinator.leaseExpiresAt ? { retryAt: row.coordinator.leaseExpiresAt } + // This process's own unexpired lease with no driver here lapses by + // itself; look again then instead of waiting for another signal. + : ownControllerProcess && row.coordinator.leaseExpiresAt && row.coordinator.leaseExpiresAt > now + && !input.handedOffForShutdown?.(row.run.id) ? { retryAt: row.coordinator.leaseExpiresAt } : ["controller_lease_held", "controller_lease_grace"].includes(takeover.reason) ? { retryAt: new Date(now.getTime() + CONTROLLER_LEASE_RENEW_MS) } : {}), } as const; diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 0e401700d39..348fcb7c29c 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -38,7 +38,7 @@ import { nativeRunnerCheckpoints, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS, } from "@paperclipai/db"; -import { beforeAll, expect, it, vi } from "vitest"; +import { beforeAll, describe, expect, it, vi } from "vitest"; import { heartbeatService } from "../heartbeat.js"; import { @@ -79,6 +79,7 @@ import { restoreNativeRunnerCheckpoint, } from "./native-runner-checkpoint.js"; import { PaperclipRunnerToolAuthority } from "./paperclip-runner-tool-authority.js"; +import { setNativeExecutionActive } from "./native-active-runs.js"; import { RUNNER_BRIDGE_READ_OPERATIONS, RUNNER_BRIDGE_SCHEMAS } from "./runner-bridge-tools.js"; import { HttpError } from "../../errors.js"; import { legacyControllerBootId } from "../legacy-controller-lease.js"; @@ -3297,6 +3298,86 @@ process.on("SIGTERM", () => child.kill("SIGTERM")); restartKind: "hard", runIds: [fixture.runId] })).toEqual([expect.objectContaining({ runId: fixture.runId, ...expected })]); }); + // Daytona wave s9330 t02 (TM1): SIGSTOP held this same server past its + // execution lease while the executor was starting the native process. On + // SIGCONT the lease renewal found the lease lapsed, the executor detached for + // "the next controller" and exited, and every recovery scan then waited on + // the recorded controller: this same live PID and boot. Nothing else can hold + // a lease that names this process, and nothing in it drives the run any more. + describe("same-boot controller whose own lease lapsed", () => { + it("restarts the bootstrap instead of waiting on itself when no turn reached the runner", async () => { + const fixture = await seedUnlaunchedRemoteClaim("SELF-LAPSED-UNCOMMANDED", + { liveController: true, leaseExpired: true, event: { eventType: "native.process_start_requested" } }); + const [claim] = await claimNativeRestartRecoveries({ db: fixture.db, controller: await currentNativeControllerIdentity(), + restartKind: "hard", runIds: [fixture.runId], liveRunExecutions: new Set() }); + expect(claim).toMatchObject({ kind: "bootstrap_incomplete", runId: fixture.runId, controllerGeneration: 2 }); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator.recoveryHistory.at(-1)).toMatchObject({ reason: "remote_runner_never_received_commands" }); + }); + + it("fails with a Retry instead of waiting on itself when the runner may have started", async () => { + const fixture = await seedUnlaunchedRemoteClaim("SELF-LAPSED-IDENTITY", + { liveController: true, leaseExpired: true, event: { eventType: "native.process_identity_recorded" } }); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: await currentNativeControllerIdentity(), + restartKind: "hard", runIds: [fixture.runId], liveRunExecutions: new Set() })) + .toEqual([expect.objectContaining({ kind: "blocked", runId: fixture.runId, reason: "ambiguous_remote_recovery_evidence" })]); + const [coordinator] = await fixture.db.select().from(nativeRunFinalizations) + .where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(coordinator).toMatchObject({ phase: "terminal_failure", leaseOwner: null, failureCode: "native_restart_recovery_blocked" }); + const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(run.status).toBe("failed"); + const [issue] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId)); + expect(await taskInterruptionPresentation(fixture.db, issue)).toMatchObject({ + state: "failed", canRetry: true, observedRunId: fixture.runId }); + const personId = randomUUID(); + await fixture.db.insert(authUsers).values({ id: personId, name: "Retry person", email: `${personId}@fixture.invalid`, + emailVerified: true, createdAt: new Date(), updatedAt: new Date() }); + await fixture.db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: personId, + status: "active", membershipRole: "member" }); + expect(await requestTaskRetry({ db: fixture.db, companyId, issueId: fixture.issueId, actorId: personId, + actorSource: "session", idempotencyKey: `self-lapsed-${fixture.runId}`, observedRunId: fixture.runId })) + .toMatchObject({ accepted: true }); + }); + + it("leaves the run to this process's executor while it still drives the run", async () => { + const fixture = await seedUnlaunchedRemoteClaim("SELF-LAPSED-EXECUTING", + { liveController: true, leaseExpired: true, event: { eventType: "native.process_start_requested" } }); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: await currentNativeControllerIdentity(), + restartKind: "hard", runIds: [fixture.runId], liveRunExecutions: new Set([fixture.runId]) })) + .toEqual([expect.objectContaining({ kind: "awaiting_evidence", reason: "current_controller_preparing" })]); + setNativeExecutionActive(fixture.runId, true); + try { + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: await currentNativeControllerIdentity(), + restartKind: "hard", runIds: [fixture.runId], liveRunExecutions: new Set() })) + .toEqual([expect.objectContaining({ kind: "awaiting_evidence", reason: "current_controller_preparing" })]); + } finally { + setNativeExecutionActive(fixture.runId, false); + } + }); + + it("leaves a run this process handed to the next server for that server", async () => { + const fixture = await seedUnlaunchedRemoteClaim("SELF-LAPSED-HANDED-OFF", + { liveController: true, leaseExpired: true, event: { eventType: "native.process_start_requested" } }); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: await currentNativeControllerIdentity(), + restartKind: "hard", runIds: [fixture.runId], liveRunExecutions: new Set(), + handedOffForShutdown: runId => runId === fixture.runId })) + .toEqual([expect.objectContaining({ kind: "awaiting_evidence", reason: "controller_still_alive" })]); + }); + + it("keeps waiting for a different live controller in this process space", async () => { + const fixture = await seedUnlaunchedRemoteClaim("OTHER-LIVE-LAPSED", + { leaseExpired: true, event: { eventType: "native.process_start_requested" } }); + // Another live process (the test runner's parent) recorded as controller. + const parentStartedAt = new Date((await readProcessStartedAt(process.ppid))!); + await fixture.db.update(nativeRunFinalizations).set({ controllerPid: process.ppid, + controllerProcessStartedAt: parentStartedAt }).where(eq(nativeRunFinalizations.runId, fixture.runId)); + expect(await claimNativeRestartRecoveries({ db: fixture.db, controller: await currentNativeControllerIdentity(), + restartKind: "hard", runIds: [fixture.runId], liveRunExecutions: new Set() })) + .toEqual([expect.objectContaining({ kind: "awaiting_evidence", reason: "controller_still_alive" })]); + }); + }); + it.each([ { attempt: 2, driverKind: "codex_app_server" }, { attempt: 3, driverKind: "codex_app_server" },