From aa94939988d578e02caca759bd3f43487769fcbb Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Thu, 12 Nov 2026 00:13:25 -0600 Subject: [PATCH] fix(native): continue a run whose remote Runner died mid-turn after a restart, without replaying the turn (#21039) 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 > - Native runs on a remote sandbox (Daytona) use a detached `paperclip-runnerd`. After a server restart, recovery reattaches to that Runner on the run's own sandbox lease > - On staging, a deploy detached the controller and left the Runner alive. The restart's reattachment then found the Runner process gone with a provider turn in flight. The run failed with `runner_remote_process_identity_unavailable` and the task waited for a person > - The run log said `startup reuse: environment (fresh) created`, so the sandbox looked re-created. It was not. The reattachment resumed the exact retained lease. The receipt reported the lease's original creation > - `requestInterruptedNativeTurnRetry` refused every unsettled turn (`provider_turn_unsettled`), and refused every system wake unless the journal proved that no turn was issued. So this run got no automatic continuation > - This pull request continues such a run once in a fresh provider session when the refused reattachment proved the Runner process gone. It records that the turn's outcome is unknown and never replays the turn. It also makes the environment receipt tell the truth > - The benefit is that a restart that kills a Runner in the middle of a turn no longer needs a Retry click ## Linked Issues or Issue Description Refs #20853, #20923, #21023, #20278, #20627, #20828 **What happened?** A staging deploy landed in the middle of a remote native Codex run on Daytona. Shutdown detached the controller and logged "durable runner left alive". The new server reattached through the run's original lease (same lease row, same provider sandbox id, same directory). Its process check then found the Runner gone: `process_verification` / `process_not_running`, `runnerLifecycle: ready`, `runnerSettled: true`, last diagnostic "WebSocket peer closed the connection", empty stderr. The provider checkpoint still had an active turn. The run failed with `runner_remote_process_identity_unavailable` / `native_continuation_requires_reconciliation`, and the task went to `blocked`. **Root cause** - `server/src/services/heartbeat/restart-interrupted-turn.ts` (`requestInterruptedNativeTurnRetry`): an unsettled last turn always returned `provider_turn_unsettled`, also when the refusal proved the Runner process gone. A system wake (here `transient_failure_retry`) continued only with `providerTurnIssued: false`. - The sandbox was not re-created. The environment and lease listings show one lease for the run, never replaced, and the final workspace snapshot used the same provider sandbox. `environmentStartupReuse` read `sandboxLeaseAcquisition: created` from the lease's first acquisition, so the reattachment's receipt said `fresh / created`. The 46 s acquire was the provider resume of that exact lease. The evidence available to GET requests cannot say what stopped the Runner process (the resume result is not kept). This pull request does not change that path. - #21023 named this run as not covered. #20853 resumes a Runner only at a settled, idle `ready` cut. Here the provider checkpoint has an active turn, so #20853 stays fail-closed, as it should. **Expected behavior** The run continues once by itself in a fresh provider session over the saved work. The turn whose outcome is unknown is recorded and is not replayed. ## What Changed - `restart-interrupted-turn.ts`: read the last `native.remote_runner_recovery_failed` refusal once. When it proves the Runner process gone, an unsettled last turn no longer blocks the automatic Retry. Before it requests the Retry, the server records `native.interrupted_turn_outcome_unknown` once (turn id, last turn event, `replayed: false`, closed `runnerExit` fields). The Retry is the existing one: saved-work recovery and stop proof first, then a fresh provider session (`forceFreshSession: true`), once per chain. - The same file: a system wake whose run is an automatic retry of owed work (`retryOfRunId` set) also continues under the run's responsible user once the refusal proved the Runner gone. Other system wakes (for example a monitor) keep their policy. - `packages/adapter-utils/src/startup-reuse.ts` and `native-session-executor.ts`: a `reattach_remote_runner` recovery reports its `environment` receipt as `reused / verified_reuse`. That path only resumes the run's exact retained lease and never acquires one. - Docs: `doc/execution-semantics.md` and `doc/run-log-events.md`. Previous-release story: no persisted shape changes. The rule reads the closed `remoteRunnerRecovery` fields that the previous release already writes; the new test uses the staging refusal payload as written. A run that a previous release already failed this way continues on the next completion or restart hook call only; runs already settled stay as they are and keep their Retry. The new event is additive. ## Verification - Red on `origin/master` (51d77648bc), with only the new tests: `cd server && npx vitest run src/services/runner-lost-mid-turn-continuation.test.ts` fails 4 of 6. The three continuation cases return `{ requested: false, reason: "provider_turn_unsettled" }`, and the monitor case returns that reason instead of `not_user_request`. - Green with the fix: 6 of 6. The cases cover a user's assignment through the restart claim, the staging shape (system `transient_failure_retry` with `retryOfRunId`), and a same-run retry. Each one checks one recorded unknown-outcome event, an idempotent second call, saved-work recovery first, one queued successor with `forceFreshSession: true`, and one Retry intent. Negative cases: a refusal that did not prove the process gone, no refusal, and a monitor wake. Each keeps the visible Retry and records nothing. - `cd packages/adapter-utils && npx vitest run src/startup-reuse.test.ts`: the new receipt case fails before (`fresh / created`) and passes after. - Focused suites pass: `workspace-recovery-retry-copyback.test.ts`, `governed-wait-runner-loss.test.ts`, `heartbeat-native-remote-failure-retention.test.ts`, `native-runner-restart-recovery.integration.test.ts` (302 passed, 14 skipped), and `native-session-executor.test.ts -t "startup reuse|restart|reattach"` (149 passed). - `pnpm --filter @paperclipai/server typecheck` and `pnpm --filter @paperclipai/adapter-utils typecheck` pass. - `pnpm verify` exits 2 because `doc/execution-semantics.md` is a global CI input. CI supplies that coverage. - Not run: a live Daytona restart. The change is in the server's recovery decision; the tests use the fixture's sandbox confinement. ## Risks - Medium. A run whose Runner died in the middle of a turn now continues by itself. The turn's outcome is unknown, so tool work from that turn can be partly done. The successor is a fresh session over the saved workspace; it does not replay the turn. Saved-work recovery stops and keeps the sandbox before the successor starts, so the old provider cannot still run. - Without positive proof that the Runner process is gone, an in-flight turn keeps the visible Retry, as before. - An automatic retry of owed work now continues after a lost Runner. Monitor and other system wakes without `retryOfRunId` do not. The bound is still one automatic Retry per chain. - The environment receipt changes only for the reattachment path. No execution decision reads it. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), via Claude Code with tool use (shell, file edits, test runs, read-only staging API reads). ## 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 Opus 5.5 --- doc/execution-semantics.md | 17 ++- doc/run-log-events.md | 15 ++ .../adapter-utils/src/startup-reuse.test.ts | 6 + packages/adapter-utils/src/startup-reuse.ts | 3 +- .../heartbeat/restart-interrupted-turn.ts | 82 +++++++++-- .../native-runtime/native-session-executor.ts | 3 +- .../runner-lost-mid-turn-continuation.test.ts | 137 ++++++++++++++++++ 7 files changed, 243 insertions(+), 20 deletions(-) create mode 100644 server/src/services/runner-lost-mid-turn-continuation.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 6407370d1ba..ac155e36497 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -1757,8 +1757,13 @@ without a semantic result. The reattachment then fails with `native_remote_recovery_lease_mismatch`, `native_plugin_recovery_lease_mismatch` or `runner_remote_process_identity_unavailable`. The server requests the task Retry itself, once per run, under the authority of the user who sent the message, -assigned the task, or clicked Retry. It does this only when the run recorded no -provider turn or its last provider turn is settled. Activity records the request +assigned the task, or clicked Retry. It does this when the run recorded no +provider turn, when its last provider turn is settled, or when the refused +reattachment proved the Runner process gone (`process_verification` / +`process_not_running`). In that last case an unsettled turn's outcome is +unknown: the server records `native.interrupted_turn_outcome_unknown` on the run +first, and the Retry starts a fresh provider session over the saved work, so the +turn is never replayed. Activity records the request as `restart-recovery` on behalf of that user. Admission keeps every gate: saved-work recovery, stop proof, approvals, pauses, budget and ownership. A Retry that the server requested gets no second automatic attempt. A system wake (for example a @@ -1769,8 +1774,12 @@ before it is sent, holds no turn command and no turn event (`providerTurnIssued: false` on `native.remote_runner_recovery_failed`). The run must also have no turn or tool event, and the Retry runs under the run's responsible user. Missing events alone never qualify, because a turn can reach -the provider before its events are stored. Other system wakes and runs with an -in-flight provider turn do not continue automatically. For these failed runs, a +the provider before its events are stored. An automatic retry of owed work (a +system wake whose run has `retryOfRunId`, such as `transient_failure_retry`) +also continues under the run's responsible user once the refusal proved its +Runner process gone, because losing the Runner to a restart is not the task's +failure. Other system wakes, and in-flight provider turns without that proof, +do not continue automatically. For these failed runs, a recovered workspace hold offers Retry (`canRetry`) and says "Retry or send a message to continue." An intentional Stop keeps "send a message" as its only continuation. diff --git a/doc/run-log-events.md b/doc/run-log-events.md index 5269e61926d..fd82824c201 100644 --- a/doc/run-log-events.md +++ b/doc/run-log-events.md @@ -31,6 +31,12 @@ or `provider_session`. Its state is `reused`, `fresh`, `unknown`, or `not_applicable`, and `evidence_missing`. No paths, identifiers, configuration values, or exception messages enter these fields. +A restart's reattachment to a remote Runner resumes the run's own retained +sandbox lease and never acquires another one. Its `environment` receipt is +`reused` / `verified_reuse`. Earlier releases reported the lease's original +acquisition there, so a reattachment to a sandbox the run created showed +`fresh` / `created` although no sandbox was created. + Provider-process receipts carry a cumulative `providerLaunchCount` for the run. `null` means unobserved; zero requires a physical launch observer. Direct ACPX processes count initial launches and respawns, including launches during retries. @@ -441,6 +447,15 @@ provider cleanup or warm transition is pending. These fields explain how the run stopped. The event is not recorded when a scheduled retry resumes a runner that exited after a settled suspension; see `doc/execution-semantics.md`. +`native.interrupted_turn_outcome_unknown` (level `warn`) is recorded once on a +failed run whose reattachment proved the Runner process gone +(`process_verification` / `process_not_running`) while its last provider turn +had no observed settlement, just before the server requests the automatic +Retry. The payload holds the provider `turnId` (or `null`), the last turn event +type, `replayed: false`, and the closed `runnerExit` fields `stage`, `reason` +and `runnerLifecycle`. It names a turn whose outcome is unknown; the Retry's +fresh session never replays it. + For a prepared remote workspace interrupted before its first executor claim, that ledger records `remote_execution_never_claimed` with a `bootstrap_incomplete` disposition. The provider attempt remains zero during diff --git a/packages/adapter-utils/src/startup-reuse.test.ts b/packages/adapter-utils/src/startup-reuse.test.ts index 9b4ae7bc012..9f4421a1fe9 100644 --- a/packages/adapter-utils/src/startup-reuse.test.ts +++ b/packages/adapter-utils/src/startup-reuse.test.ts @@ -60,6 +60,12 @@ describe("startup reuse evidence", () => { expect(environmentStartupReuse({ kind: "remote", transport: "sandbox", sandboxLeaseAcquisition: { outcome: "resumed" } })).toEqual({ state: "reused", reason: "verified_reuse" }); expect(environmentStartupReuse({ kind: "remote", transport: "sandbox", sandboxLeaseAcquisition: { outcome: "replacement", reason: "sensitive" } })).toEqual({ state: "fresh", reason: "replacement" }); }); + it("reports a restart's reattachment to the run's retained sandbox as reuse, not the lease's original creation", () => { + const retained = { kind: "remote", transport: "sandbox", sandboxLeaseAcquisition: { outcome: "created" } }; + expect(environmentStartupReuse(retained, { reattachedRetainedLease: true })).toEqual({ state: "reused", reason: "verified_reuse" }); + expect(environmentStartupReuse(retained)).toEqual({ state: "fresh", reason: "created" }); + expect(environmentStartupReuse({ kind: "local" }, { reattachedRetainedLease: true })).toEqual({ state: "not_applicable", reason: "not_applicable" }); + }); it("keeps a failing diagnostic sink out of the execution path", async () => { await expect(emitStartupReuse(async () => { throw new Error("sink unavailable"); }, "workspace", { state: "reused", reason: "verified_reuse" })).resolves.toBeUndefined(); }); diff --git a/packages/adapter-utils/src/startup-reuse.ts b/packages/adapter-utils/src/startup-reuse.ts index 749602bec08..32e643d7e7e 100644 --- a/packages/adapter-utils/src/startup-reuse.ts +++ b/packages/adapter-utils/src/startup-reuse.ts @@ -46,8 +46,9 @@ export async function emitStartupReuse( export function environmentStartupReuse(target: { kind: string; transport?: string; sandboxLeaseAcquisition?: { outcome: string; reason?: string } | null; -} | null | undefined): StartupReuseDecision { +} | null | undefined, options: { reattachedRetainedLease?: boolean } = {}): StartupReuseDecision { if (!target || target.kind === "local") return { state: "not_applicable", reason: "not_applicable" }; + if (options.reattachedRetainedLease) return { state: "reused", reason: "verified_reuse" }; const acquisition = target.sandboxLeaseAcquisition; if (target.transport !== "sandbox" || !acquisition) return { state: "unknown", reason: "evidence_missing" }; if (acquisition.outcome === "resumed") return { state: "reused", reason: "verified_reuse" }; diff --git a/server/src/services/heartbeat/restart-interrupted-turn.ts b/server/src/services/heartbeat/restart-interrupted-turn.ts index 16564be7238..628a2840974 100644 --- a/server/src/services/heartbeat/restart-interrupted-turn.ts +++ b/server/src/services/heartbeat/restart-interrupted-turn.ts @@ -1,5 +1,6 @@ import { and, desc, eq, inArray, isNull, sql } from "drizzle-orm"; import { agentWakeupRequests, heartbeatRunEvents, heartbeatRuns, issueComments, issues, type Db } from "@paperclipai/db"; +import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import type { NativeRestartRecoveryClaim } from "../native-runtime/native-restart-recovery.js"; import { isUserCommentContinuationReason } from "../explicit-native-continuation.js"; import { requestTaskRetry } from "../task-retry.js"; @@ -40,28 +41,72 @@ export function isProvenTurnSettlement(event: { eventType: string; payload: unkn const PROVIDER_TURN_ACTIVITY_EVENTS = [...TURN_STARTED_EVENTS, ...TURN_SETTLED_EVENTS, "tool.execution.started", "tool.execution.completed"]; +async function provenRunnerExit(db: Db, run: { companyId: string; id: string }) { + const [refusal] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), + eq(heartbeatRunEvents.eventType, "native.remote_runner_recovery_failed"), + )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); + const proof = asRecord(asRecord(refusal?.payload).remoteRunnerRecovery); + return proof.stage === "process_verification" && proof.reason === "process_not_running" ? proof : null; +} + /** * The run's last refused reattachment proved its Runner process gone and its * controller journal proved that no turn command was ever issued * (`providerTurnIssued: false`, recorded by the executor). No turn or tool * event may exist either. */ -async function hasProvenNoProviderTurn(db: Db, run: { companyId: string; id: string }) { +async function hasProvenNoProviderTurn(db: Db, run: { companyId: string; id: string }, runnerExit: Record | null) { + if (runnerExit?.providerTurnIssued !== false) return false; const [activity] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), inArray(heartbeatRunEvents.eventType, PROVIDER_TURN_ACTIVITY_EVENTS))).limit(1); - if (activity) return false; - const [refusal] = await db.select({ payload: heartbeatRunEvents.payload }).from(heartbeatRunEvents).where(and( + return !activity; +} + +export const INTERRUPTED_TURN_OUTCOME_UNKNOWN_EVENT = "native.interrupted_turn_outcome_unknown"; + +async function recordUnknownTurnOutcome(db: Db, run: { companyId: string; id: string; agentId: string }, + lastTurn: { eventType: string; payload: unknown }, runnerExit: Record) { + const [recorded] = await db.select({ id: heartbeatRunEvents.id }).from(heartbeatRunEvents).where(and( eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), - eq(heartbeatRunEvents.eventType, "native.remote_runner_recovery_failed"), - )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); - const proof = asRecord(asRecord(refusal?.payload).remoteRunnerRecovery); - return proof.stage === "process_verification" && proof.reason === "process_not_running" && proof.providerTurnIssued === false; + eq(heartbeatRunEvents.eventType, INTERRUPTED_TURN_OUTCOME_UNKNOWN_EVENT))).limit(1); + if (recorded) return; + const payload = asRecord(lastTurn.payload); + const turnId = asRecord(payload.prpEvent).turnId ?? payload.turnId; + await appendHeartbeatRunEvent(db, { companyId: run.companyId, runId: run.id, agentId: run.agentId, + eventType: INTERRUPTED_TURN_OUTCOME_UNKNOWN_EVENT, stream: "system", level: "warn", + message: "The Runner stopped during a provider turn, so the turn's outcome is unknown. It is not replayed; a fresh session continues from the saved work.", + payload: { turnId: typeof turnId === "string" ? turnId : null, lastTurnEvent: lastTurn.eventType, replayed: false, + runnerExit: { stage: runnerExit.stage, reason: runnerExit.reason, + ...(typeof runnerExit.runnerLifecycle === "string" ? { runnerLifecycle: runnerExit.runnerLifecycle } : {}) } } }); } /** Automatic continuation is requested at most once per chain of runs. */ export const RESTART_RETRY_KEY_PREFIX = "restart-interrupted-turn:"; +const MAX_RETRY_CHAIN = 32; + +async function chainUsedAutomaticRetry(db: Db, run: { companyId: string; id: string; retryOfRunId: string | null; + contextSnapshot: Record | null }) { + const visited = new Set(); + let current: typeof run | undefined = run; + while (current && !visited.has(current.id) && visited.size < MAX_RETRY_CHAIN) { + visited.add(current.id); + const retryId = current.contextSnapshot?.taskRetryRequestId; + if (typeof retryId === "string") { + const [receipt] = await db.select({ key: agentWakeupRequests.idempotencyKey }).from(agentWakeupRequests).where(and( + eq(agentWakeupRequests.companyId, run.companyId), sql`${agentWakeupRequests.id}::text = ${retryId}`)); + if (receipt?.key?.startsWith(`task-retry:${RESTART_RETRY_KEY_PREFIX}`)) return true; + } + if (!current.retryOfRunId) return false; + [current] = await db.select({ companyId: heartbeatRuns.companyId, id: heartbeatRuns.id, retryOfRunId: heartbeatRuns.retryOfRunId, + contextSnapshot: heartbeatRuns.contextSnapshot }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, current.retryOfRunId))); + } + return visited.size >= MAX_RETRY_CHAIN; +} + type Outcome = { requested: true; requestId: string } | { requested: false; reason: string }; /** @@ -69,11 +114,14 @@ type Outcome = { requested: true; requestId: string } | { requested: false; reas * native run. When it fails because the original runner or sandbox is gone, * request the same Retry a user would click, once and idempotently per run, * under the requesting user's authority. Only when the run's user input was not - * yet acted on, or its last provider turn is settled; an in-flight turn's - * outcome is unknown, so it keeps the visible Retry. A system wake continues - * this way only when the controller journal proved that no turn was ever - * issued, under the run's responsible user. Admission keeps every gate, - * including stop proof and saved-work recovery before the next turn. + * yet acted on, its last provider turn is settled, or the Runner process is + * proven gone; an in-flight turn's outcome is then unknown and recorded, and the + * Retry's fresh session never replays it. Without that proof the in-flight turn + * keeps the visible Retry. A system wake continues this way only when the + * controller journal proved that no turn was ever issued, or when it was an + * automatic retry of owed work whose Runner is proven gone, under the run's + * responsible user. Admission keeps every gate, including stop proof and + * saved-work recovery before the next turn. */ export async function requestRestartInterruptedTurnRetry(db: Db, claim: NativeRestartRecoveryClaim): Promise { if (claim.kind !== "reattach_remote_runner") return { requested: false, reason: "not_reattachment" }; @@ -92,7 +140,9 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): eq(heartbeatRunEvents.companyId, run.companyId), eq(heartbeatRunEvents.runId, run.id), inArray(heartbeatRunEvents.eventType, [...TURN_STARTED_EVENTS, ...TURN_SETTLED_EVENTS]), )).orderBy(desc(heartbeatRunEvents.seq)).limit(1); - if (lastTurn && !isProvenTurnSettlement(lastTurn)) return { requested: false, reason: "provider_turn_unsettled" }; + const runnerExit = await provenRunnerExit(db, run); + const unknownTurn = lastTurn && !isProvenTurnSettlement(lastTurn) ? lastTurn : null; + if (unknownTurn && !runnerExit) return { requested: false, reason: "provider_turn_unsettled" }; const [wake] = await db.select().from(agentWakeupRequests).where(and( eq(agentWakeupRequests.companyId, run.companyId), eq(agentWakeupRequests.agentId, run.agentId), eq(agentWakeupRequests.id, run.wakeupRequestId), eq(agentWakeupRequests.runId, run.id))); @@ -105,8 +155,11 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): // ever issued: nothing the run did can be repeated, so the Retry runs once // under the run's responsible user. Missing events alone prove nothing, // because a turn can reach the provider before its events are stored. - if (!wake || reason === "task_retry" || !run.responsibleUserId || !await hasProvenNoProviderTurn(db, run)) + const owedRedelivery = run.retryOfRunId !== null && runnerExit !== null; + if (!wake || reason === "task_retry" || !run.responsibleUserId + || !(owedRedelivery || await hasProvenNoProviderTurn(db, run, runnerExit))) return { requested: false, reason: "not_user_request" }; + if (owedRedelivery && await chainUsedAutomaticRetry(db, run)) return { requested: false, reason: "automatic_retry_used" }; actorId = run.responsibleUserId; } else if (reason === "task_retry") { // A Retry this hook already requested gets no second automatic attempt. @@ -130,6 +183,7 @@ export async function requestInterruptedNativeTurnRetry(db: Db, runId: string): .where(and(eq(issues.companyId, run.companyId), eq(issues.id, issueId))); if (!issue || issue.assigneeAgentId !== run.agentId || ["done", "cancelled"].includes(issue.status)) return { requested: false, reason: "task_changed" }; + if (unknownTurn && runnerExit) await recordUnknownTurnOutcome(db, run, unknownTurn, runnerExit); try { const receipt = await requestTaskRetry({ db, companyId: run.companyId, issueId, actorId, actorSource: actorId === "local-board" ? "local_implicit" : "session", diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index d474b8f4c40..00c2035846d 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -10637,7 +10637,8 @@ async function executePaperclipNativeSessionAttempt( }, }); nativeRunRecoveryFlushers.set(traceRunId, flushRecovery); - await emitStartupReuse(input.onEvent, "environment", environmentStartupReuse(input.runnerExecutionTarget)); + await emitStartupReuse(input.onEvent, "environment", environmentStartupReuse(input.runnerExecutionTarget, + { reattachedRetainedLease: input.restartRecovery?.kind === "reattach_remote_runner" })); if (!input.runnerExecutionTarget || input.runnerExecutionTarget.kind === "local") { await emitStartupReuse(input.onEvent, "workspace", { state: "not_applicable", reason: "not_applicable" }); } diff --git a/server/src/services/runner-lost-mid-turn-continuation.test.ts b/server/src/services/runner-lost-mid-turn-continuation.test.ts new file mode 100644 index 00000000000..add97007f3c --- /dev/null +++ b/server/src/services/runner-lost-mid-turn-continuation.test.ts @@ -0,0 +1,137 @@ +import { randomUUID } from "node:crypto"; +import { readFile } from "node:fs/promises"; +import path from "node:path"; +import { and, eq } from "drizzle-orm"; +import { describe, expect, it, vi } from "vitest"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { agentWakeupRequests, heartbeatRunEvents, heartbeatRuns } from "@paperclipai/db"; +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { heartbeatService } from "./heartbeat.js"; +import { requestInterruptedNativeTurnRetry, requestRestartInterruptedTurnRetry } from "./heartbeat/restart-interrupted-turn.js"; +import { readSystemRecovery } from "./system-recovery-state.js"; +import { reconcileRetainedWorkspaces } from "./workspace-recovery.js"; +import { useWorkspaceRecoveryFixture, action, db, run, seed } from "./__tests__/workspace-recovery-fixture.js"; + +const targets = vi.hoisted(() => new Map()); +vi.mock("./environment-execution-target.js", async importOriginal => { + const actual = await importOriginal(); + return { ...actual, resolveEnvironmentExecutionTarget: async (input: Parameters[0]) => + (input.leaseId ? targets.get(input.leaseId) : undefined) ?? actual.resolveEnvironmentExecutionTarget(input) }; +}); + +const TURN_ID = "01a1cdcb-8c83-7221-b602-d80d515d1616"; + +/** The refusal the previous release recorded on staging, byte for byte in shape. */ +const stagingRefusal = (reason: "process_not_running" | "process_identity_changed" = "process_not_running") => ({ + remoteRunnerRecovery: { stage: "process_verification", reason, + ...(reason === "process_not_running" ? { runnerSettled: true, runnerLifecycle: "ready", + runnerDiagnostics: ["WebSocket peer closed the connection"], runnerExitEvidence: "== runnerd.process-stderr.log" } : {}) }, +}); + +describe("a remote Runner that died in the middle of a provider turn", () => { + useWorkspaceRecoveryFixture(targets); + + async function runnerLostMidTurn(input: { wake: "user" | "system_retry" | "system_monitor"; + refusal?: "process_not_running" | "process_identity_changed" | null }) { + const f = await seed("native"); + const original = await run(f.runId); + const system = input.wake !== "user"; + const reason = input.wake === "user" ? "issue_assigned" : input.wake === "system_retry" ? "transient_failure_retry" : "issue_monitor_due"; + const [predecessor] = input.wake === "system_retry" ? await db.insert(heartbeatRuns).values({ companyId: f.companyId, + agentId: original.agentId, issueId: f.issueId, status: "failed", createdAt: new Date(Date.now() - 60 * 60_000), + finishedAt: new Date(Date.now() - 50 * 60_000), + errorCode: "environment_provider_unavailable", contextSnapshot: { issueId: f.issueId, wakeReason: "issue_assigned" } }).returning() : []; + const [wake] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: original.agentId, + source: system ? "automation" : "assignment", triggerDetail: "system", reason, status: "failed", runId: f.runId, + requestedByActorType: system ? "system" : "user", requestedByActorId: system ? null : "local-board", + finishedAt: new Date(), payload: { issueId: f.issueId } }).returning(); + await db.update(heartbeatRuns).set({ errorCode: "native_continuation_requires_reconciliation", + error: "runner_remote_process_identity_unavailable", wakeupRequestId: wake.id, responsibleUserId: "local-board", + retryOfRunId: predecessor?.id ?? null, startedAt: new Date(Date.now() - 45 * 60_000), + contextSnapshot: { issueId: f.issueId, wakeReason: reason, ...(predecessor ? { retryOfRunId: predecessor.id, + retryReason: "transient_failure", scheduledRetryAttempt: 1 } : {}) }, + }).where(eq(heartbeatRuns.id, f.runId)); + for (const eventType of ["turn.submitted", "turn.started", "tool.execution.started", "turn.accepted", "tool.execution.completed"]) + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: original.agentId, + eventType, stream: "system", level: "info", message: eventType, + payload: { prpEvent: { eventType, turnId: TURN_ID, payload: {} } } }); + if (input.refusal !== null) await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: original.agentId, eventType: "native.remote_runner_recovery_failed", stream: "system", level: "error", + message: "Could not verify the existing remote runner.", payload: stagingRefusal(input.refusal) }); + const claim = { kind: "reattach_remote_runner" as const, runId: f.runId, leaseOwner: `boot:2:${randomUUID()}`, + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard" as const, recoveryRequestId: null, + remote: { providerLeaseId: f.providerLeaseId, remoteCwd: f.remote } }; + const service = heartbeatService(db, { environmentRuntime: f.environmentRuntime }); + // Keep the agent's only slot busy so a successor stays observable as queued. + await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: original.agentId, status: "running" }); + return { ...f, agentId: original.agentId, claim, service }; + } + const queued = async (companyId: string) => db.select().from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, companyId), eq(heartbeatRuns.status, "queued"))); + const unknownOutcomeEvents = async (runId: string) => db.select().from(heartbeatRunEvents) + .where(and(eq(heartbeatRunEvents.runId, runId), eq(heartbeatRunEvents.eventType, "native.interrupted_turn_outcome_unknown"))); + + // Staging: a deploy detached the controller, the restart's reattachment then + // found the Runner process gone with a turn in flight. The previous release + // recorded `process_not_running` and the task waited for a person. + it.each([ + { title: "a user's assignment", wake: "user" as const, viaClaim: true }, + { title: "an automatic retry of the assignment (the staging shape)", wake: "system_retry" as const, viaClaim: true }, + { title: "a same-run retry of a user's assignment", wake: "user" as const, viaClaim: false }, + ])("continues $title once in a fresh session and records that the turn's outcome is unknown", async ({ wake, viaClaim }) => { + const f = await runnerLostMidTurn({ wake }); + const request = () => viaClaim ? requestRestartInterruptedTurnRetry(db, f.claim) : requestInterruptedNativeTurnRetry(db, f.runId); + expect(await request()).toEqual({ requested: true, requestId: expect.any(String) }); + expect(await request()).toMatchObject({ requested: true }); + const recorded = await unknownOutcomeEvents(f.runId); + expect(recorded).toHaveLength(1); + expect(recorded[0]).toMatchObject({ level: "warn", payload: { turnId: TURN_ID, replayed: false, + runnerExit: { stage: "process_verification", reason: "process_not_running", runnerLifecycle: "ready" } } }); + await vi.waitFor(async () => { + await Promise.all(await reconcileRetainedWorkspaces(db, f.environmentRuntime, current => f.service.resumeRemoteStopComments(current)) ?? []); + expect(readSystemRecovery((await action(f.actionId)).evidence)?.state).toBe("complete"); + }, { timeout: 10_000, interval: 10 }); + expect(await readFile(path.join(f.host, "saved.txt"), "utf8")).toBe("finished work\n"); + await vi.waitFor(async () => expect(await queued(f.companyId)).toHaveLength(1), { timeout: 10_000, interval: 50 }); + const [successor] = await queued(f.companyId); + expect(successor.contextSnapshot).toMatchObject({ previousRunId: f.runId, wakeReason: "task_retry", forceFreshSession: true }); + const retries = await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.reason, "task_retry"))); + expect(retries.filter(retry => retry.idempotencyKey?.startsWith("task-retry:"))).toEqual([expect.objectContaining({ + idempotencyKey: `task-retry:restart-interrupted-turn:${f.runId}`, requestedByActorType: "user", requestedByActorId: "local-board" })]); + }, 30_000); + + it.each([ + { title: "inherits the hook's Retry intent", inherit: true }, + { title: "reaches the hook's Retry run through its retry chain", inherit: false }, + ])("gives no second automatic Retry to a system retry that $title and then lost its Runner mid-turn", async ({ inherit }) => { + const f = await runnerLostMidTurn({ wake: "system_retry" }); + const current = await run(f.runId); + const [hookIntent] = await db.insert(agentWakeupRequests).values({ companyId: f.companyId, agentId: f.agentId, + source: "on_demand", triggerDetail: "manual", reason: "task_retry", status: "completed", + idempotencyKey: `task-retry:restart-interrupted-turn:${randomUUID()}`, requestedByActorType: "user", + requestedByActorId: "local-board", payload: { issueId: f.issueId } }).returning(); + await db.update(heartbeatRuns).set({ contextSnapshot: { ...current.contextSnapshot, wakeReason: "task_retry", + taskRetryRequestId: hookIntent.id } }).where(eq(heartbeatRuns.id, current.retryOfRunId!)); + if (inherit) await db.update(heartbeatRuns).set({ contextSnapshot: { ...current.contextSnapshot, + taskRetryRequestId: hookIntent.id } }).where(eq(heartbeatRuns.id, f.runId)); + expect(await requestRestartInterruptedTurnRetry(db, f.claim)).toEqual({ requested: false, reason: "automatic_retry_used" }); + expect(await unknownOutcomeEvents(f.runId)).toHaveLength(0); + expect((await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.reason, "task_retry")))).filter(retry => retry.id !== hookIntent.id)).toHaveLength(0); + }, 30_000); + + it.each([ + { title: "the refusal did not prove the Runner process gone", wake: "user" as const, refusal: "process_identity_changed" as const, + reason: "provider_turn_unsettled" }, + { title: "no reattachment refusal was recorded", wake: "user" as const, refusal: null, reason: "provider_turn_unsettled" }, + { title: "the source was a monitor wake, not a retry of owed work", wake: "system_monitor" as const, refusal: "process_not_running" as const, + reason: "not_user_request" }, + ])("keeps the visible Retry and records nothing when $title", async ({ wake, refusal, reason }) => { + const f = await runnerLostMidTurn({ wake, refusal }); + expect(await requestRestartInterruptedTurnRetry(db, f.claim)).toEqual({ requested: false, reason }); + expect(await unknownOutcomeEvents(f.runId)).toHaveLength(0); + expect(await db.select().from(agentWakeupRequests).where(and(eq(agentWakeupRequests.companyId, f.companyId), + eq(agentWakeupRequests.reason, "task_retry")))).toHaveLength(0); + }, 30_000); +});