From 37c50f70c29a82a866e5e58e71fb8ae0957dc72f Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Tue, 10 Nov 2026 15:16:51 -0600 Subject: [PATCH] fix(native): continue a turn lost on restore in a proven same-run resume (#20645) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - A native run keeps its provider conversation when its local runner dies. When the restored runner finds that the active turn was lost, it reports `provider_turn_lost_on_restore` and does not invent a result. > - After a control-plane restart, the admitted restart attempt may send one continuation in that same conversation (#15297). The agent then reconciles unfinished work. The task is not resent. > - A runner can also be lost without a server restart, for example when `runner.drain` times out. The run then resumes itself after the retry delay from its exact checkpoint. That path never allowed the continuation, because the executor passed `resumeInterruptedTurn` only for a `resume_dead_runner` restart claim. > - So the lost turn failed with `native_provider_terminal_failed`, automatic replacement was refused (`uncertain_provider_action`), and the user had to click Retry. > - This pull request allows the same single continuation for that resume, but only when durable records prove the same three facts that a restart claim proves. > - The benefit is that a runner lost mid-turn continues in the same conversation with no Retry. No case that is refused today becomes allowed without that proof. ## Linked Issues or Issue Description Refs #15297, #20625 **What happened?** Restart-chaos wave L7 (seed 9107, local), task codex B5 TM2, run on master after #20625. A TM3 restart resumed the run (attempt 1). During the TM2 freeze the TS runtime lost its runner: `native_session_recovery_failed: runner_local_connect_failed: PRP command runner.drain timed out`. The executor saved `retryable_failure` with `recoveryMode: "exact_checkpoint_resume"`. After the delay, recovery verified the runner process stopped (`native.local_process_stopped`). The same run then resumed as attempt 2 (`native.process_start_requested`, `providerAttempt: 2`). The restored runner reported `turn.failed` with `provider_turn_lost_on_restore` (`recoverable: true`). No continuation was sent. The run failed with `native_provider_terminal_failed`, recovery stopped with `uncertain_provider_action` / `preserve_without_replay_v1`, and the task needed Retry. **Expected behavior** The run sends the one same-conversation continuation, exactly as it does after a restart claim, and finishes without Retry. **Steps to reproduce** `node cli/node_modules/tsx/dist/cli.mjs tests/restart-chaos/main.ts --env local --adapters codex,claude --workloads A1,B5,C1 --faults TM1,TM2,TM3 --time-age-hours 1 --concurrency 3 --tasks 6 --seed 9107` (timing-dependent: the drain timeout must land during an active turn). ## What Changed - New `server/src/services/native-runtime/scheduled-resume-continuation.ts`. It decides whether a same-run scheduled resume may continue a lost turn. Each of the three proofs comes from a durable record: 1. **The previous runner is stopped.** The latest process record of the run in `heartbeat_run_events` (`native.process_start_requested`, `native.process_identity_recorded`, `native.local_process_stopped`) must be `native.local_process_stopped`. Its `processPid` must equal its `processGroupId`, and neither the pid nor the group may be alive now. This is the existing `readNativeLocalProcessStop` check, read in the claim transaction before this attempt records its own `native.process_start_requested`. 2. **The attempt was admitted under the bounded budget.** The executor's claim moves `native_run_finalizations` from phase `retryable_failure` to `observed`. It charges a provider attempt, and the claim refuses any attempt above 3 (`native_session_retry_exhausted`). The decision also requires `1 <= attempt <= NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT`. 3. **It is the same provider conversation.** The saved `failure_detail` must show `recoveryMode: "exact_checkpoint_resume"`, `recoverable: true`, `providerSessionEstablished: true` and `checkpointExists: true`, with no `successorRunId` or `replacementDenied`. At run time the existing runtime guards still apply: the backend must recover the checkpointed session (`recovered`, not a continuity-break replacement), and the lost turn must equal the checkpoint's `activeTurnId` and the snapshot's interrupted turn. The provider kinds stay codex, acpx and opencode, with no compaction, goal control, held result or pending runtime request. - `native-session-executor.ts`: computes that decision in the coordinator claim. It passes `resumeInterruptedTurn` when either the restart claim is `resume_dead_runner` (unchanged) or the decision allows it. - `doc/execution-semantics.md`: describes the rule and the refused cases. **Cases that stay refused** (each has a test in `scheduled-resume-continuation.test.ts`): - a restart-recovery claim (it keeps its own existing rule); - a remote execution target; - a coordinator that is not `retryable_failure`, including a fresh run; - `recoveryMode` `bootstrap_retry` or `ambiguous_state`, or no saved failure detail; - a failure that is not recoverable; - no established provider session; - no checkpoint; - a recorded replacement (`successorRunId` or `replacementDenied`); - an attempt outside 1..3; - no stop receipt, a stop receipt superseded by a newer launch, a stopped process that is still alive, or a run that never had a process. The runtime-side refusals are unchanged and already covered by `native-session-runtime.test.ts`: real provider errors, historical error-null failures, and no restart permission. ## Verification - `cd server && npx vitest run src/services/native-runtime/scheduled-resume-continuation.test.ts`: 22 tests, covering the decision table and the stop receipt in embedded Postgres. - `cd server && npx tsc --noEmit` passes. - Live: not proven. An L7 wave on this branch (seed 9107) did not reproduce the drain-timeout path, so this pull request has unit and database tests only. In that wave the same cell (codex B5 TM2) needed Retry for a different reason: TM2 aged its restart-recovery claim lease before dispatch, and budget admission refused with `Native recovery no longer owns provider dispatch`. This pull request does not change that path. Four other tasks passed. Claude C1 TM2 rested blocked without a Retry button, which is a separate cleanup-quarantine issue. ## Risks - This allows one more path to send the same-conversation continuation. It is limited to local runs. It needs the same proofs as a restart claim, and it uses the attempt the resume already charged. It does not change the generation cap or `restartResumeKeepsAttempt`. - A rolling deploy reads `failure_detail` and stop events written by earlier releases. Those fields and events already have this shape, so an older saved failure is judged by the same rule. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, via Claude Code with tool use (shell, file edits, test runs). Extended reasoning. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude --- doc/execution-semantics.md | 12 ++ .../src/native-session-runtime.test.ts | 3 +- .../src/native-session-runtime.ts | 4 +- ...ed-resume-continuation.integration.test.ts | 105 +++++++++++++ .../native-runtime/native-session-executor.ts | 13 +- .../scheduled-resume-continuation.test.ts | 140 ++++++++++++++++++ .../scheduled-resume-continuation.ts | 57 +++++++ 7 files changed, 330 insertions(+), 4 deletions(-) create mode 100644 server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts create mode 100644 server/src/services/native-runtime/scheduled-resume-continuation.test.ts create mode 100644 server/src/services/native-runtime/scheduled-resume-continuation.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index d0dffae1581..35a8bda18e8 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -3298,6 +3298,18 @@ results, intentional stops, and historical unmarked failures keep their existing terminal behavior. This continuation uses the already charged restart attempt and does not reset the provider-attempt budget. +The same continuation applies when the runner is lost without a server restart and +the run resumes itself after the retry delay (a local same-run exact-checkpoint +resume). Three durable records must hold. First, the latest process record of the +run is a verified stop of the previous runner (`native.local_process_stopped`) and +that process is gone. Second, the resume claimed a provider attempt within the +three-attempt budget from a `retryable_failure` coordinator. Third, that failure +saved `recoveryMode: "exact_checkpoint_resume"`, `recoverable`, +`providerSessionEstablished` and `checkpointExists`, with no replacement recorded, +and the runner restores that same checkpointed conversation. A remote run, a +bootstrap retry, an ambiguous state, a missing or superseded stop record, an +exhausted budget, or a replacement session keeps the existing failure. + A restart can also land after the agent proposed its result but before the provider turn ended. When the restored runner proves that turn ended with the result Paperclip accepted, it records a reconciliation and the run terminal, diff --git a/packages/paperclip-runner/src/native-session-runtime.test.ts b/packages/paperclip-runner/src/native-session-runtime.test.ts index 9b9d6c170ac..0dfda06c42f 100644 --- a/packages/paperclip-runner/src/native-session-runtime.test.ts +++ b/packages/paperclip-runner/src/native-session-runtime.test.ts @@ -220,7 +220,8 @@ function canonicalTestJson(value: unknown): string { const RESTART_LOST_PROCESS_GUIDANCE = "may no longer be running"; function expectRestartContinuationGuidance(text: string) { const events = String(JSON.parse(text).events); - expect(events).toContain("This is a resumed session after a Paperclip control-plane restart."); + expect(events).toContain("This is a resumed session. The process running the previous turn was lost"); + expect(events).not.toContain("control-plane restart"); expect(events).toContain("Shell commands and background processes started in earlier turns " + RESTART_LOST_PROCESS_GUIDANCE); expect(events).toContain("check the workspace for the outputs or artifacts"); expect(events).toContain("If the expected output is missing, re-run the command."); diff --git a/packages/paperclip-runner/src/native-session-runtime.ts b/packages/paperclip-runner/src/native-session-runtime.ts index 178c28ec56b..7b9d0971362 100644 --- a/packages/paperclip-runner/src/native-session-runtime.ts +++ b/packages/paperclip-runner/src/native-session-runtime.ts @@ -2731,10 +2731,10 @@ export async function executeNativeSession( message: { role: "user" as const, text: JSON.stringify({ schema: "paperclip.native-continuation.v1", events: [ - "This is a resumed session after a Paperclip control-plane restart. Paperclip restored this same conversation and workspace; the restart interrupted the previous turn.", + "This is a resumed session. The process running the previous turn was lost, for example in a Paperclip restart. Paperclip restored this same conversation and workspace; that loss interrupted the previous turn.", "Continue the current user request from the recorded progress and preserved workspace.", ...(steeredMessages.length > 0 ? [ - "During the interrupted turn the user sent the messages in steeredMessages. The restart happened before you read them. Handle each one in this turn, oldest first, together with the current request.", + "During the interrupted turn the user sent the messages in steeredMessages. The interruption happened before you read them. Handle each one in this turn, oldest first, together with the current request.", ] : []), "Shell commands and background processes started in earlier turns may no longer be running, and their tool sessions may be gone. Do not block only because such a process or its tool session cannot be found or inspected.", "Reconcile any unfinished tool or command before proceeding: check the workspace for the outputs or artifacts it should have produced. If the expected output is present, use it. If the expected output is missing, re-run the command.", 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 new file mode 100644 index 00000000000..1d02ca75eec --- /dev/null +++ b/server/src/services/native-runtime/native-scheduled-resume-continuation.integration.test.ts @@ -0,0 +1,105 @@ +import { randomUUID } from "node:crypto"; +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"; +import type { NativeExecutionInputV1 } from "@paperclipai/paperclip-runner"; +import { startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; + +const provider = vi.hoisted(() => ({ execute: vi.fn() })); +vi.mock("../../vendor/paperclip-runner/index.js", async (importOriginal) => ({ + ...(await importOriginal()), + executeNativeSession: provider.execute, +})); + +import { executePaperclipNativeSession } from "./native-session-executor.js"; +import { buildNativeCompletionContract } from "./completion-contracts.js"; +import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; + +const EXITED_PID = 4_194_304; + +describe("a scheduled same-run resume reaching the provider runtime", () => { + let temporary: Awaited>; + let db: ReturnType; + const companyId = randomUUID(); + const agentId = randomUUID(); + + beforeAll(async () => { + temporary = await startEmbeddedPostgresTestDatabase("native-scheduled-resume-continuation-"); + db = createDb(temporary.connectionString); + await db.insert(companies).values({ id: companyId, name: "Scheduled resume", issuePrefix: "NSR" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Native agent", status: "active", + adapterType: "paperclip_runner", adapterConfig: { provider: "codex", model: "test-model" } }); + }); + afterAll(async () => { await temporary?.cleanup(); }); + + 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 }) { + const issueId = randomUUID(); + const runId = randomUUID(); + const [issue] = await db.insert(issues).values({ id: issueId, companyId, title: "Resume the lost turn", + status: "in_progress", workMode: "standard", assigneeAgentId: agentId }).returning(); + const [run] = await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "running", + runtimeMode: "native", nativeIssueId: issueId, invocationSource: "assignment", triggerDetail: "system", + contextSnapshot: { issueId } }).returning(); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, issueId)); + const native = await prepareNativeHeartbeatRun({ db, run: run!, issue: issue!, environmentLeaseId: randomUUID() }); + const [contract] = await db.select().from(completionContracts).where(eq(completionContracts.issueId, issueId)); + const execution: NativeExecutionInputV1 = { + schema: "paperclip.native-execution-input.v1", + binding: { companyId, runId, issueId, agentId, executionWorkspaceId: runId }, + provider: { kind: "codex", model: "test-model" }, + task: { identifier: issueId, title: issue!.title, description: null, prompt: "Continue the work.", workMode: "standard" }, + workspace: { cwd: "/tmp", repoUrl: null, repoRef: null, branchName: null }, + session: { normalizedSessionId: native.normalizedSessionId, driverKind: "codex_app_server", protocolVersion: 1, + lifecyclePolicy: { mode: "per_turn", idleTimeoutMs: null } }, + completionContract: { id: contract!.id, sha256: contract!.canonicalSha256, schemaVersion: "paperclip.completion-contract.v1", + contract: buildNativeCompletionContract(issue!, { revision: contract!.revision }) }, + interactionResponses: [], credentialBindings: [], + }; + await db.update(heartbeatRuns).set({ runnerProfileJson: { nativeExecutionInput: execution } }).where(eq(heartbeatRuns.id, runId)); + await db.insert(nativeRunFinalizations).values({ runId, companyId, issueId, phase: "retryable_failure", attempt: 1, + failureCode: "native_session_interrupted", failureDetail: options.failureDetail }); + const identity = { processPid: EXITED_PID, processGroupId: EXITED_PID, processStartedAt: new Date().toISOString() }; + const events = [ + { eventType: "native.process_start_requested", payload: { providerAttempt: 1 } }, + { eventType: "native.process_identity_recorded", payload: identity }, + ...(options.stopRecorded ? [{ eventType: "native.local_process_stopped", payload: identity }] : []), + ]; + const [last] = await db.select({ seq: sql`coalesce(max(${heartbeatRunEvents.seq}), 0)::int` }) + .from(heartbeatRunEvents).where(eq(heartbeatRunEvents.runId, runId)); + let seq = last?.seq ?? 0; + for (const event of events) { + seq += 1; + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq, eventType: event.eventType, + stream: "system", level: "info", payload: event.payload }); + } + const passed: Array = []; + provider.execute.mockReset().mockImplementation(async (input: { resumeInterruptedTurn?: boolean }) => { + passed.push(input.resumeInterruptedTurn); + throw new Error("scheduled_resume_fixture_stop"); + }); + await executePaperclipNativeSession({ db, execution, runnerInstanceId: native.runnerInstanceId, backend: { + descriptor: async () => { throw new Error("Unexpected provider call"); }, + openSession: async () => { throw new Error("Unexpected provider call"); }, + } }).catch(() => undefined); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + return { passed, coordinator: coordinator! }; + } + + it("lets the runtime continue a lost turn after a verified stop and an admitted attempt", async () => { + const { passed, coordinator } = await resumeOnce({ failureDetail: exactResume, stopRecorded: true }); + expect(passed).toEqual([true]); + expect(coordinator.attempt).toBe(2); + }); + + it.each([ + ["no stop of the previous runner was recorded", { failureDetail: exactResume, stopRecorded: false }], + ["the saved failure is a bootstrap retry", { failureDetail: { ...exactResume, recoveryMode: "bootstrap_retry" }, stopRecorded: true }], + ["the saved failure has no checkpoint", { failureDetail: { ...exactResume, checkpointExists: false }, stopRecorded: true }], + ])("keeps the lost turn failing when %s", async (_name, options) => { + const { passed } = await resumeOnce(options); + 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 ecd8920ca7c..535f9636339 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -119,6 +119,7 @@ import { stoppedCodexTurnIsTextOnly } from "./stopped-codex-turn.js"; import { prepareVerifiedRemoteProviderPack, useVerifiedPreinstalledRemoteProviderPack, shouldUseBundledAcpxImageAssets } from "./remote-provider-pack.js"; import { uploadVerifiedProviderPack } from "./provider-pack-upload.js"; import { selectRemotePiCompanion } from "./remote-pi-companion.js"; +import { scheduledResumeContinuationAllowed } from "./scheduled-resume-continuation.js"; import { readNativeLocalProcessStop, recordedLocalExecutionStopped, PRIOR_OWNER_FENCE_REFUSED, PROCESS_IDENTITY_RECORDED, PROCESS_START_REQUESTED, withReleasedNativeRunProcess } from "../native-local-process-stop.js"; import { remoteLeaseCleanupScope } from "../remote-execution-termination.js"; import { connectorToolDefinitionsForSkills } from "../connector-runtime.js"; @@ -10063,6 +10064,7 @@ async function executePaperclipNativeSessionAttempt( let confirmedLeaseDeadline = 0; let claimedStartupContext: NativeExecutionContext | undefined; let freshClaimAttempt: number | undefined; + let scheduledResumeContinuation = false; try { attempt = await trace.measure( "native.coordinator.claim", @@ -10212,6 +10214,15 @@ async function executePaperclipNativeSessionAttempt( ); if (nextAttempt > 3) throw new Error("native_session_retry_exhausted"); + scheduledResumeContinuation = (await scheduledResumeContinuationAllowed(tx as unknown as Db, { + companyId: input.execution.binding.companyId, + runId: input.execution.binding.runId, + restartRecovery: Boolean(recovering), + targetKind: input.runnerExecutionTarget?.kind === "remote" ? "remote" : "local", + coordinatorPhase: coordinator.phase, + failureDetail: coordinator.failureDetail, + admittedAttempt: nextAttempt, + })).allowed; const nextControllerGeneration = recovering ? recovering.controllerGeneration : coordinator.controllerBootId === controller.bootId @@ -11302,7 +11313,7 @@ async function executePaperclipNativeSessionAttempt( trace.activate(runnerSessionStartupScope); const runExecution = () => executeNativeSession({ - resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner", + resumeInterruptedTurn: input.restartRecovery?.kind === "resume_dead_runner" || scheduledResumeContinuation, getFreshSessionHandoff: input.getFreshSessionHandoff, // A failed seal commits nothing: the messages stay queued for the // turn boundary instead of failing this run before its turn. diff --git a/server/src/services/native-runtime/scheduled-resume-continuation.test.ts b/server/src/services/native-runtime/scheduled-resume-continuation.test.ts new file mode 100644 index 00000000000..b48033c0eaf --- /dev/null +++ b/server/src/services/native-runtime/scheduled-resume-continuation.test.ts @@ -0,0 +1,140 @@ +import { spawn } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { once } from "node:events"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { agents, companies, createDb, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase, +} from "../../__tests__/helpers/embedded-postgres.js"; +import { + decideScheduledResumeContinuation, + scheduledResumeContinuationAllowed, + type ScheduledResumeContinuationEvidence, +} from "./scheduled-resume-continuation.js"; + +const proven: ScheduledResumeContinuationEvidence = { + restartRecovery: false, + targetKind: "local", + coordinatorPhase: "retryable_failure", + failureDetail: { + recoveryMode: "exact_checkpoint_resume", + recoverable: true, + providerSessionEstablished: true, + checkpointExists: true, + originalFailureCode: "native_session_recovery_failed", + }, + admittedAttempt: 2, + localStop: { processPid: 81495 }, +}; + +describe("same-run scheduled resume continuation of a turn lost on restore", () => { + it("allows the continuation when the stop, the admitted attempt and the exact checkpoint are all recorded", () => { + expect(decideScheduledResumeContinuation(proven)).toEqual({ allowed: true }); + expect(decideScheduledResumeContinuation({ ...proven, admittedAttempt: 3 })).toEqual({ allowed: true }); + }); + + const detail = (patch: Record) => ({ ...proven, failureDetail: { ...proven.failureDetail, ...patch } }); + it.each<[string, ScheduledResumeContinuationEvidence, string]>([ + ["a restart recovery claim owns its own rule", { ...proven, restartRecovery: true }, "restart_recovery_claim"], + ["the run executes on a remote target", { ...proven, targetKind: "remote" }, "remote_target"], + ["the coordinator is not a settled same-run retry", { ...proven, coordinatorPhase: "observed" }, "not_a_scheduled_same_run_resume"], + ["a fresh run has no coordinator phase", { ...proven, coordinatorPhase: null }, "not_a_scheduled_same_run_resume"], + ["the retry is a bootstrap retry", detail({ recoveryMode: "bootstrap_retry" }), "not_an_exact_checkpoint_resume"], + ["the state is ambiguous", detail({ recoveryMode: "ambiguous_state" }), "not_an_exact_checkpoint_resume"], + ["no failure detail was saved", { ...proven, failureDetail: null }, "not_an_exact_checkpoint_resume"], + ["the failure is not recoverable", detail({ recoverable: false }), "failure_not_recoverable"], + ["no provider session was established", detail({ providerSessionEstablished: false }), "provider_session_not_established"], + ["no checkpoint exists", detail({ checkpointExists: false }), "checkpoint_missing"], + ["a replacement run was created", detail({ successorRunId: "run-2" }), "replacement_recorded"], + ["a replacement was denied", detail({ replacementDenied: "uncertain_provider_action" }), "replacement_recorded"], + ["the attempt exceeds the budget", { ...proven, admittedAttempt: 4 }, "attempt_not_admitted"], + ["no attempt was admitted", { ...proven, admittedAttempt: 0 }, "attempt_not_admitted"], + ["the previous runner has no stop receipt", { ...proven, localStop: null }, "local_stop_unverified"], + ])("refuses when %s", (_name, evidence, reason) => { + expect(decideScheduledResumeContinuation(evidence)).toEqual({ allowed: false, reason }); + }); +}); + +const support = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = support.supported ? describe : describe.skip; + +describeEmbeddedPostgres("the durable stop receipt behind a scheduled resume continuation", () => { + let db!: ReturnType; + let stopDatabase: (() => Promise) | undefined; + let companyId!: string; + let agentId!: string; + let deadPid!: number; + + beforeAll(async () => { + const started = await startEmbeddedPostgresTestDatabase("paperclip-scheduled-resume-continuation-"); + stopDatabase = started.cleanup; + db = createDb(started.connectionString); + companyId = randomUUID(); + agentId = randomUUID(); + await db.insert(companies).values({ id: companyId, name: "Resume", issuePrefix: "SRC", + requireBoardApprovalForNewAgents: false }); + await db.insert(agents).values({ id: agentId, companyId, name: "Runner", role: "engineer", status: "active", + adapterType: "paperclip_runner", adapterConfig: { provider: "codex" }, runtimeConfig: {}, permissions: {} }); + const child = spawn(process.execPath, ["-e", ""], { stdio: "ignore" }); + await once(child, "exit"); + deadPid = child.pid!; + }, 120_000); + + afterAll(async () => { + await db?.$client.end({ timeout: 1 }); + await stopDatabase?.(); + }); + + async function seedRun(events: Array<{ eventType: string; payload: Record }>) { + const runId = randomUUID(); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, status: "failed", runtimeMode: "native", + invocationSource: "assignment", triggerDetail: "system", contextSnapshot: {} }); + let seq = 0; + for (const event of events) { + seq += 1; + await db.insert(heartbeatRunEvents).values({ companyId, runId, agentId, seq, eventType: event.eventType, + stream: "system", level: "info", payload: event.payload }); + } + return runId; + } + + const identity = () => ({ processPid: deadPid, processGroupId: deadPid, processStartedAt: "2026-11-10T18:56:54.000Z" }); + const decide = (runId: string, patch: Partial = {}) => { + const { localStop: _localStop, ...evidence } = { ...proven, ...patch }; + return scheduledResumeContinuationAllowed(db, { companyId, runId, ...evidence }); + }; + + it("allows the continuation after the previous runner's verified stop", async () => { + const runId = await seedRun([ + { eventType: "native.process_start_requested", payload: { providerAttempt: 1 } }, + { eventType: "native.process_identity_recorded", payload: identity() }, + { eventType: "native.local_process_stopped", payload: identity() }, + ]); + expect(await decide(runId)).toEqual({ allowed: true }); + }); + + it.each<[string, () => Array<{ eventType: string; payload: Record }>]>([ + ["no stop was recorded", () => [ + { eventType: "native.process_identity_recorded", payload: identity() }, + ]], + ["a newer launch follows the stop", () => [ + { eventType: "native.process_identity_recorded", payload: identity() }, + { eventType: "native.local_process_stopped", payload: identity() }, + { eventType: "native.process_start_requested", payload: { providerAttempt: 2 } }, + ]], + ["the stopped process is still alive", () => [ + { eventType: "native.local_process_stopped", + payload: { ...identity(), processPid: process.pid, processGroupId: process.pid } }, + ]], + ["the run never had a process", () => []], + ])("refuses when %s", async (_name, events) => { + const runId = await seedRun(events()); + expect(await decide(runId)).toEqual({ allowed: false, reason: "local_stop_unverified" }); + }); + + it("never reads a stop receipt for a restart claim or a remote target", async () => { + const runId = await seedRun([{ eventType: "native.local_process_stopped", payload: identity() }]); + expect(await decide(runId, { restartRecovery: true })).toEqual({ allowed: false, reason: "restart_recovery_claim" }); + expect(await decide(runId, { targetKind: "remote" })).toEqual({ allowed: false, reason: "remote_target" }); + }); +}); diff --git a/server/src/services/native-runtime/scheduled-resume-continuation.ts b/server/src/services/native-runtime/scheduled-resume-continuation.ts new file mode 100644 index 00000000000..4f315655909 --- /dev/null +++ b/server/src/services/native-runtime/scheduled-resume-continuation.ts @@ -0,0 +1,57 @@ +import type { Db } from "@paperclipai/db"; +import { readNativeLocalProcessStop } from "../native-local-process-stop.js"; +import { NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT } from "./native-restart-recovery.js"; + +export type ScheduledResumeContinuationRefusal = + | "restart_recovery_claim" + | "remote_target" + | "not_a_scheduled_same_run_resume" + | "not_an_exact_checkpoint_resume" + | "failure_not_recoverable" + | "provider_session_not_established" + | "checkpoint_missing" + | "replacement_recorded" + | "attempt_not_admitted" + | "local_stop_unverified"; + +export interface ScheduledResumeContinuationEvidence { + restartRecovery: boolean; + targetKind: "local" | "remote"; + coordinatorPhase: string | null; + failureDetail: Record | null | undefined; + admittedAttempt: number; + localStop: { processPid: number } | null; +} + +export type ScheduledResumeContinuationDecision = + | { allowed: true } + | { allowed: false; reason: ScheduledResumeContinuationRefusal }; + +export function decideScheduledResumeContinuation( + evidence: ScheduledResumeContinuationEvidence, +): ScheduledResumeContinuationDecision { + const refuse = (reason: ScheduledResumeContinuationRefusal) => ({ allowed: false as const, reason }); + if (evidence.restartRecovery) return refuse("restart_recovery_claim"); + if (evidence.targetKind !== "local") return refuse("remote_target"); + if (evidence.coordinatorPhase !== "retryable_failure") return refuse("not_a_scheduled_same_run_resume"); + const detail = evidence.failureDetail ?? {}; + if (detail.recoveryMode !== "exact_checkpoint_resume") return refuse("not_an_exact_checkpoint_resume"); + if (detail.recoverable !== true) return refuse("failure_not_recoverable"); + if (detail.providerSessionEstablished !== true) return refuse("provider_session_not_established"); + if (detail.checkpointExists !== true) return refuse("checkpoint_missing"); + if (detail.successorRunId || detail.replacementDenied) return refuse("replacement_recorded"); + if (!Number.isSafeInteger(evidence.admittedAttempt) || evidence.admittedAttempt < 1 + || evidence.admittedAttempt > NATIVE_PROVIDER_RECOVERY_ATTEMPT_LIMIT) return refuse("attempt_not_admitted"); + if (!evidence.localStop) return refuse("local_stop_unverified"); + return { allowed: true }; +} + +export async function scheduledResumeContinuationAllowed( + db: Db, + input: Omit & { companyId: string; runId: string }, +): Promise { + const { companyId, runId, ...evidence } = input; + const localStop = evidence.restartRecovery || evidence.targetKind !== "local" + ? null : await readNativeLocalProcessStop(db, companyId, runId); + return decideScheduledResumeContinuation({ ...evidence, localStop }); +}