diff --git a/doc/execution-semantics.md b/doc/execution-semantics.md index e8938daa137..7cceed149b4 100644 --- a/doc/execution-semantics.md +++ b/doc/execution-semantics.md @@ -4841,7 +4841,19 @@ the Runner holds its own authenticated connection to the controller, the monitor keeps the same runner and turn and probes again. The bound is the remote runner reconnect grace (7 minutes). When the Runner has no connection and the 30-second budget is spent, or when the bound is spent, the monitor reports -`runner_process_termination_unconfirmed` and the run rests with Retry. Monitoring checks the saved process generation when available. +`runner_process_termination_unconfirmed` and the run rests with Retry. + +A plugin RPC timeout (`PLUGIN_RPC_ERROR_CODES.TIMEOUT`) is a stall of the +environment plugin, not an answer from the sandbox. When a read-only recovery +read during reattachment times out this way, the recovery's probe budget grows +from 30 seconds to 3 minutes, capped by the caller's recovery deadline. This +covers the saved runner state, the identity marker and the process check, and +the settled-state read after a confirmed exit. A signal is never retried. Other +transport failures keep the 30-second budget. At the launch of a new remote +runner, a launch call that stalls this way is not retried. The controller reads +the identity marker for the launch nonce instead, so the one launched runner is +adopted or cleaned up by its nonce. The identity read uses the same 3-minute +stall budget. Each answer that the sandbox gives keeps its current meaning. Monitoring checks the saved process generation when available. Only a matching ownership marker and confirmed process exit permit automatic replacement. Missing or conflicting identity evidence, failed reads, and an exhausted probe budget retain the existing ownership and recovery safeguards. 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 26a0cbd008f..c9c71a3bcb3 100644 --- a/server/src/services/native-runtime/native-session-executor.test.ts +++ b/server/src/services/native-runtime/native-session-executor.test.ts @@ -84,6 +84,7 @@ import { nativeSha256 } from "./canonical.js"; import { resolveRemoteProviderPackAuthority } from "./remote-provider-pack-authority.js"; import { REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT, RemoteRunnerRecoveryError, readRemoteRunnerRecoveryError } from "./remote-runner-recovery-diagnostics.js"; import { REMOTE_RUNNER_RECONNECT_GRACE_MS } from "./native-runner-recovery-contract.js"; +import { JsonRpcCallError, PLUGIN_RPC_ERROR_CODES } from "@paperclipai/plugin-sdk"; import { agentDirectoryWorkingCopyService } from "../agent-directory-working-copies.js"; import { probeAgentDirectory } from "../agent-directory-probe.js"; import { AGENT_FILES_CONTRACT } from "../agent-file-store.js"; @@ -862,6 +863,96 @@ describe("remote controller restart adoption", () => { expect(probes).toBe(2); expect(execute.mock.calls.every(([request]) => !request.args.join(" ").match(/nohup|mkdir|rm -|mv -|kill -"/))).toBe(true); }); + const pluginStall = () => new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.TIMEOUT, + message: 'RPC call "environmentExecute" timed out after 40000ms' }); + it.each(["runner_state", "identity_marker", "process_verification"] as const)( + "reattaches the same runner through a plugin RPC stall during the %s read", async stage => { + vi.useFakeTimers(); + onTestFinished(() => { vi.useRealTimers(); }); + const { target, execute } = fixture(); + const original = execute.getMockImplementation()!; + const operation = stage === "identity_marker" ? "paperclip-runner-recovery-identity" : "paperclip-runner-recovery-check"; + let stalls = 0; + execute.mockImplementation(async request => { + if ((stage === "runner_state" ? request.args[1].includes("base64") : request.args[2] === operation) && stalls < 2) { + stalls++; + await new Promise(resolveWait => setTimeout(resolveWait, 40_000)); + throw pluginStall(); + } + return original(request); + }); + const outcome = verifyRemoteRunnerReattachment({ + claim, target: target as never, identity, runId: "run", normalizedSessionId: "session", + }).catch(error => error as Error); + await vi.advanceTimersByTimeAsync(90_000); + const adopted = await outcome; + expect(adopted).not.toBeInstanceOf(Error); + expect((adopted as { pid: number }).pid).toBe(123); + expect(stalls).toBe(2); + expect(execute.mock.calls.every(([request]) => !request.args.join(" ").match(/nohup|mkdir|rm -|mv -|kill -"/))).toBe(true); + }); + it("fails reattachment closed when the plugin stall outlasts its bound", async () => { + vi.useFakeTimers(); + onTestFinished(() => { vi.useRealTimers(); }); + const { target, execute } = fixture(); + const original = execute.getMockImplementation()!; + execute.mockImplementation(async request => { + if (request.args[2] !== "paperclip-runner-recovery-check") return original(request); + await new Promise(resolveWait => setTimeout(resolveWait, 40_000)); + throw pluginStall(); + }); + let settled = false; + const outcome = verifyRemoteRunnerReattachment({ + claim, target: target as never, identity, runId: "run", normalizedSessionId: "session", + }).catch(error => { settled = true; return error as Error; }); + await vi.advanceTimersByTimeAsync(150_000); + expect(settled).toBe(false); + await vi.advanceTimersByTimeAsync(80_000); + expect(readRemoteRunnerRecoveryError(await outcome)).toEqual({ stage: "process_verification", reason: "probe_transport_failed" }); + }); + it("keeps the 30-second bound for a transport failure that is not a plugin stall", async () => { + vi.useFakeTimers(); + onTestFinished(() => { vi.useRealTimers(); }); + const { target, execute } = fixture(); + const original = execute.getMockImplementation()!; + execute.mockImplementation(async request => { + if (request.args[2] !== "paperclip-runner-recovery-check") return original(request); + throw new Error("Sandbox with ID test-deleted-sandbox not found"); + }); + const startedAt = Date.now(); + const outcome = verifyRemoteRunnerReattachment({ + claim, target: target as never, identity, runId: "run", normalizedSessionId: "session", + }).catch(error => error as Error); + await vi.advanceTimersByTimeAsync(30_000); + expect(readRemoteRunnerRecoveryError(await outcome)).toEqual({ stage: "process_verification", reason: "probe_transport_failed" }); + expect(Date.now() - startedAt).toBeLessThanOrEqual(30_000); + }); + it("reads the settled state of an exited runner through a plugin RPC stall", async () => { + vi.useFakeTimers(); + onTestFinished(() => { vi.useRealTimers(); }); + const settled = { lifecycle: "suspended", outbox: [], pendingTerminalDelivery: null, pendingProviderCleanup: null }; + const { target, execute } = fixture(settled); + const original = execute.getMockImplementation()!; + let exited = false; + let stalls = 0; + execute.mockImplementation(async request => { + if (request.args[2] === "paperclip-runner-recovery-check") { + exited = true; + return { stdout: "", stderr: "", exitCode: 45, timedOut: false }; + } + if (exited && request.args[1].includes("base64") && stalls === 0) { + stalls++; + await new Promise(resolveWait => setTimeout(resolveWait, 40_000)); + throw pluginStall(); + } + return original(request); + }); + const outcome = verifyRemoteRunnerOrSettledExit({ claim, target: target as never, identity, + runId: "run", normalizedSessionId: "session", providerAttempt: 2 }).catch(error => error as Error); + await vi.advanceTimersByTimeAsync(60_000); + expect(await outcome).toBeUndefined(); + expect(stalls).toBe(1); + }); it("rejects a marker that cannot prove its process generation", async () => { const { target, execute } = fixture(); const original = execute.getMockImplementation()!; @@ -1672,6 +1763,50 @@ fs.readFileSync = function(path, ...args) { } finally { vi.useRealTimers(); } }); + it.each(["launch", "identity"] as const)("supervises the one launched runner through a plugin RPC stall on the %s call", async stalled => { + vi.useFakeTimers(); + let nonce = ""; + let markerWritten = false; + let stalls = 0; + const ok = { exitCode: 0, signal: null, timedOut: false, stdout: "", stderr: "" }; + const stall = async () => { + stalls++; + await new Promise(resolve => setTimeout(resolve, 40_000)); + throw new JsonRpcCallError({ code: PLUGIN_RPC_ERROR_CODES.TIMEOUT, message: 'RPC call "environmentExecute" timed out after 40000ms' }); + }; + const execute = vi.fn(async (input: { args?: string[] }) => { + const label = input.args?.[2]; + if (label === "paperclip-runner-launch") { + nonce = input.args![4]!; + markerWritten = true; + if (stalled === "launch") return stall(); + return ok; + } + if (label === "paperclip-runner-process-identity") { + if (stalled === "identity" && stalls === 0) return stall(); + return markerWritten ? { ...ok, stdout: `${nonce}\n4321\n2026-09-06T00:00:00.000Z\nrunner-remote\n` } : { ...ok, exitCode: 1 }; + } + if (label === "paperclip-runner-monitor") return { ...ok, exitCode: 45 }; + return { ...ok, exitCode: 1 }; + }); + const onSpawn = vi.fn(async () => undefined); + const handle = createRemoteRunnerProcessLauncher({ + target: { kind: "remote", transport: "sandbox", environmentId: "environment", leaseId: "lease", remoteCwd: "/workspace" }, + runner: { execute } as never, remoteBinary: "/runtime/runnerd", processIdentityPath: "/runtime/identity", + stateDirectory: "/runtime", diagnosticsDirectory: "/runtime/diagnostics", runnerInstanceId: "runner-remote", onSpawn, + })({ command: "/controller/runnerd", args: [], cwd: "/controller", environment: {} }); + const outcome = handle.completion.catch(error => error as Error); + try { + await vi.advanceTimersByTimeAsync(60_000); + expect(await outcome).toMatchObject({ code: null }); + expect(stalls).toBe(1); + expect(onSpawn).toHaveBeenCalledExactlyOnceWith({ pid: 4321, processGroupId: null, startedAt: "2026-09-06T00:00:00.000Z" }); + expect(handle.isExitConfirmed?.()).toBe(true); + expect(execute.mock.calls.filter(([x]) => x.args?.[2] === "paperclip-runner-launch")).toHaveLength(1); + expect(execute.mock.calls.some(([x]) => x.args?.[2] === "paperclip-runner-identity-failure-cleanup")).toBe(false); + } finally { vi.useRealTimers(); } + }); + it("bounds supervision without exit proof when a connected runner's probes never answer", async () => { vi.useFakeTimers(); let nonce = ""; diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index c4a1b48c36f..db648cb26ed 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -282,7 +282,7 @@ import { REMOTE_RUNNER_EXIT_EVIDENCE_SCRIPT, RemoteRunnerRecoveryError, attachRemoteRunnerRecoveryDiagnostic, boundRemoteRunnerExitEvidence, describeExitedRemoteRunner, describeExitedRemoteRunnerReason, readRemoteRunnerRecoveryError, remoteRecoveryProbeReason, } from "./remote-runner-recovery-diagnostics.js"; -import { readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS } from "./remote-runner-recovery-probe.js"; +import { isPluginRpcTimeout, readRemoteRunnerRecoveryProbe, REMOTE_RUNNER_PLUGIN_STALL_BUDGET_MS, REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS, remoteRunnerPluginStall, type RemoteRunnerPluginStall } from "./remote-runner-recovery-probe.js"; import { describeOwnFailedAttachment, inspectRemoteWarmSessionRecovery, isSettledRemoteWarmControlState, summarizePostTerminalEvents, type RemoteWarmSessionRecovery, unsettledControllerCommands } from "./remote-warm-session-recovery.js"; import { assertNativeWarmIngressRetirement, nativeWarmIngressAllocation, nativeWarmIngressIdentity, type NativeWarmIngressClaim, type NativeWarmIngressOwner } from "./native-warm-ingress-ownership.js"; import { resolveNativeStagedAttachmentRecovery, type NativeStagedAttachmentRecovery } from "./native-staged-attachment-recovery.js"; @@ -14506,6 +14506,7 @@ export async function readRemoteRunnerState(input: { runner: CommandManagedRuntimeRunner; stateDirectory: string; recoveryDeadline?: number; + stall?: RemoteRunnerPluginStall; }): Promise> { const statePath = posix.join(input.stateDirectory, "runner-state.json"); const escapedPath = statePath.replaceAll("'", "'\\''"); @@ -14519,7 +14520,7 @@ export async function readRemoteRunnerState(input: { ? await input.runner.execute(request) : await readRemoteRunnerRecoveryProbe({ runner: input.runner, request, stage: "runner_state", - deadline: input.recoveryDeadline, readOnly: true, + deadline: input.recoveryDeadline, readOnly: true, stall: input.stall, }); if (result.exitCode !== 0 || result.timedOut) { const error = new Error( @@ -14678,6 +14679,7 @@ export async function verifyRemoteRunnerReattachment(input: { normalizedSessionId: string; requireIdle?: boolean; recoveryDeadline?: number; + stall?: RemoteRunnerPluginStall; }) { const { claim, target, identity } = input; if (claim.remote.kind === "plugin_task") throw remoteRecoveryLeaseMismatch("runner_verification", "plugin_task"); @@ -14694,8 +14696,9 @@ export async function verifyRemoteRunnerReattachment(input: { const runner = target.runner; const probeDeadline = Math.min(input.recoveryDeadline ?? Number.POSITIVE_INFINITY, Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS); + const stall = input.stall ?? remoteRunnerPluginStall(input.recoveryDeadline); const stateDirectory = remoteRunnerStateDirectory(target.remoteCwd, input.normalizedSessionId); - const runnerState = await readRemoteRunnerState({ runner, stateDirectory, recoveryDeadline: probeDeadline }); + const runnerState = await readRemoteRunnerState({ runner, stateDirectory, recoveryDeadline: probeDeadline, stall }); if (input.requireIdle && (runnerState.lifecycle !== "ready" || !Array.isArray(runnerState.outbox) || runnerState.outbox.length !== 0 || runnerState.pendingTerminalDelivery !== null)) { throw new Error("runner_state_identity_mismatch: remote_warm_session_not_idle"); @@ -14711,7 +14714,7 @@ export async function verifyRemoteRunnerReattachment(input: { } const identityPath = posix.join(stateDirectory, "runner-process.identity"); const marker = await readRemoteRunnerRecoveryProbe({ - runner, stage: "identity_marker", deadline: probeDeadline, readOnly: true, + runner, stage: "identity_marker", deadline: probeDeadline, readOnly: true, stall, request: { command: "sh", args: [ @@ -14739,9 +14742,10 @@ export async function verifyRemoteRunnerReattachment(input: { if (!processIdentity) throw new RemoteRunnerRecoveryError({ stage: "identity_marker", reason: "marker_invalid" }); if (!processIdentity.processStartFingerprint) throw new RemoteRunnerRecoveryError({ stage: "identity_marker", reason: "fingerprint_missing" }); const expectedStartFingerprint = processIdentity.processStartFingerprint; - const check = async (signal?: NodeJS.Signals, deadline = Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS) => { + const check = async (signal?: NodeJS.Signals, deadline = Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS, + verificationStall?: RemoteRunnerPluginStall) => { const result = await readRemoteRunnerRecoveryProbe({ - runner, stage: "process_verification", deadline, readOnly: !signal, + runner, stage: "process_verification", deadline, readOnly: !signal, stall: verificationStall, request: { command: "sh", args: [ @@ -14766,7 +14770,7 @@ export async function verifyRemoteRunnerReattachment(input: { } return { alive, reason }; }; - const verified = await check(undefined, probeDeadline); + const verified = await check(undefined, probeDeadline, stall); if (!verified.alive) { throw new RemoteRunnerRecoveryError({ stage: "process_verification", reason: verified.reason, ...(verified.reason === "process_not_running" ? { @@ -14812,8 +14816,9 @@ export async function verifyRemoteRunnerOrSettledExit( ) { const recoveryDeadline = Math.min(input.recoveryDeadline ?? Number.POSITIVE_INFINITY, Date.now() + REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS); + const stall = input.stall ?? remoteRunnerPluginStall(input.recoveryDeadline); try { - return await verifyRemoteRunnerReattachment({ ...input, recoveryDeadline }); + return await verifyRemoteRunnerReattachment({ ...input, recoveryDeadline, stall }); } catch (error) { const diagnostic = readRemoteRunnerRecoveryError(error); const target = input.target; @@ -14823,7 +14828,7 @@ export async function verifyRemoteRunnerOrSettledExit( (!exitedBeforeSessionOpen && (diagnostic.runnerLifecycle !== "suspended" || diagnostic.runnerSettled !== true)) || target?.kind !== "remote" || target.transport !== "sandbox" || !target.runner) throw error; const state = await readRemoteRunnerState({ runner: target.runner, - stateDirectory: remoteRunnerStateDirectory(target.remoteCwd, input.normalizedSessionId), recoveryDeadline }); + stateDirectory: remoteRunnerStateDirectory(target.remoteCwd, input.normalizedSessionId), recoveryDeadline, stall }); if (REMOTE_RUNNER_IDENTITY_FIELDS.some(field => state[field] !== input.identity[field])) throw error; if (exitedBeforeSessionOpen ? !runnerStateAllowsFreshStart(state) : state.lifecycle !== "suspended" || !describeExitedRemoteRunner(state).runnerSettled) throw error; @@ -14906,7 +14911,8 @@ async function waitForRemoteRunnerProcessIdentity(input: { runnerInstanceId: string; requireProcessStartFingerprint?: boolean; }): Promise<{ pid: number; startedAt: string; processStartFingerprint?: string }> { - const deadline = Date.now() + REMOTE_RUNNER_PROCESS_IDENTITY_WAIT_MS; + let deadline = Date.now() + REMOTE_RUNNER_PROCESS_IDENTITY_WAIT_MS; + const stallDeadline = Date.now() + REMOTE_RUNNER_PLUGIN_STALL_BUDGET_MS; while (Date.now() < deadline) { const result = await input.runner .execute({ @@ -14920,7 +14926,10 @@ async function waitForRemoteRunnerProcessIdentity(input: { bypassSession: true, timeoutMs: 2_000, }) - .catch(() => null); + .catch((error: unknown) => { + if (isPluginRpcTimeout(error)) deadline = Math.max(deadline, stallDeadline); + return null; + }); const identity = result && result.exitCode === 0 && !result.timedOut ? parseRemoteRunnerProcessIdentity(result.stdout, input) @@ -15113,8 +15122,11 @@ export function createRemoteRunnerProcessLauncher(input: { timeoutMs: 20_000, bypassSession: true, onLog: input.onLog, + }).catch((error: unknown) => { + if (isPluginRpcTimeout(error)) return null; + throw error; }); - if (launchResult.exitCode !== 0 || launchResult.timedOut) { + if (launchResult && (launchResult.exitCode !== 0 || launchResult.timedOut)) { throw new Error( launchResult.timedOut ? "runner_remote_process_launch_timed_out" diff --git a/server/src/services/native-runtime/remote-runner-recovery-probe.ts b/server/src/services/native-runtime/remote-runner-recovery-probe.ts index 77486aa4496..a5724bde041 100644 --- a/server/src/services/native-runtime/remote-runner-recovery-probe.ts +++ b/server/src/services/native-runtime/remote-runner-recovery-probe.ts @@ -1,4 +1,5 @@ import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { PLUGIN_RPC_ERROR_CODES } from "@paperclipai/plugin-sdk"; import { attachRemoteRunnerRecoveryDiagnostic, RemoteRunnerRecoveryError, @@ -6,6 +7,25 @@ import { } from "./remote-runner-recovery-diagnostics.js"; export const REMOTE_RUNNER_RECOVERY_PROBE_BUDGET_MS = 30_000; +export const REMOTE_RUNNER_PLUGIN_STALL_BUDGET_MS = 180_000; + +export function isPluginRpcTimeout(error: unknown): boolean { + let current: unknown = error; + for (let depth = 0; depth < 5 && current instanceof Error; depth++) { + if ((current as { code?: unknown }).code === PLUGIN_RPC_ERROR_CODES.TIMEOUT) return true; + current = current.cause; + } + return false; +} + +export interface RemoteRunnerPluginStall { + deadline: number; + observed: boolean; +} + +export function remoteRunnerPluginStall(cap = Number.POSITIVE_INFINITY): RemoteRunnerPluginStall { + return { deadline: Math.min(cap, Date.now() + REMOTE_RUNNER_PLUGIN_STALL_BUDGET_MS), observed: false }; +} export async function readRemoteRunnerRecoveryProbe(input: { runner: CommandManagedRuntimeRunner; @@ -13,15 +33,16 @@ export async function readRemoteRunnerRecoveryProbe(input: { stage: RemoteRunnerRecoveryDiagnostic["stage"]; deadline: number; readOnly: boolean; + stall?: RemoteRunnerPluginStall; }) { let lastError: unknown = new RemoteRunnerRecoveryError({ stage: input.stage, reason: "probe_timeout", }); - // A few fast gateway failures can exhaust an attempt count in under a - // second. Read-only observations share the caller's deadline instead; - // signals still have one attempt because their outcome may be unknown. + const deadline = () => input.readOnly && input.stall?.observed + ? Math.max(input.deadline, input.stall.deadline) + : input.deadline; for (let attempt = 0; ; attempt++) { - const remaining = input.deadline - Date.now(); + const remaining = deadline() - Date.now(); if (!Number.isFinite(remaining) || remaining <= 0) break; try { const result = await input.runner.execute({ @@ -31,12 +52,13 @@ export async function readRemoteRunnerRecoveryProbe(input: { if (!result.timedOut) return result; lastError = new RemoteRunnerRecoveryError({ stage: input.stage, reason: "probe_timeout" }); } catch (error) { + if (input.readOnly && input.stall && isPluginRpcTimeout(error)) input.stall.observed = true; lastError = attachRemoteRunnerRecoveryDiagnostic(error, { stage: input.stage, reason: "probe_transport_failed", }); } const delayMs = Math.min(250 * 2 ** Math.min(attempt, 3), 2_000); - if (!input.readOnly || input.deadline - Date.now() <= delayMs) break; + if (!input.readOnly || deadline() - Date.now() <= delayMs) break; await new Promise(resolve => setTimeout(resolve, delayMs)); } throw lastError;