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; + }, + }; +}