From 6a7625ecb28f5a5b405a264da9b935c55e49ee95 Mon Sep 17 00:00:00 2001 From: Dotta Date: Fri, 30 Oct 2026 12:29:12 -0500 Subject: [PATCH] Recover retained lifecycle receipts and defer live-owner admission safely Co-Authored-By: Paperclip --- doc/execution-semantics.md | 10 +++ ...026-10-30-native-controller-checkpoints.md | 54 +++++++++++++-- .../remote-warm-controller.test-worker.ts | 3 + server/src/services/heartbeat.ts | 7 +- .../src/services/heartbeat/recovery.test.ts | 7 +- server/src/services/heartbeat/recovery.ts | 3 +- .../heartbeat/runtime-execution.test.ts | 20 ++++-- .../services/heartbeat/runtime-execution.ts | 5 ++ ...ve-controller-recovery.integration.test.ts | 52 +++++++++++++- .../native-controller-recovery.ts | 65 ++++++++++++++++-- .../native-runtime/native-restart-recovery.ts | 6 +- ...unner-restart-recovery.integration.test.ts | 35 ++++++++-- .../native-session-executor.test.ts | 48 +++++++++++-- .../native-runtime/native-session-executor.ts | 68 +++++++++++-------- 14 files changed, 321 insertions(+), 62 deletions(-) diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index 884c7ed7fb0..dc516274f72 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -2660,6 +2660,16 @@ mixed evidence holds the continuation; it does not authorize fresh execution. The 32-candidate boundary and newer-progress barriers remain in force. Exhausted history and current-run progress cannot establish fresh-bootstrap proof. +If admission finds a live retained controller owner, it preserves the unstarted +business run and records the owner's actual deadline through the existing native +resumption path. This does not consume a provider attempt or stop the retained +sandbox. A locked run/coordinator check rejects Stop, changed ownership or any +provider progress. Startup restores even an overdue wait to the same scheduler; +ordinary resumption rechecks current permissions and task authority. No idle +scan is added. Per-turn continuation reconstructs a missing suspended-state +receipt from the admitted sandbox or prepared-task owner, including after a +previous read failed with only a PRP projection written. + The version-one metadata contains a digest of the exact redacted stored snapshot, configuration digests and the PRP checkpoint binding. It is bounded to 4 KiB, excluded from public run projections, and saved by `checkpointSession` in its diff --git a/doc/plans/2026-10-30-native-controller-checkpoints.md b/doc/plans/2026-10-30-native-controller-checkpoints.md index 049f83f1855..8ba19ba1c61 100644 --- a/doc/plans/2026-10-30-native-controller-checkpoints.md +++ b/doc/plans/2026-10-30-native-controller-checkpoints.md @@ -383,11 +383,11 @@ SQL/transactions/rows during the text interval and zero idle SQL. The long sampl included normal lease renewals. Its cluster-wide WAL values include simultaneous test migrations and are diagnostic only, not a load acceptance measurement. -Workspace typecheck, build, and map structural checks passed. The full repository -test command and additional warm compatibility cases are running. The paired +Workspace typecheck, build, and map structural checks passed on the earlier +candidate; the later check dispositions are recorded below. The paired 300-controller/10,000-task workload now includes actual snapshot persistence. -Its initial host pilot missed throughput gates; an isolated Linux client/database -pilot is underway. Database cost acceptance, distinct-release application startup, +Its initial host pilot missed throughput gates; the subsequent isolated Linux +measurements are recorded below. Database cost acceptance, distinct-release application startup, compatible rollback, and native Daytona qualification remain open. The published Daytona qualification image was independently resolved from successful workflow 40709784515 to `sha256:b38ed051d00fed409092d354eea9ae1b01fb232bcb7d5a187ed763382dda2d0b` (source `8fe5d5defcdfb55a1b9049a801cfa5498629cebc`). The exact Linux Runner was @@ -413,8 +413,8 @@ dependency. No database-load acceptance claim has been made from failed attempts There is no Paperclip issue bound to this local chat; this versioned source plan is the handoff, rather than an upload to an unrelated company or issue. -Next action: complete the remaining native qualification and compatibility cells, -retain reproducible measurements, and prepare the reviewed PR. Full-home application +The remaining native qualification and compatibility cells require reproducible +measurements and a reviewed PR. Full-home application acceptance remains dependent on workspace/backup durability; no controller-only test can waive that gate. @@ -463,7 +463,7 @@ and relative gates and marks unequal-operation comparisons ineligible. Remaining critical path: finish the protected Daytona attempt and cleanup; complete full repository checks on the final rebased revision; finish the quiet steady/burst/replacement load gates and explicit lock-wait latency measurement; -qualify distinct installed A/B/A artifacts; then resolve review and prepare the +qualify distinct installed A/B/A artifacts; then resolve review on the draft PR. Current server startup smoke and three-controller component tests are useful layer-specific evidence, not substitutes for those remaining gates. @@ -475,3 +475,43 @@ checks will run on hosted Linux for the rebased candidate. The broad E2E support rerun with two workers was also stopped before rebasing; focused support checks remain the completed evidence. A separate `server-config.test.ts` diagnostic confirmed the original restricted run's `listen EPERM` cause. + + +### Rebased review and qualification follow-up + +Draft implementation: https://github.com/paperclipai/paperclip/pull/18343. +At rebased head `3aa90fbf0b5553ffa66a9cdc0fb153f348b56d75`, all 66 focused +persistence/discovery cases and full `pnpm -r typecheck` passed. The PR's general +server shard found 190 executor unit failures where synthetic IDs and a partial +Db double reached the new real discovery boundary. That fixture boundary is +being corrected; no production validation is relaxed. Greptile scored 3/5 and +identified missing prepared-task lifecycle reconstruction and retry after a +failed lifecycle read. Corrections and regression cases are under verification. +A separately identified retained-owner admission wait now uses the existing +same-run retry record/deadline, with unstarted-run and current-authority fences. + +Protected Daytona workflow `41094539946` failed before any test cell ran: the +image publisher exhausted disk (`No space left on device`, exit 102), including +its cleanup/log writes. It is not live recovery evidence, and repeating the +same build without a capacity fix is not justified. The default-branch workflow +has no per-run disk-capacity or upgrade-manifest override. CI infrastructure owns +that dependency. The invalid local credential attempt remains unresolved evidence. + +Local review-check attempts also encountered exhausted macOS Postgres semaphores; +only these owned test processes were stopped and moved to dedicated Docker +Postgres fixtures. One additional direct-read retry scenario timed out during +its source-controller fixture startup before a Runner appeared. Fixture import +readiness is now measured separately from the unchanged 60-second admission +limit. These failed/incomplete attempts remain distinct from subsequent passes. + +Review-fix verification: all 1,066 cases across native executor, durable +discovery, startup recovery and heartbeat runtime execution passed against owned +Docker Postgres. After the final database-clock and operator-Stop fences, all +nine durable discovery/import/wait tests passed and server typecheck passed. +The direct per-turn failed-read regression passed through normal admission with +three real controller processes: the failed read leaves the reconstructed PRP +projection, the next admission reconstructs the missing lifecycle receipt, and +the next replacement still continues the same provider history. Prepared-task +resumed and failed-read/retry cases also pass for Codex and Claude fixtures. +These are scripted-provider component results with remote workspace storage +available, not live Daytona or distinct-release acceptance. diff --git a/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts b/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts index c7a3301ceee..3e952c66126 100644 --- a/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts +++ b/server/src/__tests__/helpers/remote-warm-controller.test-worker.ts @@ -2,6 +2,9 @@ import { createDb } from "@paperclipai/db"; import { executePaperclipNativeSession } from "../../services/native-runtime/native-session-executor.js"; import { dockerCommandRunner } from "./docker-command-runner.js"; +// Separate TypeScript fixture import time from actual controller admission. +process.send?.({ ready: true }); + process.once("message", async (input: { connectionString: string; container: string; diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index 9894611d2f5..984186fe408 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -1,3 +1,4 @@ +import { NATIVE_RETAINED_OWNER_WAIT } from "./native-runtime/native-controller-recovery.js"; import { externalConversationSettlementWork } from "./external-conversation-settlement-work.js"; import { nativeRecoveryNoticeWork } from "./native-runtime/native-recovery-notice-work.js"; import { nativeRunProjectionWork } from "./native-runtime/native-run-projection-work.js"; @@ -3574,7 +3575,8 @@ export function heartbeatService( err.original instanceof Error ? err.original.message : String(err.original ?? ""); - const retryReasonCode = /native_finalization_missing/i.test( + const waitingForRetainedOwner = retryMessage === NATIVE_RETAINED_OWNER_WAIT; + const retryReasonCode = waitingForRetainedOwner ? NATIVE_RETAINED_OWNER_WAIT : /native_finalization_missing/i.test( retryMessage, ) ? "semantic_result_missing" @@ -3594,7 +3596,8 @@ export function heartbeatService( stream: "system", level: "warn", message: - retryReasonCode === "semantic_result_missing" + waitingForRetainedOwner ? "native continuation waiting for the retained controller lease" + : retryReasonCode === "semantic_result_missing" ? "provider turn completed without a semantic result; same-run disposition recovery persisted" : "native session transport interrupted; same-run resume persisted", payload: { diff --git a/server/src/services/heartbeat/recovery.test.ts b/server/src/services/heartbeat/recovery.test.ts index b6bdcfe6653..3c4834976ef 100644 --- a/server/src/services/heartbeat/recovery.test.ts +++ b/server/src/services/heartbeat/recovery.test.ts @@ -857,11 +857,12 @@ describePostgres("heartbeat recovery module ownership", () => { await Promise.all([...deps.activeRunExecutionPromises]); }); - it("rearms the persisted native retry deadline through the service callback", async () => { + it.each(["future", "overdue-owner"])("rearms the persisted native retry deadline through the service callback: %s", async scenario => { const run = await seedRun(); - const nextAttemptAt = new Date(Date.now() + 60_000); + const nextAttemptAt = new Date(Date.now() + (scenario === "future" ? 60_000 : -1_000)); await db.update(heartbeatRuns).set({ runtimeMode: "native", nativeIssueId: run.issueId, status: "failed" }).where(eq(heartbeatRuns.id, run.id)); - await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "retryable_failure", nextAttemptAt }); + await db.insert(nativeRunFinalizations).values({ runId: run.id, companyId: run.companyId, issueId: run.issueId!, phase: "retryable_failure", nextAttemptAt, + failureCode: scenario === "overdue-owner" ? "native_controller_recovery_waiting_for_owner" : null }); const deps = callbacks(db); const result = await createHeartbeatRecovery(db, deps).recoverNativeRunsAfterRestart(); expect(result.scheduledRetryRunIds).toEqual([run.id]); diff --git a/server/src/services/heartbeat/recovery.ts b/server/src/services/heartbeat/recovery.ts index b51bc5f53d4..6bbf4273767 100644 --- a/server/src/services/heartbeat/recovery.ts +++ b/server/src/services/heartbeat/recovery.ts @@ -1,3 +1,4 @@ +import { NATIVE_RETAINED_OWNER_WAIT } from "../native-runtime/native-controller-recovery.js"; import { nativeEnvironmentSettlementDisposition, nativeEnvironmentSettlementWork } from "../native-runtime/native-environment-settlement-work.js"; import { nativeWorkspaceFinalizationWork } from "../native-runtime/native-workspace-finalization-work.js"; import { attemptNativeWorkspaceFinalization } from "../native-runtime/native-workspace-finalizer.js"; @@ -953,7 +954,7 @@ export function createHeartbeatRecovery(db: Db, dependencies: HeartbeatRecoveryD inArray(heartbeatRuns.status, ["running", "failed"]), isNull(nativeRunFinalizations.resultId), eq(nativeRunFinalizations.phase, "retryable_failure"), - gt(nativeRunFinalizations.nextAttemptAt, now), + or(gt(nativeRunFinalizations.nextAttemptAt, now), eq(nativeRunFinalizations.failureCode, NATIVE_RETAINED_OWNER_WAIT)), ), ); for (const scheduled of scheduledNativeRetries) { diff --git a/server/src/services/heartbeat/runtime-execution.test.ts b/server/src/services/heartbeat/runtime-execution.test.ts index f40166ebd45..d8b65929c6a 100644 --- a/server/src/services/heartbeat/runtime-execution.test.ts +++ b/server/src/services/heartbeat/runtime-execution.test.ts @@ -1,3 +1,4 @@ +import { NativeControllerRecoveryWaitingError } from "../native-runtime/native-controller-recovery.js"; import { randomUUID } from "node:crypto"; import { mkdtemp, rm } from "node:fs/promises"; import { tmpdir } from "node:os"; @@ -8,7 +9,7 @@ import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vites import { trace } from "@opentelemetry/api"; import { InMemorySpanExporter, SimpleSpanProcessor, TracerProvider } from "@opentelemetry/sdk-trace"; import { createRunPreparationTiming, measureRunPhase, measureWorkspaceSyncPhase, withRunExecutionTiming } from "../execution-preparation-timing.js"; -import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "../../__tests__/helpers/embedded-postgres.js"; +import { recoveryPostgresSupport as support, startRecoveryPostgres } from "../../__tests__/helpers/recovery-postgres.js"; import * as pluginWorkspaces from "../plugin-workspaces.js"; import * as legacyRecovery from "../legacy-execution-recovery.js"; import { createAgentIdentityRedactor } from "../agent-identity-redaction.js"; @@ -41,15 +42,14 @@ vi.mock("./run-preparation.js", async original => ({ revokeHeartbeatRunGatewayTokens: vi.fn(async () => {}), })); -const support = await getEmbeddedPostgresTestSupport(); describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () => { - let database: Awaited>; + let database: Awaited>; let db: Db; let home: string; beforeAll(async () => { home = await mkdtemp(join(tmpdir(), "paperclip-runtime-execution-")); vi.stubEnv("PAPERCLIP_HOME", home); - database = await startEmbeddedPostgresTestDatabase("paperclip-runtime-execution"); + database = await startRecoveryPostgres("paperclip-runtime-execution"); db = createDb(database.connectionString); await db.execute(sql`set client_min_messages = warning`); }, EMBEDDED_POSTGRES_TEST_TIMEOUT_MS); @@ -249,6 +249,18 @@ describe.skipIf(!support.supported)("heartbeat runtime execution boundary", () = expectCleanup(input); }); + it("preserves the retained sandbox while native admission waits for its current owner", async () => { + const { input } = await fixture(true); + const error = new NativeControllerRecoveryWaitingError(new Date(Date.now() + 60_000)); + vi.mocked(executePaperclipNativeSession).mockRejectedValueOnce(error); + await expect(executeHeartbeatRuntime(db, input)).rejects.toMatchObject({ + name: "NativeSessionResumeScheduledError", original: error, + }); + expect(input.effects.onNativeSessionResumeScheduled).toHaveBeenCalledOnce(); + expect(input.effects.onProviderResourceDisposition).not.toHaveBeenCalledWith("stop_and_retain"); + expect(await finalizations()).toHaveLength(0); + }); + it("records native restoration inside the workspace finalization operation", async () => { const { input, run } = await fixture(true); const endpoint = process.env.OTEL_EXPORTER_OTLP_ENDPOINT; diff --git a/server/src/services/heartbeat/runtime-execution.ts b/server/src/services/heartbeat/runtime-execution.ts index 00430730ba3..69edcfdbee6 100644 --- a/server/src/services/heartbeat/runtime-execution.ts +++ b/server/src/services/heartbeat/runtime-execution.ts @@ -1,3 +1,4 @@ +import { NativeControllerRecoveryWaitingError } from "../native-runtime/native-controller-recovery.js"; import { DELIVERY_QUEUES } from "../delivery-work-notifications.js"; import { writeExecutionControl } from "../execution-control-notifications.js"; import { inspectPluginWorkspaceForFinalization } from "../native-runtime/native-plugin-workspace-finalization.js"; @@ -1013,6 +1014,10 @@ export async function executeHeartbeatRuntime(db: Db, input: HeartbeatRuntimeExe preserveRunGatewayTokens = true; throw adapterErr; } + if (adapterErr instanceof NativeControllerRecoveryWaitingError) { + input.effects.onNativeSessionResumeScheduled(); + throw new NativeSessionResumeScheduledError(adapterErr); + } if (adapterErr instanceof NativeRunnerOwnershipUnverifiedError) { input.effects.onNativeOwnershipHeld(); throw adapterErr; diff --git a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts index d24a299a221..2accfe2a6ca 100644 --- a/server/src/services/native-runtime/native-controller-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-controller-recovery.integration.test.ts @@ -10,7 +10,9 @@ import { startRecoveryPostgres, recoveryPostgresSupport } from "../../__tests__/ import { createIsolatedTestHome } from "../../__tests__/helpers/isolated-test-home.js"; import { prepareNativeHeartbeatRun } from "./prepare-native-run.js"; import { nativeRunnerCheckpointWriter } from "./native-runner-checkpoint.js"; -import { createNativeControllerRecovery, importNativeControllerRecovery, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { createNativeControllerRecovery, deferNativeControllerRecovery, importNativeControllerRecovery, NATIVE_RETAINED_OWNER_WAIT, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { claimNativeSessionResumptions } from "./native-finalization-reconciler.js"; +import { claimNativeRestartRecoveries } from "./native-restart-recovery.js"; import { readNativeControllerRecoveryInventory } from "./native-controller-recovery-inventory.js"; describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable discovery and import", () => { @@ -102,6 +104,54 @@ describe.skipIf(!recoveryPostgresSupport.supported)("native controller durable d } }); + it("keeps an unstarted follow-up on the same run until the retained owner's deadline", async () => { + const fixture = await seed(), runId = randomUUID(); + const scope = { ...fixture.scope, runId }; + const deadline = new Date(Date.now() + 60_000); + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: scope.issueId, + nativeSessionId: scope.normalizedSessionId, status: "running", runtimeMode: "native", + // The admission boundary has already stored its input; the deferral + // and resumption owners never execute this transport-independent fixture. + runnerProfileJson: { nativeExecutionInput: { binding: scope } } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, scope.issueId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: scope.issueId, runId, phase: "observed" }); + await deferNativeControllerRecovery(db, { ...scope, retryAt: deadline }); + const read = () => db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)); + expect((await read())[0]).toMatchObject({ phase: "retryable_failure", attempt: 0, leaseOwner: null, + nextAttemptAt: deadline, failureCode: NATIVE_RETAINED_OWNER_WAIT }); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId] })).toEqual([]); + // Even after expiry, restart adoption cannot reinterpret this pre-provider + // wait as a crashed provider. The ordinary resumption owner consumes it. + expect(await claimNativeRestartRecoveries({ db, runIds: [runId], restartKind: "hard", now: deadline })).toEqual([]); + const claims = await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId], now: deadline }); + expect(claims).toHaveLength(1); + const nextDeadline = new Date(deadline.getTime() + 60_000); + await deferNativeControllerRecovery(db, { ...scope, leaseOwner: claims[0].leaseOwner, retryAt: nextDeadline }); + expect((await read())[0]).toMatchObject({ attempt: 0, leaseOwner: null, nextAttemptAt: nextDeadline }); + await db.update(heartbeatRuns).set({ status: "cancelled" }).where(eq(heartbeatRuns.id, runId)); + expect(await claimNativeSessionResumptions({ db, runnerInstanceId: "successor", runIds: [runId], now: nextDeadline })).toEqual([]); + }); + + it.each(["live-owner", "expired-owner", "provider-attempt", "stopped", "operator-stop"])("does not publish an admission wait after authority changes: %s", async scenario => { + const fixture = await seed(), runId = randomUUID(); + const scope = { ...fixture.scope, runId }; + await db.insert(heartbeatRuns).values({ id: runId, companyId, agentId, nativeIssueId: scope.issueId, + status: scenario === "stopped" ? "cancelled" : "running", runtimeMode: "native", + resultJson: scenario === "operator-stop" ? { cancellation: { source: "operator", expected: true, + initiator: { type: "user" }, reason: "Stop", recordedAt: new Date().toISOString() } } : null, + runnerProfileJson: { nativeExecutionInput: { binding: scope } } }); + await db.update(issues).set({ executionRunId: runId }).where(eq(issues.id, scope.issueId)); + await db.insert(nativeRunFinalizations).values({ companyId, issueId: scope.issueId, runId, phase: "observed", + attempt: scenario === "provider-attempt" ? 1 : 0, + leaseOwner: scenario.endsWith("-owner") ? "other-controller" : null, + leaseExpiresAt: scenario === "expired-owner" ? new Date(Date.now() - 1_000) : null }); + await expect(deferNativeControllerRecovery(db, { ...scope, + leaseOwner: scenario === "expired-owner" ? "other-controller" : undefined, + retryAt: new Date(Date.now() + 60_000) })) + .rejects.toThrow("wait_authority_changed"); + expect((await db.select().from(nativeRunFinalizations).where(eq(nativeRunFinalizations.runId, runId)))[0].phase).toBe("observed"); + }); + it("reports bounded retained-pointer coverage without loading historical payloads", async () => { const fixture = await seed(); await importNativeControllerRecovery(db, { ...fixture.scope, expectedConfigDigest: compatibility.configDigest, supervisor: fixture.supervisor }); diff --git a/server/src/services/native-runtime/native-controller-recovery.ts b/server/src/services/native-runtime/native-controller-recovery.ts index cccb3aa27af..4c5fd4ac419 100644 --- a/server/src/services/native-runtime/native-controller-recovery.ts +++ b/server/src/services/native-runtime/native-controller-recovery.ts @@ -1,7 +1,9 @@ -import { and, eq, ne } from "drizzle-orm"; -import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, type Db } from "@paperclipai/db"; +import { and, eq, ne, sql } from "drizzle-orm"; +import { environmentLeases, heartbeatRuns, issues, nativeRunFinalizations, nativeRunnerCheckpoints, type Db } from "@paperclipai/db"; import type { AdapterExecutionTarget } from "@paperclipai/adapter-utils/execution-target"; import type { DurablePrpCheckpoint, PersistedNativeSession } from "../../vendor/paperclip-runner/index.js"; +import { notifyNativeFinalizationWork } from "../native-finalization-notifications.js"; +import { readRunCancellation } from "../run-cancellation.js"; import { nativeSha256 } from "./canonical.js"; import { findNativeSessionResumeRun, isNativeSessionId, isUnusedNativeSessionBootstrap, nativeSessionProviderEvidence } from "./native-session-resume.js"; import { loadNativeRunnerCheckpoint } from "./native-runner-checkpoint.js"; @@ -63,6 +65,61 @@ export type NativeControllerRecoveryOutcome = | { kind: "missing_evidence" | "unsupported" | "dependency_unavailable"; reason: string; cause?: unknown } | { kind: "no_prior_execution" }; +export const NATIVE_RETAINED_OWNER_WAIT = "native_controller_recovery_waiting_for_owner"; + +export class NativeControllerRecoveryWaitingError extends Error { + constructor(readonly retryAt: Date) { + super(NATIVE_RETAINED_OWNER_WAIT); + this.name = "NativeControllerRecoveryWaitingError"; + } +} + +/** Admission has not crossed the provider boundary. Retain this same business + * run on the existing retry owner; never turn an ownership wait into a failure + * or invent a new provider attempt. The usual resumption gates recheck Stop, + * permissions and task ownership when this deadline arrives. */ +export async function deferNativeControllerRecovery(db: Db, input: { + companyId: string; agentId: string; issueId: string; runId: string; + leaseOwner?: string; retryAt: Date; +}): Promise { + await db.transaction(async tx => { + const [task] = await tx.select({ executionRunId: issues.executionRunId }).from(issues) + .where(and(eq(issues.companyId, input.companyId), eq(issues.id, input.issueId))).for("update"); + const [owner] = await tx.select().from(nativeRunFinalizations).where(and( + eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId), + eq(nativeRunFinalizations.runId, input.runId))).for("update"); + const [run] = await tx.select().from(heartbeatRuns).where(and( + eq(heartbeatRuns.id, input.runId), eq(heartbeatRuns.companyId, input.companyId), + eq(heartbeatRuns.agentId, input.agentId), eq(heartbeatRuns.nativeIssueId, input.issueId))).for("update"); + const [authority] = await tx.select({ runId: nativeRunnerCheckpoints.runId }).from(nativeRunnerCheckpoints) + .where(and(eq(nativeRunnerCheckpoints.companyId, input.companyId), eq(nativeRunnerCheckpoints.runId, input.runId))).limit(1); + const progress = await nativeSessionProviderEvidence(tx as unknown as Db, [input.runId]); + const cancellation = record(run?.resultJson?.nativeCancellation); + const stop = readRunCancellation(run?.resultJson); + const [clock] = await tx.execute<{ now: Date }>(sql`select clock_timestamp() as now`); + if (!run || run.runtimeMode !== "native" || run.status !== "running" || task?.executionRunId !== input.runId + || !owner || owner.phase !== "observed" || owner.resultId || owner.attempt !== 0 + || owner.controllerGeneration !== 0 || owner.leaseOwner !== (input.leaseOwner ?? null) + || (input.leaseOwner && (!owner.leaseExpiresAt || owner.leaseExpiresAt <= new Date(clock.now))) + || run.processPid || run.processGroupId || run.processStartedAt || authority || progress.size + || !record(run.runnerProfileJson).nativeExecutionInput || run.resultJson?.startupCancellation + || (stop?.source === "operator" && stop.expected) + || (["run", "issue"].includes(String(cancellation.scope)) && ["pending", "acknowledged"].includes(String(cancellation.dispatchState)))) { + throw new Error("native_controller_recovery_wait_authority_changed"); + } + await notifyNativeFinalizationWork(tx, input); + await tx.update(nativeRunFinalizations).set({ + phase: "retryable_failure", leaseOwner: null, leaseExpiresAt: null, controlDeadlineAt: null, + nextAttemptAt: input.retryAt, recoveryState: "resuming_session", failureCode: NATIVE_RETAINED_OWNER_WAIT, + failureDetail: { originalFailureCode: NATIVE_RETAINED_OWNER_WAIT, + message: "Waiting for the retained native controller owner to release its lease.", + recoveryMode: record(run.runnerProfileJson).sessionCheckpoint ? "same_session_resume" : "bootstrap_retry", + providerSessionEstablished: false, providerEventsExist: false }, + updatedAt: new Date(), + }).where(eq(nativeRunFinalizations.runId, input.runId)); + }); +} + /** Bind a read of retained Runner lifecycle to the surviving sandbox. The * ordinary admission owner still checks current permissions before execution. */ export async function assertNativeRetainedSandboxBinding(db: Db, input: { @@ -118,10 +175,10 @@ export async function resolveNativeControllerRecovery(db: Db, input: { // The normal admission and remote untouched-session claim still execute. return { kind: "no_prior_execution" }; } - const [owner] = await db.select({ leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations) + const [owner] = await db.select({ leaseOwner: nativeRunFinalizations.leaseOwner, leaseExpiresAt: nativeRunFinalizations.leaseExpiresAt }).from(nativeRunFinalizations) .where(and(eq(nativeRunFinalizations.runId, source.id), eq(nativeRunFinalizations.companyId, input.companyId), eq(nativeRunFinalizations.issueId, input.issueId))); - if (owner?.leaseExpiresAt && owner.leaseExpiresAt.getTime() > Date.now()) { + if (owner?.leaseOwner && owner.leaseExpiresAt && owner.leaseExpiresAt.getTime() > Date.now()) { return { kind: "waiting_for_owner", deadline: owner.leaseExpiresAt }; } if (source.runtimeMode !== "native" || !["succeeded", "failed", "cancelled", "interrupted", "timed_out"].includes(source.status) diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index 1aec2ee27b2..4a42c652759 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -1,3 +1,4 @@ +import { NATIVE_RETAINED_OWNER_WAIT } from "./native-controller-recovery.js"; import { notifyNativeEnvironmentSettlement } from "../native-environment-settlement-notifications.js"; import { notifyExternalAgentSettlement } from "../../modules/external-agents/index.js"; import { notifyNativeStatusReassessment } from "../native-status-reassessment-notifications.js"; @@ -15,7 +16,7 @@ import { recordNativeLocalProcessStop, } from "../native-local-process-stop.js"; import { randomUUID } from "node:crypto"; -import { and, desc, eq, inArray, isNotNull, isNull, lte, or, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, isNotNull, isNull, lte, ne, or, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agents, @@ -515,6 +516,9 @@ export async function claimNativeRestartRecoveries(input: { eq(agents.adapterType, "paperclip_runner"), inArray(heartbeatRuns.status, ["running", "failed"]), isNull(nativeRunFinalizations.resultId), + // A pre-provider retained-owner wait belongs to normal same-run + // resumption, including when its deadline elapsed while we were down. + or(isNull(nativeRunFinalizations.failureCode), ne(nativeRunFinalizations.failureCode, NATIVE_RETAINED_OWNER_WAIT)), or( eq(nativeRunFinalizations.phase, "observed"), and( diff --git a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts index 8a16497fa93..9a0042b97f6 100644 --- a/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts +++ b/server/src/services/native-runtime/native-runner-restart-recovery.integration.test.ts @@ -322,10 +322,10 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( }; } - it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "legacy", "changed", "same-server", "diskless", "diskless-per-turn"])("preserves remote warm continuity across completed turns: %s", async (scenario) => { + it.skipIf(!process.env.PAPERCLIP_TEST_RUNNER_CONTAINER).each(["warm", "hard", "legacy", "changed", "same-server", "diskless", "diskless-per-turn", "diskless-per-turn-read-retry"])("preserves remote warm continuity across completed turns: %s", async (scenario) => { const hardRestart = !["warm", "same-server"].includes(scenario); const diskless = scenario.startsWith("diskless"); - const perTurn = scenario === "diskless-per-turn"; + const perTurn = scenario.startsWith("diskless-per-turn"); const container = process.env.PAPERCLIP_TEST_RUNNER_CONTAINER!; const tunnel = await createRunnerTestTlsTunnel({ port: Number(process.env.PAPERCLIP_TEST_RUNNER_PORT), key: process.env.PAPERCLIP_TEST_RUNNER_TLS_KEY!, cert: process.env.NODE_EXTRA_CA_CERTS! }); @@ -392,18 +392,22 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( const executeInController = (executionOptions: Parameters[0]) => new Promise<{ exitCode: number | null }>((resolveResult, reject) => { controller = spawn(process.execPath, ["--import", createRequire(import.meta.url).resolve("tsx"), resolve("server/src/__tests__/helpers/remote-warm-controller.test-worker.ts")], { stdio: ["ignore", "pipe", "pipe", "ipc"], env: process.env }); - const timer = setTimeout(() => { controller?.kill("SIGKILL"); reject(new Error("Warm controller startup exceeded 60 seconds")); }, 60_000); + let timer = setTimeout(() => { controller?.kill("SIGKILL"); reject(new Error("Controller fixture imports exceeded 120 seconds")); }, 120_000); controller.stdout?.on("data", data => logs.push(String(data))); controller.stderr?.on("data", data => logs.push(String(data))); controller.on("error", error => { clearTimeout(timer); reject(error); }); controller.on("exit", (code, signal) => { clearTimeout(timer); reject(new Error(`Warm controller exited: ${code}/${signal}: ${logs.join("").slice(-3_000)}`)); }); - controller.on("message", (message: { log?: { text: string }; result?: { exitCode: number | null }; error?: string }) => { + controller.on("message", (message: { ready?: boolean; log?: { text: string }; result?: { exitCode: number | null }; error?: string }) => { + if (message.ready) { + clearTimeout(timer); + timer = setTimeout(() => { controller?.kill("SIGKILL"); reject(new Error(`Warm controller admission exceeded 60 seconds: ${logs.join("").slice(-3_000)}`)); }, 60_000); + controller!.send?.({ connectionString: temporary.connectionString, container, ingressUrl: `wss://127.0.0.1:${tunnel.port}`, + options: { ...executionOptions, db: undefined, onLog: undefined, runnerExecutionTarget: { ...executionOptions.runnerExecutionTarget, runner: undefined } } }); + } if (message.log) logs.push(message.log.text); if (message.error) { clearTimeout(timer); reject(new Error(message.error)); } if (message.result) { clearTimeout(timer); resolveResult(message.result); } }); - controller.send?.({ connectionString: temporary.connectionString, container, ingressUrl: `wss://127.0.0.1:${tunnel.port}`, - options: { ...executionOptions, db: undefined, onLog: undefined, runnerExecutionTarget: { ...executionOptions.runnerExecutionTarget, runner: undefined } } }); }); try { const firstOptions = { ...options, execution, runnerExecutionTarget: target }; @@ -549,6 +553,23 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( ...(perTurn ? { sessionCheckpoint: admission.checkpoint } : {}) } }).where(eq(heartbeatRuns.id, row.id)); }; await admitContinuation(nextOptions.execution); + if (scenario === "diskless-per-turn-read-retry") { + let rejectedReads = 0; + await expect(executePaperclipNativeSession({ ...nextOptions, + runnerExecutionTarget: { ...nextOptions.runnerExecutionTarget, runner: { ...runner, + execute: async request => { + if (request.args?.some(arg => arg.includes("base64 <") && arg.includes("runner-state.json"))) { + rejectedReads++; + throw new Error("fixture_remote_state_timeout"); + } + return runner.execute(request); + }, + } }, + })).rejects.toThrow("fixture_remote_state_timeout"); + expect(rejectedReads).toBe(1); + // Retry with the partial PRP projection left by the failed read. No + // fixture writes a suspended receipt or changes the surviving Runner. + } const second = diskless ? await executeInController(nextOptions) : await executePaperclipNativeSession(nextOptions); expect(second.exitCode).toBe(0); const historyAfter = await runner.execute({ command: "sh", args: ["-c", `cat '${remoteCwd}'/.paperclip-runtime/paperclip-runner/sessions/*/filesystem/codex-home/fake-codex-state.json`] }); @@ -618,7 +639,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( await closeIdleWarmNativeSessionsForRestart(); await tunnel.close(); } - }, 180_000); + }, 360_000); async function persistRestartEvidence(input: { fixture: Awaited>; diff --git a/server/src/services/native-runtime/native-session-executor.test.ts b/server/src/services/native-runtime/native-session-executor.test.ts index b7775c11fda..44300bde734 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -1,6 +1,7 @@ import { onRetainedNativeSessionCleanupRelease } from "../retained-native-session-cleanup-notifications.js"; import { readIdleLocalWork, markIdleStartupComplete, markIdleIngressTracked } from "../idle-local-work.js"; import * as runnerCheckpoint from "./native-runner-checkpoint.js"; +import * as controllerRecovery from "./native-controller-recovery.js"; import { DurablePrpControlPlane } from "../../vendor/paperclip-runner/index.js"; import * as environmentTaskRuntime from "../environment-task-runtime.js"; import * as pluginTaskRuntime from "./plugin-runner-task.js"; @@ -410,6 +411,10 @@ import { import { onNativeRunnerEvidence } from "./native-runner-evidence-signal.js"; beforeEach(() => { + // These transport-wiring fixtures use a partial Db double and synthetic IDs. + // The real selector/bootstrap proof is exercised with Postgres by the durable + // discovery and diskless admission suites, not by this backend unit boundary. + vi.spyOn(controllerRecovery, "resolveNativeControllerRecovery").mockResolvedValue({ kind: "no_prior_execution" }); state.drainMaintenance.mockReset().mockResolvedValue(undefined); usageRecorder.capture.mockReset().mockImplementation(async receipt => receipt); usageRecorder.create.mockReset().mockResolvedValue({ capture: usageRecorder.capture, @@ -12063,6 +12068,27 @@ describe("runnerd provider runtime wiring", () => { ); }); + it.each(["warm", "per_turn"] as const)("defers a %s retained-owner wait before claiming a provider attempt", async mode => { + const fresh = scopedNativeExecution(`retained-owner-${mode}`); + fresh.session.lifecyclePolicy = mode === "warm" + ? { mode: "warm", idleTimeoutMs: 300_000 } : { mode: "per_turn" }; + const deadline = new Date(Date.now() + 60_000); + vi.mocked(controllerRecovery.resolveNativeControllerRecovery).mockResolvedValueOnce({ kind: "waiting_for_owner", deadline }); + const defer = vi.spyOn(controllerRecovery, "deferNativeControllerRecovery").mockResolvedValueOnce(); + const onExecutionLeaseAcquired = vi.fn(); + const updates: Array<{ table: unknown; values: Record }> = []; + const db = leaseDb(fresh, {}, {}, updates); + await expect(executePaperclipNativeSession({ db, execution: fresh, runnerInstanceId: "waiting-controller", + useRunnerd: true, onExecutionLeaseAcquired, + runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: "/workspace", + environmentId: "retained-environment", leaseId: "retained-lease" }, + })).rejects.toMatchObject({ name: "NativeControllerRecoveryWaitingError", retryAt: deadline }); + expect(defer).toHaveBeenCalledWith(db, expect.objectContaining({ ...fresh.binding, retryAt: deadline })); + expect(onExecutionLeaseAcquired).not.toHaveBeenCalled(); + expect(updates.some(update => update.table === nativeRunFinalizations && update.values.attempt !== undefined)).toBe(false); + expect(state.createBackend).not.toHaveBeenCalled(); + }); + it("rejects overlapping runs for the same runnerd provider session scope", async () => { const first = { ...execution, @@ -12448,7 +12474,7 @@ describe("runnerd provider runtime wiring", () => { expect(execute.mock.calls.some(([request]) => request.args.includes("paperclip-runner-launch") || request.args.includes("--build-metadata"))).toBe(false); }); - it.each(["claude", "codex"].flatMap(provider => (["per_turn", "warm"] as const).flatMap(policy => ["canonical", "local-receipt", "local-receipt-missing-terminal", "local-receipt-pending-command", "quarantined", "empty-runner", "empty-runner-quarantined", "nonempty-runner", "wrong-identity", "active", "foreign-scope", "no-stop-proof", "not-suspended", "pending-outbox", "pending-command", "missing-terminal", "ambiguous", "contradictory"].map(mode => [provider, policy, mode] as const))))( + it.each(["claude", "codex"].flatMap(provider => (["per_turn", "warm"] as const).flatMap(policy => ["canonical", "resumed", "resumed-timeout", "local-receipt", "local-receipt-missing-terminal", "local-receipt-pending-command", "quarantined", "empty-runner", "empty-runner-quarantined", "nonempty-runner", "wrong-identity", "active", "foreign-scope", "no-stop-proof", "not-suspended", "pending-outbox", "pending-command", "missing-terminal", "ambiguous", "contradictory"].filter(mode => policy === "per_turn" || !mode.startsWith("resumed")).map(mode => [provider, policy, mode] as const))))( "recovers a blocked %s %s plugin checkpoint on a replacement lease: %s", async (provider, policy, mode) => { const { permissionPolicy: _permissionPolicy, ...claudeProfile } = QUALIFIED_ACPX_PROFILES.claude; const first = { ...scopedNativeExecution(`plugin-${provider}-${policy}-${mode}`), @@ -12496,8 +12522,13 @@ describe("runnerd provider runtime wiring", () => { if (mode === "no-stop-proof") throw new Error("Plugin Runner retained session is not authorized"); return "old-task"; }); + let timedOut = false; const task = vi.fn(async (operation: any) => { environmentTaskOperationSchema.parse(operation); + if (mode === "resumed-timeout" && operation.kind === "read_state" && !timedOut) { + timedOut = true; + throw new Error("fixture_remote_state_timeout"); + } if (operation.kind === "read_state") return { kind: "state", taskId: "replacement-task", state: { ...durableRunnerState(identity, mode === "not-suspended" ? "ready" : "suspended"), outbox: mode === "pending-outbox" ? [{}] : [], pendingTerminalDelivery: null, @@ -12516,7 +12547,7 @@ describe("runnerd provider runtime wiring", () => { managedAiCredentialHome: home, managedAiCredentialIdentity: "revision", managedAiAccountIdentity: "account", runnerEnvironment: { ANTHROPIC_API_KEY: "synthetic-fixture" }, runnerExecutionTarget: { kind: "remote", transport: "sandbox", remoteCwd: ".", leaseId: "replacement-lease", runnerTask: task, - sandboxLeaseAcquisition: { outcome: "replacement", providerLeaseId: "replacement-task" } } as never }; + sandboxLeaseAcquisition: { outcome: mode.startsWith("resumed") ? "resumed" : "replacement", providerLeaseId: "replacement-task" } } as never }; if (throughExecutor) { const validated = validatePrpStructuredRunResult({ schema: "paperclip.run_result.v1", reportedWorkDisposition: "done", summary: "done", completionClaim: { contractRevision: next.completionContract.contract.revision, objectiveSatisfied: true, criteria: [], remainingWork: [] }, @@ -12526,12 +12557,19 @@ describe("runnerd provider runtime wiring", () => { normalizedSessionId: next.session.normalizedSessionId, providerSessionId: "original-provider-history", driverKind: next.session.driverKind, driverVersion: "1", nativeEventCount: 1, highestContiguousSourceSeq: 1, usage: null }); } + if (mode === "resumed-timeout") { + await expect(createRunnerdBackend(launch)).rejects.toThrow("fixture_remote_state_timeout"); + expect(existsSync(join(scoped, "plugin-runner-state.json"))).toBe(false); + expect(await readFile(join(scoped, "control-plane/control-plane-state.json"), "utf8")).toBe(journal); + } const backend = throughExecutor ? executePaperclipNativeSession({ ...launch, useRunnerd: true }) : createRunnerdBackend(launch); - if (["canonical", "local-receipt", "quarantined", "empty-runner", "empty-runner-quarantined"].includes(mode)) { + if (["canonical", "resumed", "resumed-timeout", "local-receipt", "quarantined", "empty-runner", "empty-runner-quarantined"].includes(mode)) { await expect(backend).resolves.toBeDefined(); expect(state.createBackend).toHaveBeenCalledOnce(); - expect(owner).toHaveBeenCalledWith(expect.objectContaining({ resumeFrom: "old-task", priorIdentity: identity })); - expect(resume).toHaveBeenCalledWith(db, expect.objectContaining({ leaseId: "replacement-lease", priorIdentity: identity, inspectCheckpoint: true })); + expect(owner).toHaveBeenCalledWith(expect.objectContaining({ priorIdentity: identity, + ...(!mode.startsWith("resumed") ? { resumeFrom: "old-task" } : {}) })); + if (!mode.startsWith("resumed")) expect(resume).toHaveBeenCalledWith(db, expect.objectContaining({ leaseId: "replacement-lease", priorIdentity: identity, inspectCheckpoint: true })); + else expect(task).toHaveBeenCalledWith({ kind: "read_state", identity }); expect(JSON.parse(await readFile(join(scoped, "plugin-runner-state.json"), "utf8"))).toMatchObject({ ...identity, lifecycle: "suspended" }); expect(await readFile(join(scoped, "control-plane/control-plane-state.json"), "utf8")).toBe(journal); } else { diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 614dca7f0cb..9cfd4f07a88 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -179,7 +179,7 @@ import { nativeRunResults, } from "@paperclipai/db"; import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; -import { assertNativeRetainedSandboxBinding, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; +import { assertNativeRetainedSandboxBinding, deferNativeControllerRecovery, NativeControllerRecoveryWaitingError, resolveNativeControllerRecovery } from "./native-controller-recovery.js"; import { redactRunnerApiPayloads } from "./runner-api-log-redaction.js"; import { appendHeartbeatRunEvent } from "../heartbeat-run-events.js"; import { appendNativeCredentialUpdate } from "./native-credential-updates.js"; @@ -5755,32 +5755,42 @@ async function restorePerTurnRunnerdController(input: { const identity = readRunnerdDurableIdentity(root); const reference = identity ?? readRunnerdDurableIdentity(root, true); if (!reference) { - const restored = await restoreNativeControllerProjections(input.db, input.execution, root); - const target = input.runnerExecutionTarget; - // Prepared tasks use their existing read_state owner below. Direct sandbox - // sessions also keep lifecycle evidence on the execution host, not on B's disk. - if (restored && target.transport === "sandbox" && !target.runnerTask) { - const prior = readRunnerdDurableIdentity(root); - if (!prior || typeof prior.runId !== "string") throw new Error("native_controller_recovery_missing_evidence:restored_identity"); - if (!target.runner) throw new Error("native_controller_recovery_dependency_unavailable:runner_state_reader"); - await assertNativeRetainedSandboxBinding(input.db, { ...input.execution.binding, runId: prior.runId, target }); - const stateDirectory = posix.join(target.remoteCwd, ".paperclip-runtime", "paperclip-runner", "sessions", - createHash("sha256").update(nativeSessionKey(input.execution)).digest("hex"), "runner"); - const state = await readRemoteRunnerState({ runner: target.runner, stateDirectory }); - if (state.lifecycle !== "suspended" || !Array.isArray(state.outbox) || state.outbox.length !== 0 - || state.pendingTerminalDelivery != null || state.pendingProviderCleanup != null || state.warmTransition != null) { - throw new Error("native_controller_recovery_missing_evidence:runner_not_settled"); - } - savePluginRunnerCheckpoint(root, state, prior); - } - return; + if (!await restoreNativeControllerProjections(input.db, input.execution, root)) return; + } else { + if (!durableIdentityMatchesSession(reference, input.execution)) return; + if (identity?.runId === input.execution.binding.runId) return; + const restored = await restoreNativeRunnerCheckpoint(input.db, + { companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, runId: reference.runId }, + root, nativeSessionKey(input.execution)); + if (!restored && !identity) throw new Error("native_runner_checkpoint_missing"); } - if (!durableIdentityMatchesSession(reference, input.execution)) return; - if (identity?.runId === input.execution.binding.runId) return; - const restored = await restoreNativeRunnerCheckpoint(input.db, - { companyId: input.execution.binding.companyId, issueId: input.execution.binding.issueId, runId: reference.runId }, - root, nativeSessionKey(input.execution)); - if (!restored && !identity) throw new Error("native_runner_checkpoint_missing"); + const target = input.runnerExecutionTarget; + const prior = readRunnerdDurableIdentity(root); + // PRP and lifecycle evidence have different owners. A failed remote read can + // leave a valid PRP projection, so its presence must not suppress this retry. + // Never overwrite contradictory local Runner evidence with a remote receipt. + if (target.transport !== "sandbox" || target.sandboxLeaseAcquisition?.outcome !== "resumed" + || !durableIdentityMatchesSession(prior, input.execution) || prior.runId === input.execution.binding.runId + || runnerdAuthorityLifecycle(root, prior) !== "absent") return; + let state: Record; + if (target.runnerTask) { + // The admitted task owner binds this read to the surviving physical task; + // its suspended identity still belongs to the previous business turn. + const result = environmentTaskResultSchema.parse(await target.runnerTask({ kind: "read_state", identity: prior })); + if (result.kind !== "state") throw new Error("Plugin Runner checkpoint unavailable"); + state = result.state; + } else { + if (!target.runner) throw new Error("native_controller_recovery_dependency_unavailable:runner_state_reader"); + await assertNativeRetainedSandboxBinding(input.db, { ...input.execution.binding, runId: prior.runId, target }); + const stateDirectory = posix.join(target.remoteCwd, ".paperclip-runtime", "paperclip-runner", "sessions", + createHash("sha256").update(nativeSessionKey(input.execution)).digest("hex"), "runner"); + state = await readRemoteRunnerState({ runner: target.runner, stateDirectory }); + } + if (state.lifecycle !== "suspended" || !Array.isArray(state.outbox) || state.outbox.length !== 0 + || state.pendingTerminalDelivery != null || state.pendingProviderCleanup != null || state.warmTransition != null) { + throw new Error("native_controller_recovery_missing_evidence:runner_not_settled"); + } + savePluginRunnerCheckpoint(root, state, prior); } /** Materialize opaque controller state before ordinary warm admission reads it. @@ -5831,7 +5841,7 @@ async function restoreNativeControllerProjections(db: Db, execution: NativeExecu }); if (recovered.kind === "no_prior_execution") return false; if (recovered.kind === "waiting_for_owner") { - throw Object.assign(new Error("native_controller_recovery_waiting_for_owner"), { retryAt: recovered.deadline }); + throw new NativeControllerRecoveryWaitingError(recovered.deadline); } if (recovered.kind !== "recoverable") throw new Error(`native_controller_recovery_${recovered.kind}:${recovered.reason}`, { cause: recovered.cause }); recovered.checkpoint.restoreSessionDirectory(root); @@ -8047,6 +8057,10 @@ export async function executePaperclipNativeSession(input: { return await executePaperclipNativeSessionWithinScope(preparedInput); } catch (error) { executionFailure = error; + if (error instanceof NativeControllerRecoveryWaitingError) { + await deferNativeControllerRecovery(input.db, { ...input.execution.binding, + leaseOwner: input.leaseOwner, retryAt: error.retryAt }); + } const remoteRunnerRecovery = readRemoteRunnerRecoveryError(error); if (remoteRunnerRecovery) { await input.onEvent?.({