diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts index c40cb4dd51e..9129ea6374f 100644 --- a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -9,7 +9,7 @@ import { tmpdir } from "node:os"; import { join } from "node:path"; import { eq } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; -import { agentTaskSessions, agents, companies, createDb, environmentLeases, environments, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db"; +import { agentTaskSessions, agents, companies, createDb, environmentLeases, environments, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations, nativeRunResults, nativeRunnerCheckpoints } from "@paperclipai/db"; import { DurablePrpControlPlane, parseNativeExecutionInput, type PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/helpers/recovery-postgres.js"; import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; @@ -192,6 +192,61 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d executionTargetKind: "local" })).toMatchObject({ recoveryMode: "exact_checkpoint_resume", providerSessionEstablished: true }); }); + it("retries a remote follow-up whose own server stopped serving before its runner, carrying only the predecessor session", async () => { + // Daytona wave s9203: a graceful drain closed the runtime API listener while + // a follow-up was still preparing. Its only checkpoint was the settled + // predecessor's provider session, rebound at admission. + const fixture = await seed(), runId = randomUUID(); + const checkpoint: PersistedNativeSession = { ...fixture.snapshot, backendKind: "runner", driverKind: "codex_app_server", + providerSessionId: "predecessor-thread", sessionId: "predecessor-thread", cursor: "predecessor-cursor", + activeTurnId: null, terminal: null, terminalTurns: [], pendingRuntimeRequests: [] }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { ...fixture.profile, sessionCheckpoint: checkpoint } }) + .where(eq(heartbeatRuns.id, fixture.scope.runId)); + const rebound = { ...checkpoint, cursor: null, identity: { ...checkpoint.identity, runId } }; + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: fixture.scope.issueId, + nativeSessionId: fixture.scope.normalizedSessionId, runnerInstanceId: randomUUID(), status: "running", runtimeMode: "native", + createdAt: new Date(Date.now() + 1_000), + runnerProfileJson: { sessionCheckpoint: rebound, nativePreparationFencingVersion: 1 } }); + const evidence = (sourceFailureCode: Parameters[0]["sourceFailureCode"], + executionTargetKind: "remote" | "local" = "remote") => + nativeProviderRecoveryEvidence({ db, runId, executionTargetKind, sourceFailureCode }); + expect(await evidence("local_runtime_api_listener_unavailable")).toMatchObject({ recoveryMode: "bootstrap_retry", + providerSessionEstablished: false, providerEventsExist: false, checkpointExists: true }); + // Only this server's own shutdown proves the runner was never reached. + expect((await evidence("provider_process_exited")).recoveryMode).toBe("ambiguous_state"); + // A provider session no settled predecessor saved could be this run's own. + await db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { ...rebound, providerSessionId: "unknown-thread" }, + nativePreparationFencingVersion: 1 } }).where(eq(heartbeatRuns.id, runId)); + expect((await evidence("local_runtime_api_listener_unavailable")).recoveryMode).toBe("ambiguous_state"); + // A turn of this run is provider work, whatever stopped it. + await db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: { ...rebound, cursor: "own-turn" }, + nativePreparationFencingVersion: 1 } }).where(eq(heartbeatRuns.id, runId)); + expect((await evidence("local_runtime_api_listener_unavailable")).recoveryMode).toBe("ambiguous_state"); + await db.update(heartbeatRuns).set({ runnerProfileJson: { sessionCheckpoint: rebound, nativePreparationFencingVersion: 1 } }) + .where(eq(heartbeatRuns.id, runId)); + // Any Runner controller checkpoint of this run, even one whose identity + // no longer matches the run, proves a Runner was reached. + const core = new DurablePrpControlPlane({ stateDirectory: join(root, runId, "reached-runner"), + identity: { runId, normalizedSessionId: fixture.scope.normalizedSessionId, runnerInstanceId: randomUUID(), + environmentLeaseId: fixture.native.environmentLeaseId, turnId: fixture.native.turnId, itemId: fixture.native.itemId }, + expectedRunnerVersion: "discovery-fixture", expectedRunnerDigest: `sha256:${"a".repeat(64)}` }); + const runnerOwner = { companyId, issueId: fixture.scope.issueId, runId, authorityRunId: runId, + bootId: randomUUID(), generation: 1, leaseOwner: randomUUID() }; + await db.insert(nativeRunFinalizations).values({ companyId, issueId: fixture.scope.issueId, runId, phase: "observed", + controllerBootId: runnerOwner.bootId, controllerGeneration: 1, leaseOwner: runnerOwner.leaseOwner, + leaseExpiresAt: new Date(Date.now() + 60_000) }); + core.setCheckpointPersistence(nativeRunnerCheckpointWriter(db, runnerOwner)); + try { + await core.flushCheckpoint(); + expect((await evidence("local_runtime_api_listener_unavailable")).recoveryMode).toBe("ambiguous_state"); + } finally { await core.stop(); } + await db.delete(nativeRunnerCheckpoints).where(eq(nativeRunnerCheckpoints.runId, runId)); + expect((await evidence("local_runtime_api_listener_unavailable")).recoveryMode).toBe("bootstrap_retry"); + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq: 1, stream: "system", level: "info", + eventType: "session.resumed" }); + expect((await evidence("local_runtime_api_listener_unavailable")).recoveryMode).toBe("ambiguous_state"); + }); + it("uses the database clock when an import finds an expired owner", async () => { const fixture = await seed(); const now = Date.now(); diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index bd81b559f24..bc7a277b294 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -861,7 +861,7 @@ async function hasNativeRunnerOriginatedEvidence( /** A settled earlier run of the same native session saved this exact provider * session. Admission rebinds that checkpoint to a follow-up run, so the * follow-up's copy of the provider session is carried, not its own work. */ -async function carriesSettledPredecessorSession( +export async function carriesSettledPredecessorSession( tx: Pick, run: { companyId: string; id: string; nativeSessionId: string | null; createdAt: Date }, providerSessionId: unknown, diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 3b37d6c3145..2029f7c7442 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -347,6 +347,7 @@ import { NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT, type NativeControllerIdentity, type NativeRestartRecoveryClaim, + carriesSettledPredecessorSession, } from "./native-restart-recovery.js"; type ActiveNativeSession = { @@ -8731,7 +8732,7 @@ export async function nativeProviderRecoveryEvidence(input: { const run = await input.db .select({ runnerProfileJson: heartbeatRuns.runnerProfileJson, companyId: heartbeatRuns.companyId, nativeIssueId: heartbeatRuns.nativeIssueId, nativeSessionId: heartbeatRuns.nativeSessionId, - runnerInstanceId: heartbeatRuns.runnerInstanceId }) + runnerInstanceId: heartbeatRuns.runnerInstanceId, id: heartbeatRuns.id, createdAt: heartbeatRuns.createdAt }) .from(heartbeatRuns) .where(eq(heartbeatRuns.id, input.runId)) .limit(1) @@ -8765,6 +8766,7 @@ export async function nativeProviderRecoveryEvidence(input: { ? await loadNativeUnclaimedBootstrapCheckpoint(input.db, input.bootstrap.binding, input.bootstrap.identity) : null; let controllerCheckpointMatches = true; + let controllerCheckpointAbsent = false; if (input.executionTargetKind === "remote" && checkpointRecord.backendKind === "runner" && record(run?.runnerProfileJson).nativePreparationFencingVersion === NATIVE_PREPARATION_FENCING_VERSION) { controllerCheckpointMatches = false; @@ -8772,6 +8774,7 @@ export async function nativeProviderRecoveryEvidence(input: { const header = await loadNativeRunnerCheckpointHeader(input.db, { companyId: run.companyId, issueId: run.nativeIssueId, runId: input.runId, }); + controllerCheckpointAbsent = !header; if (header && Buffer.byteLength(header) <= NATIVE_CHECKPOINT_HEADER_INSPECTION_MAX_BYTES) { const identity = inspectDurablePrpCheckpointHeader(header); const staged = inspectStagedRunAttachmentHeader(header); @@ -8811,6 +8814,30 @@ export async function nativeProviderRecoveryEvidence(input: { "provider_session_open_timeout", "local_runtime_api_listener_unavailable", ]).has(input.sourceFailureCode); + // A closed runtime API listener means this server stopped serving (a + // graceful drain) before the attempt could reach its remote runner. A + // follow-up may still carry its settled predecessor's provider session, + // rebound at admission with no cursor, turn, result or request of its own. + // With no runner checkpoint and no provider event for this run, that + // carried session is not this run's provider work: retry its startup on the + // same run instead of holding the task for a manual Retry. + const carriedAdmissionCheckpointOnly = checkpointExists && !providerEventsExist + && input.sourceFailureCode === "local_runtime_api_listener_unavailable" + && controllerCheckpointAbsent && !!run + && checkpointRecord.cursor == null && checkpointRecord.activeTurnId == null + && checkpointRecord.semanticResult == null && checkpointRecord.terminal == null + && (!Array.isArray(checkpointRecord.terminalTurns) || checkpointRecord.terminalTurns.length === 0) + && (checkpointRecord.pendingRuntimeRequests == null || (Array.isArray(checkpointRecord.pendingRuntimeRequests) + && checkpointRecord.pendingRuntimeRequests.length === 0)) + && await carriesSettledPredecessorSession(input.db, run, checkpointRecord.providerSessionId); + if (carriedAdmissionCheckpointOnly) { + return { + recoveryMode: "bootstrap_retry", + providerSessionEstablished: false, + providerEventsExist: false, + checkpointExists: true, + }; + } if (!checkpointExists && !providerEventsExist && (definitelyPreSession || unclaimedBootstrap && input.bootstrap?.transientIngressFailure)) { return { @@ -12463,7 +12490,9 @@ async function executePaperclipNativeSessionAttempt( ? "Inspect the persisted native session after its bounded resume budget was exhausted." : recoveryEvidence.recoveryMode === "bootstrap_retry" - ? `Retry provider bootstrap on this same run${nativeStartupRetryWait(providerCooldownAt)}; durable evidence proves no provider session or provider event was created.` + ? recoveryEvidence.checkpointExists + ? `Retry provider bootstrap on this same run${nativeStartupRetryWait(providerCooldownAt)}; durable evidence proves this run did no provider work of its own, and it resumes the provider session carried from its settled predecessor.` + : `Retry provider bootstrap on this same run${nativeStartupRetryWait(providerCooldownAt)}; durable evidence proves no provider session or provider event was created.` : "Resume this same run from its exact persisted native provider checkpoint after the retry delay.", }), nextAttemptAt,