From 6596b09e373127e2f9283a29b77c18a0e564eeec Mon Sep 17 00:00:00 2001 From: Dotta Date: Thu, 10 Sep 2026 13:01:08 -0500 Subject: [PATCH] fix: verify and adopt native sandbox runners after app restart Keep remote recovery out of host PID classification. Verify the exact remote state and process before adoption, preserve the controller journal on unavailable evidence, and retain local restart behavior. Cover externally owned state across real controller restarts. Co-Authored-By: Paperclip --- doc/sandbox-work-folders.md | 13 ++ .../src/live/runnerd-codex-transport.ts | 2 +- server/src/services/heartbeat.ts | 9 +- .../native-restart-recovery.test.ts | 10 ++ .../native-runtime/native-restart-recovery.ts | 49 +++++- ...unner-restart-recovery.integration.test.ts | 35 ++-- .../native-session-executor.test.ts | 52 ++++++ .../native-runtime/native-session-executor.ts | 94 ++++++++-- .../remote-runner-recovery.test.ts | 127 ++++++++++++++ .../native-runtime/remote-runner-recovery.ts | 163 ++++++++++++++++++ 10 files changed, 518 insertions(+), 36 deletions(-) create mode 100644 server/src/services/native-runtime/remote-runner-recovery.test.ts create mode 100644 server/src/services/native-runtime/remote-runner-recovery.ts diff --git a/doc/sandbox-work-folders.md b/doc/sandbox-work-folders.md index e2124f19c88..4f90de6c938 100644 --- a/doc/sandbox-work-folders.md +++ b/doc/sandbox-work-folders.md @@ -138,6 +138,19 @@ The new scoped-shell startup setting is not added to an existing unscoped Codex session's protected launch arguments. Its durable provider profile remains unchanged during attachment. +For an active native sandbox run, the app keeps the control-plane journal while +the sandbox keeps the runner journal. An app restart must verify the exact +remote run, session, runner, and lease identity; a missing controller-side runner +file is not evidence of corruption. Recovery checks the remote process marker, +Linux process fingerprint, and command-line binding before adopting a live +runner. The runner must then authenticate to its existing durable PRP authority. +Adoption preserves the provider attempt and does not launch another runner or +replace its artifacts. A dead runner can restart only from verified suspended +state. Missing or conflicting evidence preserves the controller journal and +fails recovery. Host PID checks do not establish remote process ownership. +Remote runners have a five-minute reconnect grace period for app replacement; +liveness probes share one request in flight and run at most once per second. + Acceptance must resume representative pre-upgrade legacy and native tasks with committed, staged, unstaged, and untracked work, verify their original paths and usable continuation, and exercise their existing restore mechanism after a diff --git a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts index f26452c4896..09557e4de48 100644 --- a/packages/paperclip-runner/src/live/runnerd-codex-transport.ts +++ b/packages/paperclip-runner/src/live/runnerd-codex-transport.ts @@ -1030,7 +1030,7 @@ export interface CapabilityRunnerdCodexTransportOptions { /** Active-connection recovery budget. Omitted for the existing local mode. */ runnerReconnectGraceMs?: number; /** - * A verified local runner that outlived its controller. Adoption registers + * A verified runner that outlived its controller. Adoption registers * the durable authority and waits for this exact process to reconnect; it * never calls the process launcher while the process remains alive. */ diff --git a/server/src/services/heartbeat.ts b/server/src/services/heartbeat.ts index d6593a0c0e2..a9f779f5ca9 100644 --- a/server/src/services/heartbeat.ts +++ b/server/src/services/heartbeat.ts @@ -13203,6 +13203,7 @@ export function heartbeatService( const claims = dispositions.filter( (disposition): disposition is NativeRestartRecoveryClaim => disposition.kind === "reattach_existing_runner" || + disposition.kind === "reconcile_remote_runner" || disposition.kind === "resume_dead_runner" || disposition.kind === "bootstrap_incomplete", ); @@ -13211,6 +13212,7 @@ export function heartbeatService( if (run) { const isClaim = disposition.kind === "reattach_existing_runner" || + disposition.kind === "reconcile_remote_runner" || disposition.kind === "resume_dead_runner" || disposition.kind === "bootstrap_incomplete"; await appendRunEvent(run, { @@ -13218,7 +13220,9 @@ export function heartbeatService( stream: "system", level: disposition.kind === "blocked" ? "warn" : "info", message: - disposition.kind === "reattach_existing_runner" + disposition.kind === "reconcile_remote_runner" + ? "Verifying the remote native runner authority after server restart" + : disposition.kind === "reattach_existing_runner" ? "Recovering the existing native runner process after server restart" : disposition.kind === "resume_dead_runner" ? "Resuming the durable native provider session after runner process loss" @@ -18046,7 +18050,8 @@ export function heartbeatService( if ( runOptions.nativeLeaseOwner && run.runtimeMode === "native" && - runOptions.nativeRestartRecovery?.kind !== "reattach_existing_runner" + runOptions.nativeRestartRecovery?.kind !== "reattach_existing_runner" && + runOptions.nativeRestartRecovery?.kind !== "reconcile_remote_runner" ) { // A numeric PID or process-group ID is a liveness signal, never an // ownership capability: the OS may have recycled it after the service diff --git a/server/src/services/native-runtime/native-restart-recovery.test.ts b/server/src/services/native-runtime/native-restart-recovery.test.ts index f320d5def3f..50b4465fbff 100644 --- a/server/src/services/native-runtime/native-restart-recovery.test.ts +++ b/server/src/services/native-runtime/native-restart-recovery.test.ts @@ -11,9 +11,19 @@ describe("native restart recovery classification", () => { it("keeps controller-only recovery out of the provider retry budget", () => { expect(nextNativeProviderAttempt(2, "reattach_existing_runner")).toBe(2); expect(nextNativeProviderAttempt(2, "bootstrap_incomplete")).toBe(2); + expect(nextNativeProviderAttempt(2, "reconcile_remote_runner")).toBe(2); expect(nextNativeProviderAttempt(2, "resume_dead_runner")).toBe(3); }); + it.each([true, false])("does not use host process evidence for remote recovery (%s)", (hostAlive) => { + const evidence = { remote: true, runnerPidAlive: hostAlive, runnerGroupAlive: hostAlive, + processStartMatches: hostAlive, knownProviderProcessAlive: hostAlive, + hasCheckpoint: true, checkpointIdentityMatches: true, hasProviderEvidence: true }; + expect(classifyNativeRunnerRecoveryEvidence(evidence).claimKind).toBe("reconcile_remote_runner"); + expect(classifyNativeRunnerRecoveryEvidence({ ...evidence, checkpointIdentityMatches: false }).claimKind).toBeNull(); + expect(classifyNativeRunnerRecoveryEvidence({ ...evidence, hasProviderEvidence: false }).claimKind).toBeNull(); + }); + it.each([ { name: "reattaches an exact live runner", diff --git a/server/src/services/native-runtime/native-restart-recovery.ts b/server/src/services/native-runtime/native-restart-recovery.ts index cf97bfc56ca..bd95c9a6ca4 100644 --- a/server/src/services/native-runtime/native-restart-recovery.ts +++ b/server/src/services/native-runtime/native-restart-recovery.ts @@ -35,6 +35,15 @@ export type NativeRestartRecoveryClaim = startedAt: string; }; } + | { + kind: "reconcile_remote_runner"; + runId: string; + leaseOwner: string; + controllerGeneration: number; + providerAttempt: number; + restartKind: NativeRestartKind; + recoveryRequestId: string | null; + } | { kind: "resume_dead_runner"; runId: string; @@ -67,6 +76,7 @@ export function nextNativeProviderAttempt( recoveryKind?: NativeRestartRecoveryClaim["kind"], ): number { return recoveryKind === "reattach_existing_runner" || + recoveryKind === "reconcile_remote_runner" || recoveryKind === "bootstrap_incomplete" ? currentAttempt : currentAttempt + 1; @@ -266,6 +276,7 @@ export async function evaluateNativeProviderProcesses(input: { } export function classifyNativeRunnerRecoveryEvidence(input: { + remote?: boolean; runnerPidAlive: boolean; runnerGroupAlive: boolean; processStartMatches: boolean; @@ -278,6 +289,20 @@ export function classifyNativeRunnerRecoveryEvidence(input: { claimKind: NativeRestartRecoveryClaim["kind"] | null; reason: string; } { + if (input.remote) { + // Host PIDs say nothing about a sandbox or SSH process. Claim only the + // exact persisted run; the executor must inspect its remote authority + // before deciding whether to adopt a live runner or resume suspended work. + return input.hasCheckpoint && input.checkpointIdentityMatches === true && input.hasProviderEvidence + ? { + claimKind: "reconcile_remote_runner", + reason: "remote_authority_requires_verification", + } + : { + claimKind: null, + reason: "remote_recovery_checkpoint_unverified", + }; + } if (input.runnerPidAlive && input.processStartMatches) { return { claimKind: "reattach_existing_runner", @@ -357,7 +382,8 @@ function historyEntry(input: { stateRootAction: input.disposition === "reattach_existing_runner" ? "reopen_exact_root" - : input.disposition === "resume_dead_runner" + : input.disposition === "resume_dead_runner" || + input.disposition === "reconcile_remote_runner" ? "reuse_exact_root" : input.disposition === "bootstrap_incomplete" ? "quarantine_incomplete_root_then_bootstrap" @@ -546,8 +572,13 @@ export async function claimNativeRestartRecoveries(input: { } as const; } - const runnerPidAlive = processIsAlive(row.run.processPid); - const runnerGroupAlive = processGroupIsAlive(row.run.processGroupId); + const environment = row.run.contextSnapshot?.paperclipEnvironment; + const remote = environment !== null && + typeof environment === "object" && + !Array.isArray(environment) && + ["sandbox", "ssh"].includes(String((environment as Record).driver)); + const runnerPidAlive = !remote && processIsAlive(row.run.processPid); + const runnerGroupAlive = !remote && processGroupIsAlive(row.run.processGroupId); const observedRunnerStart = row.run.processPid && runnerPidAlive ? await observedProcessStart(row.run.processPid) @@ -671,14 +702,17 @@ export async function claimNativeRestartRecoveries(input: { } } const providerProcesses = await evaluateNativeProviderProcesses({ - identities: providerProcessIdentities.filter( + identities: (remote ? [] : providerProcessIdentities).filter( (identity) => identity.pid !== row.run.processPid, ), }); const hasProviderEvidence = hasCheckpointProviderIdentity || providerEvents.length > 0; - const classification = classifyNativeRunnerRecoveryEvidence({ + const classification = remote && (environment as Record).driver === "ssh" + ? { claimKind: null, reason: "remote_recovery_transport_unsupported" } + : classifyNativeRunnerRecoveryEvidence({ + remote, runnerPidAlive, runnerGroupAlive, processStartMatches: exactRunnerIdentity, @@ -765,7 +799,8 @@ export async function claimNativeRestartRecoveries(input: { controllerProcessStartedAt: controller.processStartedAt, controllerGeneration: generation, recoveryState: - claimKind === "reattach_existing_runner" + claimKind === "reattach_existing_runner" || + claimKind === "reconcile_remote_runner" ? "awaiting_runner_reattach" : claimKind === "resume_dead_runner" ? "resuming_session" @@ -808,7 +843,7 @@ export async function claimNativeRestartRecoveries(input: { errorCode: null, nativePhase: "observed", nativePhaseUpdatedAt: now, - ...(claimKind === "reattach_existing_runner" + ...(claimKind === "reattach_existing_runner" || claimKind === "reconcile_remote_runner" ? {} : { processPid: null, 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 629e83757f5..b49c96ac215 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 @@ -352,13 +352,17 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( .where(eq(nativeRunFinalizations.runId, input.fixture.runId)); } - realProcessIt("adopts one active runner across hot and hard controller restarts without duplicating steering", async () => { - const fixture = await seedRun("LIVE"); + realProcessIt.each([false, true])("adopts one active runner across hot and hard controller restarts without duplicating steering (external state: %s)", async (externalState) => { + const fixture = await seedRun(externalState ? "LIVE-REMOTE" : "LIVE"); const stateDirectory = resolve(runtimeRoot, fixture.runId); const baseOptions = transportOptions(fixture, stateDirectory); const options = { ...baseOptions, codexArgs: [...baseOptions.codexArgs, "--linger-after-turn-start"], + ...(externalState ? { + runnerStateDirectory: resolve(stateDirectory, "external-runner"), + readRunnerState: async () => JSON.parse(await readFile(resolve(stateDirectory, "external-runner", "runner-state.json"), "utf8")), + } : {}), }; const first = createRunnerdCodexTransport(options); let runnerPid: number | null = null; @@ -392,6 +396,17 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( providerPid: first.evidence().providerPid, providerSessionId: String(thread.id), }); + const processIdentity = { + pid: runnerPid, + processGroupId: first.evidence().runnerProcessGroupId, + startedAt: (await readProcessStartedAt(runnerPid))!, + }; + if (externalState) { + await fixture.db.update(heartbeatRuns).set({ + contextSnapshot: { paperclipEnvironment: { driver: "sandbox" } }, + }).where(eq(heartbeatRuns.id, fixture.runId)); + expect(existsSync(resolve(stateDirectory, "runner", "runner-state.json"))).toBe(false); + } await first.detachControllerForRestart(); const [claim] = await claimNativeRestartRecoveries({ @@ -403,13 +418,13 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( runIds: [fixture.runId], }); expect(claim).toMatchObject({ - kind: "reattach_existing_runner", + kind: externalState ? "reconcile_remote_runner" : "reattach_existing_runner", runId: fixture.runId, controllerGeneration: 2, providerAttempt: 0, - process: { pid: runnerPid }, + ...(externalState ? {} : { process: { pid: runnerPid } }), }); - if (!claim || claim.kind !== "reattach_existing_runner") { + if (!claim || (claim.kind !== "reattach_existing_runner" && claim.kind !== "reconcile_remote_runner")) { throw new Error("Expected live-runner recovery claim"); } @@ -421,7 +436,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( resumeDynamicTools: [], runnerProcessLauncher: duplicateLauncher, adoptExistingRunner: { - ...claim.process, + ...(claim.kind === "reattach_existing_runner" ? claim.process : processIdentity), isAlive: () => processAlive(runnerPid), }, }); @@ -460,13 +475,13 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( runIds: [fixture.runId], }); expect(hardClaim).toMatchObject({ - kind: "reattach_existing_runner", + kind: externalState ? "reconcile_remote_runner" : "reattach_existing_runner", runId: fixture.runId, controllerGeneration: 3, providerAttempt: 0, - process: { pid: runnerPid }, + ...(externalState ? {} : { process: { pid: runnerPid } }), }); - if (!hardClaim || hardClaim.kind !== "reattach_existing_runner") { + if (!hardClaim || (hardClaim.kind !== "reattach_existing_runner" && hardClaim.kind !== "reconcile_remote_runner")) { throw new Error("Expected second live-runner recovery claim"); } const secondDuplicateLauncher = vi.fn(() => { @@ -477,7 +492,7 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", ( resumeDynamicTools: [], runnerProcessLauncher: secondDuplicateLauncher, adoptExistingRunner: { - ...hardClaim.process, + ...(hardClaim.kind === "reattach_existing_runner" ? hardClaim.process : processIdentity), isAlive: () => processAlive(runnerPid), }, }); 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 63e8a02bbbb..e24058aa172 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -5560,6 +5560,58 @@ describe("runnerd provider runtime wiring", () => { } }); + it.each(["live", "suspended", "mismatched", "unavailable"] as const)("verifies remote restart authority without requiring a host runner copy (%s)", async (scenario) => { + const stateBase = await mkdtemp(join(tmpdir(), "paperclip-remote-restart-")); + const previous = process.env.PAPERCLIP_RUNNER_STATE_DIR; + process.env.PAPERCLIP_RUNNER_STATE_DIR = stateBase; + const identity = { runId: execution.binding.runId, normalizedSessionId: execution.session.normalizedSessionId, + runnerInstanceId: "remote-restart-runner", environmentLeaseId: "remote-restart-lease" }; + const execute = vi.fn(async (input: { command: string; args: string[] }) => { + expect(input.command).toBe("node"); + const request = JSON.parse(input.args[2]!); + if (scenario === "unavailable") throw new Error("provider unavailable"); + return { exitCode: 0, timedOut: false, stdout: JSON.stringify({ + identity: { ...identity, ...(scenario === "mismatched" ? { runId: "other-run" } : {}) }, + stateDirectory: request.stateDirectory, lifecycle: scenario === "suspended" ? "suspended" : "ready", + alive: scenario !== "suspended", process: { pid: 4321, startedAt: "2026-09-10T00:00:00.000Z", nonce: "ec0e1ae3-0614-44fc-a352-bb03d89134d7", startTicks: "13579" }, + }) }; + }); + const target = { kind: "remote", transport: "sandbox", remoteCwd: "/home/daytona/repos/project", providerKey: "daytona", runner: { execute } } as never; + try { + state.createBackend.mockClear(); state.createTransport.mockClear(); + await createRunnerdBackend({ db: leaseDb(execution), execution, runnerInstanceId: identity.runnerInstanceId }); + state.createBackend.mock.calls[0]![1].codexTransportFactory!(); + const root = state.createTransport.mock.calls[0]![0].stateDirectory!; + await mkdir(join(root, "control-plane"), { recursive: true }); + const controlPlanePath = join(root, "control-plane", "control-plane-state.json"); + const original = JSON.stringify(durableControlPlaneState(identity)); + await writeFile(controlPlanePath, original); + state.createBackend.mockClear(); state.createTransport.mockClear(); + const resumed = createRunnerdBackend({ db: leaseDb(execution), execution, runnerInstanceId: identity.runnerInstanceId, + runnerExecutionTarget: target, restartRecovery: { + kind: "reconcile_remote_runner", runId: execution.binding.runId, leaseOwner: "controller-owner", + controllerGeneration: 2, providerAttempt: 1, restartKind: "hard", recoveryRequestId: null, + } }); + if (scenario === "unavailable" || scenario === "mismatched") { + await expect(resumed).rejects.toThrow(); + expect(state.createBackend).not.toHaveBeenCalled(); + } else { + await expect(resumed).resolves.toBeDefined(); + state.createBackend.mock.calls[0]![1].codexTransportFactory!(); + const options = state.createTransport.mock.calls[0]![0] as RunnerTransportOptions & { adoptExistingRunner?: { pid: number } }; + expect(options.prpIdentity).toMatchObject({ runId: identity.runId, runnerInstanceId: identity.runnerInstanceId }); + expect(options.adoptExistingRunner?.pid).toBe(scenario === "live" ? 4321 : undefined); + } + expect(await readFile(controlPlanePath, "utf8")).toBe(original); + await expect(access(join(stateBase, "quarantine"))).rejects.toThrow(); + expect(execute).toHaveBeenCalledOnce(); + } finally { + if (previous === undefined) delete process.env.PAPERCLIP_RUNNER_STATE_DIR; + else process.env.PAPERCLIP_RUNNER_STATE_DIR = previous; + await rm(stateBase, { recursive: true, force: true }); + } + }); + it("resumes an existing scoped authority only for the exact current run", async () => { const stateBase = await mkdtemp( join(tmpdir(), "paperclip-current-scoped-state-"), diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index 6e6eca4c87b..2c6fee0d42b 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -1,6 +1,11 @@ import { createHash, randomUUID } from "node:crypto"; import { execFileSync } from "node:child_process"; import { hasSandboxPerformanceTrace } from "../sandbox-performance.js"; +import { + adoptVerifiedRemoteRunner, + verifyRemoteRunnerRecovery, + type RemoteRunnerRecovery, +} from "./remote-runner-recovery.js"; import { getActiveStepContext } from "@paperclipai/adapter-utils/acpx-engine/startup-timing"; import { chmodSync, @@ -1618,8 +1623,40 @@ async function migrateRunnerdStateRootForExecution(input: { allowVerifiedBackup: boolean; allowRetainedWarmRunner: boolean; restartRecovery?: NativeRestartRecoveryClaim; -}): Promise { + runnerExecutionTarget?: AdapterExecutionTarget | null; +}): Promise { const scoped = scopedRunnerdStateRoot(input.execution); + if (input.restartRecovery?.kind === "reconcile_remote_runner") { + const target = input.runnerExecutionTarget; + const identity = readRunnerdDurableIdentity(scoped); + if ( + input.restartRecovery.runId !== input.execution.binding.runId || + !durableIdentityMatchesSession(identity, input.execution) || + identity.runId !== input.execution.binding.runId || + target?.kind !== "remote" || + target.transport !== "sandbox" || + !target.runner || + runnerdAuthorityLifecycle(scoped, identity) === "indeterminate" + ) { + throw new Error("runner_remote_recovery_unverified"); + } + // Active sandbox authority is remote; a suspended host backup need not + // exist yet. Preserve this exact root even if the remote probe fails. + return verifyRemoteRunnerRecovery({ + runner: target.runner, + identity: { + runId: identity.runId, + normalizedSessionId: identity.normalizedSessionId, + runnerInstanceId: identity.runnerInstanceId, + environmentLeaseId: identity.environmentLeaseId, + }, + stateDirectory: posix.join( + target.remoteCwd, ".paperclip-runtime", "paperclip-runner", "sessions", + createHash("sha256").update(nativeSessionKey(input.execution)).digest("hex"), + "runner", + ), + }); + } if (existsSync(scoped)) { if (!isSafeNativeStateDirectory(scoped)) { throw new Error("runner_state_directory_unsafe"); @@ -3815,8 +3852,9 @@ async function executePaperclipNativeSessionWithinScope( if (environmentScope) { await trace.end(environmentScope, { endedAtMs: environmentEndedAtMs }); } + let verifiedRemoteRecovery: RemoteRunnerRecovery | undefined; if (input.useRunnerd) { - await migrateRunnerdStateRootForExecution({ + verifiedRemoteRecovery = await migrateRunnerdStateRootForExecution({ db: input.db, execution: input.execution, allowVerifiedBackup: @@ -3828,6 +3866,7 @@ async function executePaperclipNativeSessionWithinScope( // durable-state verifier continues to require a suspended runner. allowRetainedWarmRunner: hasIdleWarmNativeSessionOwner(input), restartRecovery: input.restartRecovery, + runnerExecutionTarget: input.runnerExecutionTarget, }); } const durableRunnerBinding = input.useRunnerd @@ -3958,7 +3997,10 @@ async function executePaperclipNativeSessionWithinScope( } const nextAttempt = nextNativeProviderAttempt( coordinator.attempt, - recovering?.kind, + recovering?.kind === "reconcile_remote_runner" && + verifiedRemoteRecovery?.alive === false + ? "resume_dead_runner" + : recovering?.kind, ); const nextControllerGeneration = recovering ? recovering.controllerGeneration @@ -4454,7 +4496,7 @@ async function executePaperclipNativeSessionWithinScope( runnerInstanceId: effectiveRunnerInstanceId, durableEnvironmentLeaseId: durableRunnerBinding?.environmentLeaseId, trace, - }) + }, verifiedRemoteRecovery) : null; nativeSessionExecuteStartedAtMs = Date.now(); native = await trace.measure( @@ -6370,7 +6412,7 @@ export async function createRunnerdBackend(input: { contextSnapshot: Record; }, ) => Promise; -}): Promise { +}, verifiedRemoteRecovery?: RemoteRunnerRecovery): Promise { const sessionScopeId = nativeSessionScopeKey(input.execution); if (initializingSessionToolAuthorities.has(sessionScopeId)) { throw new Error("native_session_supervisor_busy"); @@ -6386,7 +6428,7 @@ export async function createRunnerdBackend(input: { executingRunnerdSessionScopes.get(sessionScopeId) !== input.execution.binding.runId ) { - await migrateRunnerdStateRootForExecution({ + verifiedRemoteRecovery = await migrateRunnerdStateRootForExecution({ db: input.db, execution: input.execution, allowVerifiedBackup: @@ -6394,9 +6436,12 @@ export async function createRunnerdBackend(input: { input.runnerExecutionTarget.transport === "sandbox", allowRetainedWarmRunner: false, restartRecovery: input.restartRecovery, + runnerExecutionTarget: input.runnerExecutionTarget, }); } - return await createRunnerdBackendWithinSessionClaim(input, sessionScopeId); + return await createRunnerdBackendWithinSessionClaim( + input, sessionScopeId, verifiedRemoteRecovery, + ); } finally { initializingSessionToolAuthorities.delete(sessionScopeId); } @@ -6405,6 +6450,7 @@ export async function createRunnerdBackend(input: { async function createRunnerdBackendWithinSessionClaim( input: Parameters[0], sessionScopeId: string, + verifiedRemoteRecovery?: RemoteRunnerRecovery, ): Promise { const target = input.runnerExecutionTarget ?? { kind: "local" as const }; const authority = new PaperclipRunnerToolAuthority(input.db, { @@ -6760,6 +6806,20 @@ async function createRunnerdBackendWithinSessionClaim( ) return; selectedRemoteMode = requiredMode; + if (verifiedRemoteRecovery?.alive) { + // The executor is already using these artifacts. Verify in place; do + // not replace binaries, provider packs, or credential files beneath it. + await verifyRemoteRunner(requiredMode); + if (requiresRemoteProviderPack) { + if (!stagedRemoteProviderPackRoot) throw new Error("runner_remote_recovery_unverified"); + await verifyRemoteProviderPack(stagedRemoteProviderPackRoot); + activeRemoteProviderPackRoot = stagedRemoteProviderPackRoot; + } + remotePrepared = true; + remoteHarnessStatePrepared = true; + resolveRemoteRunnerProcessSpawned?.(); + return; + } let usedPreinstalledRunner = false; const explicitRemoteBinary = input.runnerRemoteBinaryPath?.trim() || null; if (mayUsePreinstalledRunnerArtifact(explicitRemoteBinary)) { @@ -7899,6 +7959,15 @@ async function createRunnerdBackendWithinSessionClaim( input.restartRecovery?.kind === "reattach_existing_runner" ? input.restartRecovery.process : null; + const adoptedRunner = verifiedRemoteRecovery?.alive && remoteCommandRunner + ? adoptVerifiedRemoteRunner(remoteCommandRunner, verifiedRemoteRecovery) + : adoptedProcess + ? { + ...adoptedProcess, + isAlive: () => verifiedRecoveryProcessIsAlive(adoptedProcess), + signal: (signal: NodeJS.Signals) => signalVerifiedRecoveryProcess(adoptedProcess, signal), + } + : undefined; const executeCurrentToolAuthority = ( call: Parameters[0], ) => { @@ -8043,15 +8112,8 @@ async function createRunnerdBackendWithinSessionClaim( ? resolveSourceCodexHome(input.runnerEnvironment ?? process.env) : undefined, runnerProcessLauncher: remoteProcessLauncher, - runnerReconnectGraceMs: remoteTarget ? 120_000 : undefined, - adoptExistingRunner: adoptedProcess - ? { - ...adoptedProcess, - isAlive: () => verifiedRecoveryProcessIsAlive(adoptedProcess), - signal: (signal) => - signalVerifiedRecoveryProcess(adoptedProcess, signal), - } - : undefined, + runnerReconnectGraceMs: remoteTarget ? 300_000 : undefined, + adoptExistingRunner: adoptedRunner, environment: effectiveRunnerEnvironment, onDiagnostic: (message) => { void input.onLog?.( diff --git a/server/src/services/native-runtime/remote-runner-recovery.test.ts b/server/src/services/native-runtime/remote-runner-recovery.test.ts new file mode 100644 index 00000000000..0e27eadb1a3 --- /dev/null +++ b/server/src/services/native-runtime/remote-runner-recovery.test.ts @@ -0,0 +1,127 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import * as fs from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { createRequire } from "node:module"; +import { runInNewContext } from "node:vm"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { adoptVerifiedRemoteRunner, REMOTE_RECOVERY_PROBE, verifyRemoteRunnerRecovery } from "./remote-runner-recovery.js"; + +const require = createRequire(import.meta.url); +const directories: string[] = []; +afterEach(() => { for (const dir of directories.splice(0)) fs.rmSync(dir, { recursive: true, force: true }); }); + +function fixture() { + const root = fs.realpathSync(fs.mkdtempSync(join(tmpdir(), "remote-recovery-"))); + directories.push(root); + const dir = join(root, "runner"), proc = join(root, "proc", "4321"); + fs.mkdirSync(dir); fs.mkdirSync(proc, { recursive: true }); + const identity = { runId: "run-a", normalizedSessionId: "session-a", runnerInstanceId: "runner-a", environmentLeaseId: "lease-a" }; + const marker = "ec0e1ae3-0614-44fc-a352-bb03d89134d7\n4321\n2026-09-10T17:00:00.000Z\nrunner-a\n"; + fs.writeFileSync(join(dir, "runner-process.identity"), marker); + fs.writeFileSync(join(dir, "runner-state.json"), JSON.stringify({ schema: "paperclip.runner.durable.state.v1", ...identity, lifecycle: "ready", privateJournal: "must not leave sandbox" })); + const fields = ["S", ...Array(18).fill("0"), "13579"]; + fs.writeFileSync(join(proc, "stat"), "4321 (paperclip runner) " + fields.join(" ")); + fs.writeFileSync(join(proc, "cmdline"), ["runnerd", "--runner-id", "runner-a", "--state-dir", dir, "--run-id", "run-a", "--session-id", "session-a", "--environment-lease-id", "lease-a"].join("\0")); + const kill = vi.fn(); + // Real bounded file-descriptor reads and path checks on every OS; only the + // Linux process table and signal syscall are replaced by this fixture. + const remoteFs = { + ...fs, + realpathSync: (path: string) => path.startsWith("/proc/") ? path : fs.realpathSync(path), + openSync: (path: string, flags: number) => fs.openSync(path.startsWith("/proc/") ? join(root, path.slice(1)) : path, flags), + }; + const invoke = (overrides: Record = {}) => { + let stdout = ""; + runInNewContext(REMOTE_RECOVERY_PROBE, { + require: (name: string) => name === "node:fs" ? remoteFs : require(name), + Buffer, + process: { argv: ["node", JSON.stringify({ identity, stateDirectory: dir, mode: "state", ...overrides })], kill }, + console: { log: (value: string) => { stdout += value; } }, + }, { timeout: 1000 }); + return JSON.parse(stdout); + }; + return { root, dir, proc, identity, invoke, kill }; +} + +describe("remote recovery evidence", () => { + it("reads the remote authority and exact process without returning the journal", () => { + const f = fixture(), result = f.invoke(); + expect(result).toMatchObject({ identity: f.identity, alive: true, lifecycle: "ready", process: { pid: 4321, startTicks: "13579" } }); + expect(JSON.stringify(result)).not.toContain("privateJournal"); + expect(f.kill).toHaveBeenCalledExactlyOnceWith(4321, 0); + }); + + it.each(["runId", "normalizedSessionId", "runnerInstanceId", "environmentLeaseId"])("rejects a different %s", (field) => { + const f = fixture(); + expect(() => f.invoke({ identity: { ...f.identity, [field]: "other" } })).toThrow(); + }); + + it("adopts a warm process whose launch run preceded its current durable run", () => { + const f = fixture(), file = join(f.proc, "cmdline"); + fs.writeFileSync(file, fs.readFileSync(file, "utf8").replace("run-a", "original-cold-run")); + expect(f.invoke()).toMatchObject({ identity: f.identity, alive: true }); + }); + + it.each(["runner-state.json", "runner-process.identity"])("rejects a symlink for %s", (name) => { + const f = fixture(), original = join(f.dir, name), elsewhere = join(f.root, "private"); + fs.renameSync(original, elsewhere); fs.symlinkSync(elsewhere, original); + expect(() => f.invoke()).toThrow(); + }); + + it("rejects an ancestor symlink", () => { + const f = fixture(), alias = join(f.root, "alias"); fs.symlinkSync(f.dir, alias); + expect(() => f.invoke({ stateDirectory: alias })).toThrow(); + }); + + it("bounds reads and rejects malformed or oversized state", () => { + const f = fixture(), path = join(f.dir, "runner-state.json"); + fs.writeFileSync(path, ""); expect(() => f.invoke()).toThrow(); + fs.truncateSync(path, 16 * 1024 * 1024 + 1); expect(() => f.invoke()).toThrow(); + }); + + it("rejects a recycled process before signalling", () => { + const f = fixture(), observed = f.invoke(); + const stat = join(f.proc, "stat"); fs.writeFileSync(stat, fs.readFileSync(stat, "utf8").replace("13579", "24680")); + expect(() => f.invoke({ mode: "process", process: observed.process, signal: "SIGTERM" })).toThrow(); + expect(f.kill.mock.calls.every((call) => call[1] === 0)).toBe(true); + }); + + it("rejects a changed process marker before signalling", () => { + const f = fixture(), observed = f.invoke(); + expect(() => f.invoke({ mode: "process", process: { ...observed.process, nonce: "other" }, signal: "SIGTERM" })).toThrow(); + expect(f.kill.mock.calls.every((call) => call[1] === 0)).toBe(true); + }); + + it("only signals a process with all original fingerprints intact", () => { + const f = fixture(), observed = f.invoke(); + f.invoke({ mode: "process", process: observed.process, signal: "SIGTERM" }); + expect(f.kill).toHaveBeenLastCalledWith(4321, "SIGTERM"); + }); + + it.each(["cmdline", "stat"])("treats a vanished %s as dead", (name) => { + const f = fixture(); fs.unlinkSync(join(f.proc, name)); + expect(f.invoke().alive).toBe(false); + }); + + it("requires suspended state for a dead remote process", async () => { + const f = fixture(), proof = f.invoke(); + const execute = vi.fn(async () => ({ exitCode: 0, timedOut: false, stdout: JSON.stringify({ ...proof, alive: false }) })); + const runner = { execute } as unknown as CommandManagedRuntimeRunner; + await expect(verifyRemoteRunnerRecovery({ runner, identity: f.identity, stateDirectory: f.dir })).rejects.toThrow("runner_remote_recovery_unverified"); + proof.lifecycle = "suspended"; + await expect(verifyRemoteRunnerRecovery({ runner, identity: f.identity, stateDirectory: f.dir })).resolves.toMatchObject({ alive: false }); + expect(() => adoptVerifiedRemoteRunner(runner, { ...proof, alive: false })).toThrow(); + }); + + it("coalesces concurrent liveness calls and propagates provider failures", async () => { + const f = fixture(), proof = f.invoke(); + const execute = vi.fn(async () => ({ exitCode: 0, timedOut: false, stdout: JSON.stringify(proof) })); + const runner = { execute } as unknown as CommandManagedRuntimeRunner; + const adopted = adoptVerifiedRemoteRunner(runner, proof); + expect(await Promise.all([adopted.isAlive(), adopted.isAlive(), adopted.isAlive()])).toEqual([true, true, true]); + expect(execute).toHaveBeenCalledOnce(); + execute.mockRejectedValueOnce(new Error("provider unavailable")); + await expect(adopted.signal("SIGTERM")).rejects.toThrow("provider unavailable"); + }); +}); diff --git a/server/src/services/native-runtime/remote-runner-recovery.ts b/server/src/services/native-runtime/remote-runner-recovery.ts new file mode 100644 index 00000000000..05da14c06d8 --- /dev/null +++ b/server/src/services/native-runtime/remote-runner-recovery.ts @@ -0,0 +1,163 @@ +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; + +export type RemoteRecoveryIdentity = { + runId: string; + normalizedSessionId: string; + runnerInstanceId: string; + environmentLeaseId: string; +}; + +type RemoteProcess = { + nonce: string; + pid: number; + startedAt: string; + startTicks: string | null; +}; + +export type RemoteRunnerRecovery = { + identity: RemoteRecoveryIdentity; + stateDirectory: string; + lifecycle: string; + process: RemoteProcess; + alive: boolean; +}; + +// The sandbox already needs Node for work-folder transfers. Return only the +// binding and process fingerprint, never the PRP journal or provider output. +// Every read is bounded, rejects symlinks, and pins an opened regular file. +export const REMOTE_RECOVERY_PROBE = String.raw` +const fs = require('node:fs'), path = require('node:path'), assert = require('node:assert/strict'); +const request = JSON.parse(process.argv[1]); +const dir = request.stateDirectory; +assert(path.isAbsolute(dir) && path.normalize(dir) === dir && fs.realpathSync(dir) === dir); +function read(file, limit) { + assert(fs.realpathSync(path.dirname(file)) === path.dirname(file)); + const fd = fs.openSync(file, fs.constants.O_RDONLY | fs.constants.O_NOFOLLOW); + try { + const stat = fs.fstatSync(fd); assert(stat.isFile() && stat.size <= limit); + const data = Buffer.alloc(limit + 1); let length = 0, count; + do { count = fs.readSync(fd, data, length, data.length - length, null); length += count; } while (count && length < data.length); + assert(length <= limit); return data.subarray(0, length).toString('utf8'); + } finally { fs.closeSync(fd); } +} +const marker = read(path.join(dir, 'runner-process.identity'), 4096).trim().split('\n'); +assert(marker.length === 4); +const [nonce, rawPid, startedAt, runnerId] = marker; +const pid = Number(rawPid); +assert(/^[0-9a-f-]{36}$/.test(nonce) && Number.isSafeInteger(pid) && pid > 0 && Number.isFinite(Date.parse(startedAt))); +assert.equal(runnerId, request.identity.runnerInstanceId); +let alive = false, startTicks = null; +try { + process.kill(pid, 0); + const stat = read('/proc/' + pid + '/stat', 65536); + const fields = stat.slice(stat.lastIndexOf(')') + 2).trim().split(/\s+/); + startTicks = fields[19]; assert(/^\d+$/.test(startTicks)); + alive = fields[0] !== 'Z' && fields[0] !== 'X'; + if (alive) { + const args = read('/proc/' + pid + '/cmdline', 65536).split('\0'); + function flag(name, expected) { const index = args.indexOf(name); assert(index >= 0 && args[index + 1] === expected); } + flag('--runner-id', runnerId); flag('--state-dir', dir); + // run.attach rotates the durable run binding on a warm process. Its + // original --run-id is not the current authority; runner-state proves it. + flag('--session-id', request.identity.normalizedSessionId); + flag('--environment-lease-id', request.identity.environmentLeaseId); + } +} catch (error) { if (error.code !== 'ESRCH' && error.code !== 'ENOENT') throw error; alive = false; } +const fingerprint = { nonce, pid, startedAt, startTicks }; +if (request.process) { + for (const key of ['nonce', 'pid', 'startedAt']) assert.equal(fingerprint[key], request.process[key]); + if (alive) assert.equal(startTicks, request.process.startTicks); +} +let lifecycle; +if (request.mode === 'state') { + const state = JSON.parse(read(path.join(dir, 'runner-state.json'), 16 * 1024 * 1024)); + assert.equal(state.schema, 'paperclip.runner.durable.state.v1'); + for (const [key, value] of Object.entries(request.identity)) assert.equal(state[key], value); + lifecycle = state.lifecycle; +} +if (request.signal && alive) { + assert(['SIGTERM', 'SIGINT', 'SIGKILL'].includes(request.signal)); + process.kill(pid, request.signal); +} +console.log(JSON.stringify({ identity: request.identity, stateDirectory: dir, lifecycle, process: fingerprint, alive })); +`; + +async function probe( + runner: CommandManagedRuntimeRunner, + request: { + identity: RemoteRecoveryIdentity; + stateDirectory: string; + mode: "state" | "process"; + process?: RemoteProcess; + signal?: NodeJS.Signals; + }, +): Promise { + const result = await runner.execute({ + command: "node", + args: ["-e", REMOTE_RECOVERY_PROBE, JSON.stringify(request)], + bypassSession: true, + timeoutMs: 10_000, + }); + if (result.exitCode !== 0 || result.timedOut || result.stdout.length > 8192) { + throw new Error("runner_remote_recovery_unverified"); + } + let value: RemoteRunnerRecovery; + try { value = JSON.parse(result.stdout); } catch { + throw new Error("runner_remote_recovery_unverified"); + } + if (value.stateDirectory !== request.stateDirectory || + !value.identity || Object.entries(request.identity).some(([k, v]) => value.identity[k as keyof RemoteRecoveryIdentity] !== v) || + typeof value.alive !== "boolean" || !value.process || + !Number.isSafeInteger(value.process.pid) || value.process.pid <= 0 || + typeof value.process.nonce !== "string" || + !Number.isFinite(Date.parse(value.process.startedAt))) { + throw new Error("runner_remote_recovery_unverified"); + } + return value; +} + +export async function verifyRemoteRunnerRecovery(input: { + runner: CommandManagedRuntimeRunner; + identity: RemoteRecoveryIdentity; + stateDirectory: string; +}): Promise { + const result = await probe(input.runner, { + identity: input.identity, stateDirectory: input.stateDirectory, mode: "state", + }); + // An unconfirmed dead executor can leave provider children alive. Only a + // durably suspended executor authorizes a fresh process; a live executor + // must be adopted and prove its existing PRP credentials before any work. + if ((!result.alive && result.lifecycle !== "suspended") || + !["connecting", "ready", "backpressure", "recoverable_failure", "suspended"].includes(result.lifecycle)) { + throw new Error("runner_remote_recovery_unverified"); + } + return result; +} + +export function adoptVerifiedRemoteRunner( + runner: CommandManagedRuntimeRunner, + evidence: RemoteRunnerRecovery, +) { + if (!evidence.alive) throw new Error("runner_remote_recovery_unverified"); + let pending: Promise | undefined; + let lastAliveAt = 0; + return { + pid: evidence.process.pid, + processGroupId: null, + startedAt: evidence.process.startedAt, + isAlive: () => { + // Both startup and the process monitor ask for liveness. Coalesce them + // and cap remote probes at one per second, with one RPC in flight. + if (Date.now() - lastAliveAt < 1000) return Promise.resolve(true); + pending ??= probe(runner, { ...evidence, mode: "process" }) + .then((result) => { if (result.alive) lastAliveAt = Date.now(); return result.alive; }) + .finally(() => { pending = undefined; }); + return pending; + }, + signal: async (signal: NodeJS.Signals) => { + const result = await probe(runner, { ...evidence, mode: "process", signal }); + lastAliveAt = 0; + return result.alive; + }, + }; +}