From 177f642da772a177d105d21f6a291214dc675aee Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sat, 7 Nov 2026 15:12:36 -0600 Subject: [PATCH] fix(recovery): continue native runs that failed before their first claim (#19911) 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 go through a coordinator claim before any provider work starts > - A run can fail before that claim, for example while it restores the previous harness backup > - The failure path then records a no-replay hold, as if the run had acted > - Workspace recovery saves the files, but the task stays blocked and asks a person to send a message > - Recoveries must continue without a human step when no side effect is possible > - This pull request lets the existing safe-replacement worker continue such runs once, after their files are saved > - The benefit is that these tasks resume automatically, and runs that may have acted keep today's reconciliation ## Linked Issues or Issue Description **What happened?** Native runs that failed before their first coordinator claim were never continued automatically. They had no claim, no provider process and no provider or tool events. Their result recorded `executionRecovery: { kind: "bootstrap", providerWorkStarted: false }`. The failure path still moved the coordinator to `terminal_failure` with `native_continuation_requires_reconciliation` and recorded a resolved `preserve_without_replay_v1` hold (replay blocked, board owner). Workspace recovery then saved the files, but the next action stayed "Saved work recovered. Send a message to continue." `nativeReplacementPending` accepted only a few failure codes, and the interrupted-preparation path covers only legacy runs. **Expected behavior** When a native run provably failed before its first claim, Paperclip continues it once, automatically, after its files are saved (or when it had no workspace to save). Runs with a claim, a process, a provider or tool event, or recorded provider work keep the existing reconciliation. **Steps to reproduce** 1. Start a native run with a remote workspace. 2. Make it fail before `native.coordinator.claim` (for example, abort the harness-backup restore). 3. Let workspace recovery save the files. 4. See that the task stays blocked and asks for a message. **Paperclip version or commit** Observed on e2d490c61. Reproduced on master before this change. Refs #19906 (fixes the restore deadline that caused the observed failures). ## What Changed - `native-safe-replacement.ts`: add one class-level candidate, "failed before its first claim". It requires the `native_continuation_requires_reconciliation` coordinator at attempt 0 and controller generation 0, with no lease, assessment, decision or result. - The candidate also requires no process identity, no `sessionIdAfter`, no provider/turn/item/tool/`run.result` events, no tool invocations, `providerWorkStarted` not `true`, no Stop or cancellation, no explicit user continuation, and no later run for the task. - A no-replay hold must have completed file recovery (the existing `completedWorkspaceRecoveryCondition`). A blocked task must carry this failure's block receipt at the current status version. - Discovery checks saved-file evidence (snapshot, stopped sandbox) or a workspace-less run before it selects the candidate. - The reconciler skips the stopped-session probe for this class. It feeds the decision with known failure meaning and a complete effect inventory, since no provider ran. It rechecks the candidate under the task and coordinator locks before it inserts the successor. - The successor is the existing fresh-session `native_safe_replacement`. Its context carries `nativePreClaimContinuationOf`, so a failed successor is not continued again. The existing transaction resolves the hold (`replay: verified_safe_replacement`) and restores the task block. - `doc/execution-semantics.md`: describe the rule. Previous-release story: holds recorded by earlier releases (resolved, `replay: blocked`) carry forward. The test seeds that exact shape. No schema or persisted identity changes. ## Verification - `cd server && npx vitest run src/services/native-preclaim-continuation.test.ts` - Before the fix: "continues automatically exactly once after its saved work is recovered" fails (no successor). - After the fix: all 9 tests pass. - The tests cover: the staging shape with an old-release hold continues exactly once under concurrent and repeated sweeps; it waits for saved files; it does not chain from its own successor; and it keeps reconciliation when the coordinator claimed, a provider event exists, a tool call started, provider work is recorded, a process started, or a later run owns the task. - `npx vitest run src/services/native-runtime/native-safe-replacement.test.ts` passes (62 tests). - `pnpm --filter @paperclipai/server typecheck` passes. ## Risks - The successor uses a fresh provider session, like the other safe replacements. History comes from `retryOfRunId`. - The new predicate adds one indexed `not exists` on `heartbeat_runs (company_id, context_snapshot->>'issueId', created_at)` per candidate. It runs only for the narrow pre-claim failure code. - A run whose prepared workspace uses the older `nativeWorkspaceSync` descriptor without remote-workspace evidence stays held, as today. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, with tool use and code execution in Claude Code. ## 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 - [x] All Paperclip CI gates are green - [x] 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 | 14 ++ .../native-preclaim-continuation.test.ts | 167 ++++++++++++++++++ .../native-runtime/native-safe-replacement.ts | 108 +++++++++-- 3 files changed, 279 insertions(+), 10 deletions(-) create mode 100644 server/src/services/native-preclaim-continuation.test.ts diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index da7020e5abe..93a75dccc37 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4033,6 +4033,20 @@ its bound recovery and finalization receipts prove completion. Unrelated holds, later task blocks, normal dispatch admission, and the three-attempt budget remain effective. +A native run can also fail before its first coordinator claim, for example +while it restores the previous harness backup. The claim is the only gate to +provider dispatch, so a coordinator still at attempt 0 and controller generation +0 proves that no provider process, turn or tool started. The failure path records +`native_continuation_requires_reconciliation` with a no-replay file hold. Once +that hold's file recovery completes, or the run had no workspace to save, the +same safe-replacement worker schedules one fresh-session continuation and +retires the hold and its task block. It requires no process identity, no +provider, turn, item or tool events, no tool invocations, no recorded provider +work, no Stop, no explicit user continuation and no later run for the task. +Holds recorded by earlier releases take this path. The successor is marked +`nativePreClaimContinuationOf`; if it also fails, it keeps the ordinary +reconciliation hold instead of chaining another continuation. + Historical replacement discovery reads at most 100 candidates per pass and keeps a cursor only for the remaining scan. A work notification preserves forward progress and requests a final rescan, so held earlier rows cannot hide a later diff --git a/server/src/services/native-preclaim-continuation.test.ts b/server/src/services/native-preclaim-continuation.test.ts new file mode 100644 index 00000000000..95929b02e1f --- /dev/null +++ b/server/src/services/native-preclaim-continuation.test.ts @@ -0,0 +1,167 @@ +import { randomUUID } from "node:crypto"; +import { and, eq } from "drizzle-orm"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { agentWakeupRequests, heartbeatRuns, issueRecoveryActions, issues, nativeRunFinalizations, nativeRunResults } from "@paperclipai/db"; +import type { AdapterSandboxExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; +import { appendHeartbeatRunEvent } from "./heartbeat-run-events.js"; +import { recoverRetainedWorkspace } from "./workspace-recovery.js"; +import { adoptSystemRecovery } from "./system-recovery.js"; +import { readWorkspaceRecovery } from "./workspace-recovery-receipt.js"; +import { getExecutionBlocker } from "./execution-blocker.js"; +import { invalidateNativeReplacementDiscovery, reconcileSafeNativeReplacements } from "./native-runtime/native-safe-replacement.js"; +import * as storageBackend from "../storage/index.js"; +import { acceptedRemoteExport, action, connectionString, db, run, useWorkspaceRecoveryFixture } 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) }; +}); + +/** + * The staging shape of a native run that failed before its first coordinator + * claim (2026-11-07: a resumed run aborted at the harness-backup restore + * deadline). No provider process, no claim, no provider or tool events. The + * failure path moved the coordinator to `terminal_failure` with attempt 0 and + * an earlier release recorded a resolved `preserve_without_replay_v1` hold + * that blocks replay and the task. + */ +async function preClaimNativeFailure() { + const f = await acceptedRemoteExport("published"); + const source = await run(f.runId); + await db.update(nativeRunFinalizations).set({ resultId: null }).where(eq(nativeRunFinalizations.runId, f.runId)); + await db.delete(nativeRunResults).where(eq(nativeRunResults.runId, f.runId)); + await db.update(nativeRunFinalizations).set({ phase: "terminal_failure", attempt: 0, leaseOwner: null, leaseExpiresAt: null, + failureCode: "native_continuation_requires_reconciliation", failureDetail: null }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + const recovery = { schema: "paperclip.workspace-restore-recovery.v1", leaseIds: [f.lease.id] }; + const remoteWorkspace = source.runnerProfileJson!.remoteWorkspace as { remoteCwd: string }; + await db.update(heartbeatRuns).set({ + status: "failed", finishedAt: new Date(), error: "The operation was aborted", + errorCode: "native_continuation_requires_reconciliation", completionContractId: null, + sessionIdBefore: "ses_previous_provider_session", nativeSessionId: randomUUID(), nativePhase: "terminal_failure", + contextSnapshot: { issueId: f.issueId, wakeReason: "External chat message received" }, + resultJson: { stopReason: "adapter_failed", timeoutFired: false, timeoutSource: "config", + executionRecovery: { kind: "bootstrap", providerWorkStarted: false }, timeoutConfigured: false, effectiveTimeoutSec: 0, + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery }, + runnerProfileJson: { ...source.runnerProfileJson, + nativeExecutionInput: { workspace: { cwd: remoteWorkspace.remoteCwd } } }, + }).where(eq(heartbeatRuns.id, f.runId)); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: source.agentId, + eventType: "lifecycle", stream: "system", level: "info", message: "run started" }); + await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, agentId: source.agentId, + eventType: "error", stream: "system", level: "error", message: "The operation was aborted" }); + await db.update(issues).set({ status: "blocked", statusVersion: 18, executionRunId: null, checkoutRunId: null }) + .where(eq(issues.id, f.issueId)); + // The hold exactly as e2d490c61 recorded it (resolved, replay blocked, board-owned). + await db.update(issueRecoveryActions).set({ + cause: "native_continuation_requires_reconciliation", fingerprint: `native-workspace:${f.runId}`, + status: "resolved", outcome: "blocked", ownerType: "board", resolvedAt: new Date(), + nextAction: "The original sandbox is retained for workspace repair. Verify its exact stop receipt, recover the missing files, and record workspaceRepairEvidence without changing the current task. Repair does not replay the stopped run. Explicitly remove the retained allocation after recovery.", + evidence: { runId: f.runId, originalFailureCode: null, continuationDelivery: "invalidated", + workspaceRestoreFailure: "restore_failed", workspaceRestoreRecovery: recovery, + automaticRecovery: { policy: "preserve_without_replay_v1", runId: f.runId, replay: "blocked", + actionOutcome: "unknown", recordedAt: new Date().toISOString() }, + nativeFailureBlock: { runId: f.runId, statusVersion: 18 } }, + }).where(eq(issueRecoveryActions.id, f.actionId)); + await adoptSystemRecovery(db, f); + return { ...f, agentId: source.agentId }; +} + +async function successorsOf(runId: string) { + return db.select().from(heartbeatRuns).where(eq(heartbeatRuns.retryOfRunId, runId)); +} + +async function sweep() { + invalidateNativeReplacementDiscovery(db); + return reconcileSafeNativeReplacements(db); +} + +describe("native run that failed before its first coordinator claim", () => { + useWorkspaceRecoveryFixture(targets); + // A published remote workspace: the shape staging persists and restores. + const priorEnv = { databaseUrl: process.env.DATABASE_URL, masterKey: process.env.PAPERCLIP_SECRETS_MASTER_KEY }; + beforeEach(() => { + const local = storageBackend.getStorageService(); + vi.spyOn(storageBackend, "getStorageService").mockReturnValue({ ...local, provider: "s3" }); + process.env.DATABASE_URL = connectionString; process.env.PAPERCLIP_SECRETS_MASTER_KEY = "63".repeat(32); + }); + afterEach(() => { + vi.restoreAllMocks(); + if (priorEnv.databaseUrl === undefined) delete process.env.DATABASE_URL; else process.env.DATABASE_URL = priorEnv.databaseUrl; + if (priorEnv.masterKey === undefined) delete process.env.PAPERCLIP_SECRETS_MASTER_KEY; + else process.env.PAPERCLIP_SECRETS_MASTER_KEY = priorEnv.masterKey; + }); + + it("continues automatically exactly once after its saved work is recovered", async () => { + const f = await preClaimNativeFailure(); + await recoverRetainedWorkspace(f.input); + expect(readWorkspaceRecovery((await action(f.actionId)).evidence)).toMatchObject({ status: "recovered" }); + + await Promise.all([sweep(), sweep()]); + await sweep(); + + const successors = await successorsOf(f.runId); + expect(successors).toHaveLength(1); + expect(successors[0]).toMatchObject({ status: "scheduled_retry", agentId: f.agentId, + scheduledRetryReason: "native_safe_replacement", + contextSnapshot: { issueId: f.issueId, retryOfRunId: f.runId, nativePreClaimContinuationOf: f.runId } }); + const [wake] = await db.select().from(agentWakeupRequests).where(eq(agentWakeupRequests.runId, successors[0].id)); + expect(wake).toMatchObject({ status: "queued", requestedByActorType: "system" }); + const [coordinator] = await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, f.runId)); + expect(coordinator.failureDetail).toMatchObject({ successorRunId: successors[0].id, preClaimContinuation: true }); + const hold = await action(f.actionId); + expect(hold).toMatchObject({ status: "resolved", outcome: "handed_back", + evidence: { automaticRecovery: { replay: "verified_safe_replacement", successorRunId: successors[0].id } } }); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task).toMatchObject({ status: "in_progress", executionRunId: null, checkoutRunId: null }); + expect(await getExecutionBlocker(db, f.companyId, f.issueId)).toBeNull(); + }, 60_000); + + it("waits for saved work before continuing", async () => { + const f = await preClaimNativeFailure(); + await sweep(); + expect(await successorsOf(f.runId)).toHaveLength(0); + const [task] = await db.select().from(issues).where(eq(issues.id, f.issueId)); + expect(task.status).toBe("blocked"); + }, 60_000); + + it("does not chain a second automatic continuation from its own successor", async () => { + const f = await preClaimNativeFailure(); + await db.update(heartbeatRuns).set({ contextSnapshot: { issueId: f.issueId, nativePreClaimContinuationOf: randomUUID() } }) + .where(eq(heartbeatRuns.id, f.runId)); + await recoverRetainedWorkspace(f.input); + await sweep(); + expect(await successorsOf(f.runId)).toHaveLength(0); + }, 60_000); + + it.each(["claimed", "provider_event", "tool_call", "uncertain_side_effect", "process", "superseded"] as const)( + "keeps reconciliation when the run may have acted: %s", async gate => { + const f = await preClaimNativeFailure(); + await recoverRetainedWorkspace(f.input); + expect(readWorkspaceRecovery((await action(f.actionId)).evidence)).toMatchObject({ status: "recovered" }); + if (gate === "claimed") await db.update(nativeRunFinalizations).set({ attempt: 1, controllerGeneration: 1 }) + .where(eq(nativeRunFinalizations.runId, f.runId)); + if (gate === "provider_event") await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: f.agentId, eventType: "turn.started", stream: "system", level: "info", payload: {} }); + if (gate === "tool_call") await appendHeartbeatRunEvent(db, { companyId: f.companyId, runId: f.runId, + agentId: f.agentId, eventType: "tool.execution.started", stream: "system", level: "info", + payload: { name: "paperclip_create_issue", executionId: randomUUID() } }); + if (gate === "uncertain_side_effect") { + const current = await run(f.runId); + await db.update(heartbeatRuns).set({ resultJson: { ...current.resultJson, + executionRecovery: { kind: "bootstrap", providerWorkStarted: true } } }).where(eq(heartbeatRuns.id, f.runId)); + } + if (gate === "process") await db.update(heartbeatRuns).set({ processPid: 2_000_000_000, processStartedAt: new Date() }) + .where(eq(heartbeatRuns.id, f.runId)); + if (gate === "superseded") await db.insert(heartbeatRuns).values({ companyId: f.companyId, agentId: f.agentId, + status: "succeeded", contextSnapshot: { issueId: f.issueId }, createdAt: new Date(Date.now() + 1000) }); + await sweep(); + expect(await successorsOf(f.runId)).toHaveLength(0); + const hold = await action(f.actionId); + expect(hold).toMatchObject({ outcome: "blocked", evidence: { automaticRecovery: { replay: "blocked" } } }); + const [task] = await db.select().from(issues).where(and(eq(issues.id, f.issueId))); + expect(task.status).toBe("blocked"); + }, 60_000); +}); diff --git a/server/src/services/native-runtime/native-safe-replacement.ts b/server/src/services/native-runtime/native-safe-replacement.ts index b2ce3bef1bc..cb0b6100da5 100644 --- a/server/src/services/native-runtime/native-safe-replacement.ts +++ b/server/src/services/native-runtime/native-safe-replacement.ts @@ -29,6 +29,7 @@ import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { hasRemoteTerminationReceipt } from "../remote-execution-termination.js"; import { NATIVE_PROVIDER_DURABLE_EVENT_TYPES } from "./native-retired-backup-source.js"; import { completedWorkspaceRecoveryCondition } from "../workspace-recovery-receipt.js"; +import { nativeRunHasNoWorkspaceAccess } from "../workspace-restore-recovery-state.js"; export const NATIVE_SAFE_REPLACEMENT_REASON = "native_safe_replacement"; const record = (v: unknown): Record => @@ -107,6 +108,66 @@ async function historicalListenerGuardEvidence(db: ReplacementDb, run: typeof he return Boolean(evidence); } +/** + * A native run that failed before its first coordinator claim. The claim is the + * only gate to provider dispatch: it moves the coordinator past attempt 0 and + * records a controller generation. A run that never claimed started no provider + * process, turn or tool, so a fresh session can continue the task once its + * prepared files are saved. Any claim, process, provider or tool event, user + * Stop, or recorded provider work keeps today's reconciliation. One automatic + * continuation per chain: a continuation that itself fails is not continued. + */ +export const NATIVE_PRE_CLAIM_FAILURE_CODE = "native_continuation_requires_reconciliation"; +function preClaimFailurePending() { + const providerEvents = sql.join([...NATIVE_PROVIDER_DURABLE_EVENT_TYPES].map(value => sql`${value}`), sql`, `); + return and( + eq(nativeRunFinalizations.failureCode, NATIVE_PRE_CLAIM_FAILURE_CODE), + eq(heartbeatRuns.runtimeMode, "native"), + eq(nativeRunFinalizations.attempt, 0), eq(nativeRunFinalizations.controllerGeneration, 0), + isNull(nativeRunFinalizations.leaseOwner), isNull(nativeRunFinalizations.leaseExpiresAt), + isNull(nativeRunFinalizations.assessmentId), isNull(nativeRunFinalizations.decisionId), + isNull(heartbeatRuns.processPid), isNull(heartbeatRuns.processGroupId), + isNull(heartbeatRuns.processStartedAt), isNull(heartbeatRuns.sessionIdAfter), + sql`${heartbeatRuns.nativeIssueId} = ${nativeRunFinalizations.issueId}`, + sql`coalesce(${heartbeatRuns.resultJson}#>'{executionRecovery,providerWorkStarted}', 'false'::jsonb) = 'false'::jsonb`, + sql`not (coalesce(${heartbeatRuns.resultJson}, '{}'::jsonb) ?| array['cancellation', 'nativeCancellation', 'executionCancellation'])`, + sql`coalesce(${heartbeatRuns.contextSnapshot}->>'nativePreClaimContinuationOf', '') = ''`, + sql`not exists (select 1 from ${heartbeatRunEvents} event where event.company_id = ${heartbeatRuns.companyId} + and event.run_id = ${heartbeatRuns.id} and (event.source_event_id is not null + or event.event_type in (${providerEvents}) or event.event_type ~ '^(turn|item|tool|semantic_tool|run[.]result)[.]'))`, + sql`not exists (select 1 from ${toolInvocations} invocation where invocation.company_id = ${heartbeatRuns.companyId} + and invocation.run_id = ${heartbeatRuns.id})`, + // Saved files come first. A no-replay hold settles only with its completed recovery. + sql`not exists (select 1 from ${issueRecoveryActions} + where ${issueRecoveryActions.companyId} = ${heartbeatRuns.companyId} + and ${issueRecoveryActions.sourceIssueId} = ${nativeRunFinalizations.issueId} + and ${issueRecoveryActions.evidence}->>'runId' = ${heartbeatRuns.id}::text + and (${issueRecoveryActions.evidence} ? 'explicitUserContinuation' + or (${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked' + and not coalesce((${completedWorkspaceRecoveryCondition(heartbeatRuns.id)}), false))))`, + // The task has not moved on: a later run (such as a user's message) owns it. + sql`not exists (select 1 from ${heartbeatRuns} later where later.company_id = ${heartbeatRuns.companyId} + and (later.context_snapshot ->> 'issueId') = ${nativeRunFinalizations.issueId}::text + and later.created_at > ${heartbeatRuns.createdAt} and later.id <> ${heartbeatRuns.id})`, + sql`exists (select 1 from ${issues} task where task.company_id = ${heartbeatRuns.companyId} + and task.id = ${nativeRunFinalizations.issueId} and task.assignee_agent_id = ${heartbeatRuns.agentId} + and task.status in ('todo', 'in_progress', 'in_review', 'blocked') + and (task.status <> 'blocked' or exists (select 1 from ${issueRecoveryActions} hold + where hold.company_id = task.company_id and hold.source_issue_id = task.id and hold.kind = 'active_run_watchdog' + and hold.cause = ${nativeRunFinalizations.failureCode} and hold.evidence->>'runId' = ${heartbeatRuns.id}::text + and hold.evidence#>>'{nativeFailureBlock,runId}' = ${heartbeatRuns.id}::text + and hold.evidence#>'{nativeFailureBlock,statusVersion}' = to_jsonb(task.status_version))))`, + ); +} + +async function preClaimFailureEvidence(db: ReplacementDb, run: typeof heartbeatRuns.$inferSelect) { + const [evidence] = await db.select({ id: heartbeatRuns.id }).from(heartbeatRuns) + .innerJoin(nativeRunFinalizations, eq(nativeRunFinalizations.runId, heartbeatRuns.id)).where(and( + eq(heartbeatRuns.companyId, run.companyId), eq(heartbeatRuns.id, run.id), preClaimFailurePending(), + )).limit(1); + return Boolean(evidence); +} + /** A saved remote source is proved by its current ledger, never a controller path. */ async function remoteWorkspaceReplacementEvidence( db: ReplacementDb, @@ -178,7 +239,7 @@ export function nativeReplacementPending() { or(inArray( nativeRunFinalizations.failureCode, ["native_provider_terminal_failed", "native_session_cleanup_quarantined", "provider_transport_failed"], - ), historicalListenerGuardPending()), + ), historicalListenerGuardPending(), preClaimFailurePending()), isNull(nativeRunFinalizations.resultId), sql`coalesce(${nativeRunFinalizations.failureDetail}->>'successorRunId', '') = ''`, sql`coalesce(${nativeRunFinalizations.failureDetail}->>'replacementDenied', '') = ''`, @@ -239,6 +300,8 @@ async function discoverNativeReplacements(db: Db, after: string | undefined, lim } } } + if (row.coordinator.failureCode === NATIVE_PRE_CLAIM_FAILURE_CODE) + ready = await preClaimContinuationReady(db, row.run, row.coordinator.issueId); if (ready) candidates.push(row); if (candidates.length >= limit || scanned >= REPLACEMENT_SCAN_LIMIT) return { candidates, scanned, after }; } @@ -246,6 +309,18 @@ async function discoverNativeReplacements(db: Db, after: string | undefined, lim } } +/** Saved files and a stopped sandbox, or a run that had no workspace to save. */ +async function preClaimContinuationReady(db: Db, run: typeof heartbeatRuns.$inferSelect, issueId: string) { + const leases = await db.select().from(environmentLeases).where(and( + eq(environmentLeases.companyId, run.companyId), eq(environmentLeases.heartbeatRunId, run.id), + )); + if (leases.some(lease => lease.releasedAt === null)) return false; + const evidence = await remoteWorkspaceReplacementEvidence(db, run, issueId, leases); + const workspaceReady = evidence ? evidence.workspacePreserved && evidence.providerStopped + : run.runnerProfileJson?.nativeWorkspaceSync == null && nativeRunHasNoWorkspaceAccess(run); + return workspaceReady && await replacementHistoryReady(db, run, issueId, true) === true; +} + function settleReplacementDiscovery(state: ReturnType, after: string | undefined, generation: number) { const rescan = state.rescan || generation !== state.generation; state.after = after; state.complete = !after && !rescan; @@ -309,11 +384,13 @@ export async function reconcileSafeNativeReplacements( if (leases.some((lease) => lease.releasedAt === null)) continue; const historicalListener = coordinator.failureCode === "native_session_interrupted"; if (historicalListener && !await historicalListenerGuardEvidence(db, run)) continue; - const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener + const preClaim = coordinator.failureCode === NATIVE_PRE_CLAIM_FAILURE_CODE; + if (preClaim && !await preClaimFailureEvidence(db, run)) continue; + const stoppedSession = coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener && !preClaim ? await options.verifyStoppedSession?.(run) ?? null : null; // A transport label alone is not evidence. Keep inspecting these candidates // as process cleanup and the final transcript become durable. - if (coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener && !stoppedSession) continue; + if (coordinator.failureCode !== "native_provider_terminal_failed" && !historicalListener && !preClaim && !stoppedSession) continue; const invocations = await db .select() .from(toolInvocations) @@ -407,12 +484,17 @@ export async function reconcileSafeNativeReplacements( const historyComplete = history === true; const remoteWorkspace = await remoteWorkspaceReplacementEvidence(db, run, coordinator.issueId, leases); if (historicalListener && (!remoteWorkspace?.workspacePreserved || !remoteWorkspace.providerStopped || invocations.length)) continue; + const noWorkspaceToSave = preClaim && !remoteWorkspace + && run.runnerProfileJson?.nativeWorkspaceSync == null && nativeRunHasNoWorkspaceAccess(run); + if (preClaim && (invocations.length || !noWorkspaceToSave + && (!remoteWorkspace?.workspacePreserved || !remoteWorkspace.providerStopped))) continue; const decision = decideNativeReplacement({ failedSession: true, // A facade transport failure can conceal an authoritative protocol // rejection. A stopped process and read receipts do not resolve that. + // Before its first claim the run could not dispatch provider work. failureMeaningKnown: - Boolean(stoppedSession) || (typeof coordinator.failureDetail?.originalFailureCode === "string" && + preClaim || Boolean(stoppedSession) || (typeof coordinator.failureDetail?.originalFailureCode === "string" && ![ "notification_transport_failed", "provider_turn_failed", @@ -424,15 +506,15 @@ export async function reconcileSafeNativeReplacements( !processAlive(run.processPid) && !processAlive(run.processGroupId ? -run.processGroupId : null) && (remoteWorkspace?.providerStopped ?? true), - workspacePreserved: remoteWorkspace ? remoteWorkspace.workspacePreserved : + workspacePreserved: noWorkspaceToSave || (remoteWorkspace ? remoteWorkspace.workspacePreserved : typeof workspace.cwd === "string" && (await stat(workspace.cwd).then( (s) => s.isDirectory(), () => false, - )), + ))), historyComplete, effectInventoryComplete: - Boolean(stoppedSession) || (record(run.runnerProfileJson).recoveryEventInventoryVersion === 1 && + preClaim || Boolean(stoppedSession) || (record(run.runnerProfileJson).recoveryEventInventoryVersion === 1 && record(execution.provider).kind === "codex"), attempts: coordinator.attempt, invocations, @@ -541,7 +623,7 @@ export async function reconcileSafeNativeReplacements( ) return false; let historicalRun: typeof heartbeatRuns.$inferSelect | undefined; - if (historicalListener) { + if (historicalListener || preClaim) { const [currentRun] = await tx.select().from(heartbeatRuns).where(and( eq(heartbeatRuns.id, run.id), eq(heartbeatRuns.companyId, run.companyId), )).for("update"); @@ -569,6 +651,7 @@ export async function reconcileSafeNativeReplacements( sql`${issueRecoveryActions.evidence}->'automaticRecovery'->>'replay' = 'blocked'`, )).for("update") : []; if (historicalListener && (!historicalRun || !await historicalListenerGuardEvidence(tx, historicalRun))) return false; + if (preClaim && (!historicalRun || !await preClaimFailureEvidence(tx, historicalRun))) return false; if (remoteWorkspace) { const currentWorkspace = await remoteWorkspaceReplacementEvidence(tx, run, coordinator.issueId, leases, true); if (!currentWorkspace?.workspacePreserved || !currentWorkspace.providerStopped) return false; @@ -615,6 +698,8 @@ export async function reconcileSafeNativeReplacements( forceFreshSession: true, recoveryIncidentRootRunId: record(run.contextSnapshot).recoveryIncidentRootRunId ?? run.id, + // Bounds pre-claim continuation to one automatic successor per chain. + ...(preClaim ? { nativePreClaimContinuationOf: run.id } : {}), }; const [wake] = await tx .insert(agentWakeupRequests) @@ -645,7 +730,7 @@ export async function reconcileSafeNativeReplacements( contextSnapshot: context, retryOfRunId: run.id, scheduledRetryAt: dueAt, - scheduledRetryAttempt: current.attempt, + scheduledRetryAttempt: preClaim ? 1 : current.attempt, scheduledRetryReason: NATIVE_SAFE_REPLACEMENT_REASON, }); options.failpoint?.("successor_inserted"); @@ -659,6 +744,7 @@ export async function reconcileSafeNativeReplacements( failureDetail: { ...current.failureDetail, ...(stoppedSession ? { stoppedTextTurn: stoppedSession.evidence } : {}), + ...(preClaim ? { preClaimContinuation: true } : {}), successorRunId, nextAction: "Continue in the linked fresh provider session after the retry delay.", @@ -699,7 +785,9 @@ export async function reconcileSafeNativeReplacements( status: "resolved", outcome: "handed_back", resolvedAt: now, updatedAt: now, wakePolicy: null, monitorPolicy: null, nextAction: `Safe continuation is scheduled in run ${successorRunId}.`, - resolutionNote: "Verified process termination and action receipts allow a fresh session to continue.", + resolutionNote: preClaim + ? "The run stopped before it started any provider work, and its saved files are recovered. A fresh session continues the task." + : "Verified process termination and action receipts allow a fresh session to continue.", evidence: { ...hold.evidence, verifiedReplacement: { successorRunId, recordedAt: now.toISOString() }, ...(hold.evidence.automaticRecovery ? { automaticRecovery: { ...record(hold.evidence.automaticRecovery), replay: "verified_safe_replacement", successorRunId,