diff --git a/server/src/services/native-local-process-stop.ts b/server/src/services/native-local-process-stop.ts index d5df0a12fea..479d524c611 100644 --- a/server/src/services/native-local-process-stop.ts +++ b/server/src/services/native-local-process-stop.ts @@ -346,6 +346,19 @@ export async function withReleasedNativeRunProcess { + const [latest, launch] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }) + .from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.companyId, companyId), eq(heartbeatRunEvents.runId, runId), + isNull(heartbeatRunEvents.sourceEventId), + inArray(heartbeatRunEvents.eventType, [PROCESS_START_REQUESTED, PROCESS_IDENTITY_RECORDED, LOCAL_PROCESS_STOPPED]))) + .orderBy(desc(heartbeatRunEvents.seq)).limit(2); + if (latest?.eventType !== LOCAL_PROCESS_STOPPED || latest.payload?.basis !== "unidentified_runner_absent" || + launch?.eventType !== PROCESS_START_REQUESTED || launch.payload?.hostname !== os.hostname()) return false; + const { noLocalRunnerProcessForRun } = await import("./native-runtime/native-session-executor.js"); + return noLocalRunnerProcessForRun(runId); +} + export async function readNativeLocalProcessStop(db: Db, companyId: string, runId: string) { const [event] = await db.select({ eventType: heartbeatRunEvents.eventType, payload: heartbeatRunEvents.payload }) .from(heartbeatRunEvents) diff --git a/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts b/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts index 22c96d48e3f..dd94207b60d 100644 --- a/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts +++ b/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts @@ -1,4 +1,7 @@ import { randomUUID } from "node:crypto"; +import { spawn } from "node:child_process"; +import { once } from "node:events"; +import { hostname } from "node:os"; import { eq, sql } from "drizzle-orm"; import { afterAll, beforeAll, describe, expect, it, vi } from "vitest"; import { agents, companies, completionContracts, createDb, heartbeatRunEvents, heartbeatRuns, issues, nativeRunFinalizations } from "@paperclipai/db"; @@ -36,7 +39,8 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { const exactResume = { recoveryMode: "exact_checkpoint_resume", recoverable: true, providerSessionEstablished: true, checkpointExists: true, originalFailureCode: "native_session_recovery_failed" }; - async function resumeOnce(options: { failureDetail: Record; stopRecorded: boolean; viaReconciler?: boolean }) { + async function resumeOnce(options: { failureDetail: Record; stopRecorded: boolean; viaReconciler?: boolean; + failedRelaunch?: boolean; unidentifiedStopRecorded?: boolean; liveRunnerForRun?: boolean }) { const issueId = randomUUID(); const runId = randomUUID(); const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Resume the lost turn", @@ -72,6 +76,10 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { { eventType: "native.process_start_requested", payload: { providerAttempt: 1 } }, { eventType: "native.process_identity_recorded", payload: identity }, ...(options.stopRecorded ? [{ eventType: "native.local_process_stopped", payload: identity }] : []), + ...(options.failedRelaunch ? [{ eventType: "native.process_start_requested", + payload: { hostname: hostname(), providerAttempt: 1, requestedAt: new Date().toISOString() } }] : []), + ...(options.unidentifiedStopRecorded ? [{ eventType: "native.local_process_stopped", + payload: { processPid: null, processGroupId: null, processStartedAt: null, basis: "unidentified_runner_absent" } }] : []), ]; const [last] = await db.select({ seq: sql`coalesce(max(${heartbeatRunEvents.seq}), 0)::int` }) .from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); @@ -81,11 +89,15 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq, eventType: event.eventType, stream: "system", level: "info", payload: event.payload }); } + await db.update(heartbeatRuns).set({ nextEventSeq: seq + 1 }).where(eq(heartbeatRuns.id, runId)); const passed: Array = []; provider.execute.mockReset().mockImplementation(async (input: { resumeInterruptedTurn?: boolean }) => { passed.push(input.resumeInterruptedTurn); throw new Error("scheduled_resume_fixture_stop"); }); + const liveRunner = options.liveRunnerForRun + ? spawn(process.execPath, ["-e", "setTimeout(() => {}, 60_000)", "--", "--run-id", runId], { stdio: "ignore" }) : null; + if (liveRunner) await once(liveRunner, "spawn"); let leaseOwner: string | undefined; if (options.viaReconciler) { const claims = await claimNativeSessionResumptions({ db, runnerInstanceId: native.runnerInstanceId, runIds: [runId] }); @@ -95,7 +107,7 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, leaseOwner, backend: { descriptor: async () => { throw new Error("Unexpected provider call"); }, openSession: async () => { throw new Error("Unexpected provider call"); }, - } }).catch(() => undefined); + } }).catch(() => undefined).finally(() => { liveRunner?.kill("SIGKILL"); }); const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); return { passed, coordinator: coordinator! }; } @@ -106,6 +118,23 @@ describe("a scheduled same-run resume reaching the provider runtime", () => { expect(coordinator.attempt).toBe(2); }); + it("continues the lost turn after a relaunch that failed before its runner existed", async () => { + const { passed, coordinator } = await resumeOnce({ failureDetail: exactResume, stopRecorded: true, failedRelaunch: true }); + expect(passed).toEqual([true]); + expect(coordinator.attempt).toBe(2); + }); + + it.each([false, true])("reuses the saved failed-relaunch stop on a later scheduled attempt (runner alive: %s)", async (liveRunnerForRun) => { + const { passed } = await resumeOnce({ failureDetail: exactResume, stopRecorded: true, failedRelaunch: true, + unidentifiedStopRecorded: true, liveRunnerForRun }); + expect(passed).toEqual([!liveRunnerForRun]); + }); + + it("keeps the lost turn failing after a failed relaunch while a runner for the run is still alive", async () => { + const { passed } = await resumeOnce({ failureDetail: exactResume, stopRecorded: true, failedRelaunch: true, liveRunnerForRun: true }); + expect(passed).toEqual([false]); + }); + it("keeps the lost turn failing when the reconciler claimed a resume without a stop receipt", async () => { const { passed } = await resumeOnce({ failureDetail: exactResume, stopRecorded: false, viaReconciler: true }); expect(passed).toEqual([false]); diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 0d23246118f..418dca7eb98 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10827,7 +10827,7 @@ async function executePaperclipNativeSessionAttempt( ); if (nextAttempt > 3) throw new Error("native_session_retry_exhausted"); - scheduledResumeContinuation = (await scheduledResumeContinuationAllowed(tx as unknown as Db, { + const scheduledResumeDecision = await scheduledResumeContinuationAllowed(tx as unknown as Db, { companyId: input.execution.binding.companyId, runId: input.execution.binding.runId, restartRecovery: Boolean(recovering), @@ -10838,7 +10838,12 @@ async function executePaperclipNativeSessionAttempt( ? (coordinator.recoveryHistory.at(-1) as Record | undefined) ?? null : null, controllerBootId: controller.bootId, admittedAttempt: nextAttempt, - })).allowed; + }); + scheduledResumeContinuation = scheduledResumeDecision.allowed; + if (!scheduledResumeDecision.allowed && coordinator.phase === "retryable_failure" + && record(coordinator.failureDetail).recoveryMode === "exact_checkpoint_resume") + logger.warn({ runId: input.execution.binding.runId, attempt: nextAttempt, reason: scheduledResumeDecision.reason }, + "scheduled same-run resume will not continue the interrupted provider turn"); const nextControllerGeneration = recovering ? recovering.controllerGeneration : coordinator.controllerBootId === controller.bootId diff --git a/server/src/services/native-runtime/scheduled-resume-continuation.ts b/server/src/services/native-runtime/scheduled-resume-continuation.ts index 215e540754a..94d40f37293 100644 --- a/server/src/services/native-runtime/scheduled-resume-continuation.ts +++ b/server/src/services/native-runtime/scheduled-resume-continuation.ts @@ -1,5 +1,7 @@ import type { Db } from "@paperclipai/db"; -import { readNativeLocalProcessStop } from "../native-local-process-stop.js"; +import { and, eq } from "drizzle-orm"; +import { heartbeatRuns } from "@paperclipai/db"; +import { readNativeLocalProcessStop, readUnidentifiedLocalRunnerStop, recordUnidentifiedLocalRunnerStop } from "../native-local-process-stop.js"; import { NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT } from "./native-restart-recovery.js"; export const SAME_RUN_RESUME_DISPOSITION = "same_run_resume"; @@ -44,7 +46,7 @@ export interface ScheduledResumeContinuationEvidence { resumeClaim?: Record | null; controllerBootId?: string; admittedAttempt: number; - localStop: { processPid: number } | null; + localStop: { processPid: number | null } | null; } export type ScheduledResumeContinuationDecision = @@ -79,5 +81,14 @@ export async function scheduledResumeContinuationAllowed( const { companyId, runId, ...evidence } = input; const localStop = evidence.restartRecovery || evidence.targetKind !== "local" ? null : await readNativeLocalProcessStop(db, companyId, runId); - return decideScheduledResumeContinuation({ ...evidence, localStop }); + const decision = decideScheduledResumeContinuation({ ...evidence, localStop }); + if (decision.allowed || decision.reason !== "local_stop_unverified") return decision; + if (await readUnidentifiedLocalRunnerStop(db, companyId, runId)) + return decideScheduledResumeContinuation({ ...evidence, localStop: { processPid: null } }); + const [run] = await db.select({ id: heartbeatRuns.id, companyId: heartbeatRuns.companyId, agentId: heartbeatRuns.agentId, + runtimeMode: heartbeatRuns.runtimeMode, processPid: heartbeatRuns.processPid, processGroupId: heartbeatRuns.processGroupId, + runnerProfileJson: heartbeatRuns.runnerProfileJson }) + .from(heartbeatRuns).where(and(eq(heartbeatRuns.id, runId), eq(heartbeatRuns.companyId, companyId))).limit(1); + if (!run || !await recordUnidentifiedLocalRunnerStop(db, run)) return decision; + return decideScheduledResumeContinuation({ ...evidence, localStop: { processPid: null } }); } diff --git a/tests/restart-chaos/README.md b/tests/restart-chaos/README.md index 8402bf5ccb6..2c4505b1687 100644 --- a/tests/restart-chaos/README.md +++ b/tests/restart-chaos/README.md @@ -1139,6 +1139,14 @@ only once is never reached with `nth` 2 or more, and the task fails `recovery_st 2026-11-11 an A1 task with K2 on Daytona never reached `directory-lock`: pick a workload whose recovery collects remote agent files. +`runnerd-start:1` with K1 (seed 5303) found the recovery-chain gap on 2026-11-11: the first recovery +attempt (`resume_dead_runner`) failed its spawn, which persisted a scheduled same-run resume. The second +attempt then refused to continue the interrupted turn (`local_stop_unverified`: the newest launch request +had no process identity and no stop receipt), so the restored session reported +`provider_turn_lost_on_restore` and the run ended `native_provider_terminal_failed`. A scheduled resume now +proves on this host that the failed launch left no runner and no runner state, records that stop, and +continues the turn. The cell passes only when the task finishes with no failed attempt. + Every task the armed fault crossed gets two more checks: `recovery_step_failed_once` (the receipt in `recovery-step-receipts.jsonl`; a step never reached tested nothing) and `no_failed_attempt`. The fault's receipt in `faults.json` names the armed step (`recoveryStep`).