diff --git a/server/src/services/native-runtime/native-session-executor.ts b/server/src/services/native-runtime/native-session-executor.ts index e1e347fed8d..89b14a11a2c 100644 --- a/server/src/services/native-runtime/native-session-executor.ts +++ b/server/src/services/native-runtime/native-session-executor.ts @@ -59,6 +59,7 @@ import { environmentTaskOperationSchema, environmentTaskResultSchema } from "@pa import { preparePluginRunnerCredentials } from "./plugin-runner-credentials.js"; import { createPluginRunnerTaskOwner, verifyPluginRunnerTaskReattachment } from "./plugin-runner-task.js"; import { findNativePredecessorRetirementUnproven, NATIVE_PREDECESSOR_RETIREMENT_UNPROVEN_CODE, NativePredecessorRetirementUnprovenError } from "./predecessor-retirement.js"; +import { retireSandboxRunnerPortHolders } from "./sandbox-runner-port-holders.js"; import { NativeExecutionContext } from "./native-execution-context.js"; import { NativeStartupCleanupPendingError, clearNativeStartupContext } from "./native-startup-cleanup.js"; import { readUnreadSteeredMessages } from "./unread-steered-messages.js"; @@ -819,6 +820,25 @@ function sandboxProviderLeaseId(target?: AdapterExecutionTarget | null): string * reports the typed pre-provider failure that sends the next attempt to a * fresh sandbox instead of failing the task. */ +async function retireInProcessWarmOwnerOnSandbox(runId: string, providerLeaseId: string | null): Promise { + for (const [scope, entry] of warmNativeSessions) { + if (entry.runId !== runId || entry.released || (providerLeaseId && entry.providerLeaseId !== providerLeaseId)) continue; + if (entry.busy || entry.preparingRunId) { + throw new NativePredecessorRetirementUnprovenError(runId, providerLeaseId, new Error("runner_port_holder_busy")); + } + if (entry.idleTimer !== null) clearTimeout(entry.idleTimer); + entry.idleTimer = null; + entry.busy = true; + entry.closeOnReleaseReason = "warm Runner held the sandbox Runner port for another session"; + try { + await retireWarmPredecessor(entry, entry.closeOnReleaseReason); + forgetWarmNativeSession(scope, entry); + } finally { entry.busy = false; } + return true; + } + return false; +} + async function retireWarmPredecessor(entry: WarmNativeSession, reason: string, options?: Parameters[2]) { // Chaos cell SR1U: a sandbox owner whose stop stays unconfirmed even when @@ -12045,6 +12065,22 @@ async function executePaperclipNativeSessionAttempt( remoteCleanupLease = await readRemoteCleanupLease(); return true; }; + const portTarget = input.runnerExecutionTarget; + if (input.useRunnerd && !input.backend && !existingWarmSession && portTarget?.kind === "remote" && portTarget.transport === "sandbox" + && portTarget.runner && !portTarget.runnerTask && portTarget.sandboxLeaseAcquisition?.outcome === "resumed" + && input.restartRecovery?.kind !== "reattach_remote_runner" + && input.restartRecovery?.kind !== "reattach_existing_runner") { + const runtimeRoot = posix.join(portTarget.remoteCwd, ".paperclip-runtime", "paperclip-runner"); + const providerLeaseId = portTarget.sandboxLeaseAcquisition?.providerLeaseId ?? null; + const retirePortHolders = () => retireSandboxRunnerPortHolders({ + db: input.db, companyId: input.execution.binding.companyId, agentId: input.execution.binding.agentId, + runId: input.execution.binding.runId, runner: portTarget.runner!, runtimeRoot, + ownSessionRoot: posix.join(runtimeRoot, "sessions", createHash("sha256").update(nativeSessionKey(input.execution)).digest("hex")), + providerLeaseId, adoptedRunId: remoteWarmRecovery?.identity.runId ?? null, + retireInProcess: holderRunId => retireInProcessWarmOwnerOnSandbox(holderRunId, providerLeaseId), onLog: input.onLog, + }); + await trace.measure("runner.port_holders.retire", () => executionContext ? executionContext.wait(retirePortHolders) : retirePortHolders()); + } nativeSessionExecuteStartedAtMs = Date.now(); native = await trace.measure( "native.session.execute", diff --git a/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts b/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts index b19150beafb..042910e3d35 100644 --- a/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts +++ b/server/src/services/native-runtime/native-warm-checkpoint-ownership.integration.test.ts @@ -148,7 +148,9 @@ describeEmbeddedPostgres("managed warm checkpoint ownership", () => { sandboxLeaseAcquisition: { outcome: acquisition, providerLeaseId }, runner: { execute: async (command: { args?: string[] }) => command.args?.[1] === "uname -s; uname -m" ? { exitCode: 0, timedOut: false, stdout: `${process.platform === "darwin" ? "Darwin" : "Linux"}\n${process.arch === "arm64" ? "arm64" : "x86_64"}\n`, stderr: "" } - : { exitCode: 1, timedOut: false, stdout: "", stderr: "" }, syncIn: async () => {} } } as never, + : command.args?.[2] === "paperclip-runner-port-holders" + ? { exitCode: 0, timedOut: false, stdout: "", stderr: "" } + : { exitCode: 1, timedOut: false, stdout: "", stderr: "" }, syncIn: async () => {} } } as never, instructionWorkingCopy: { runId: fixture.runId, root, checkpointWarm: async () => { if (instructionErrorWithDetach) { enteredClose(); await closeGate; throw initiatingError; } if (scenario === "instruction_error") throw initiatingError; diff --git a/server/src/services/native-runtime/sandbox-runner-port-holders.integration.test.ts b/server/src/services/native-runtime/sandbox-runner-port-holders.integration.test.ts new file mode 100644 index 00000000000..7cebee35b59 --- /dev/null +++ b/server/src/services/native-runtime/sandbox-runner-port-holders.integration.test.ts @@ -0,0 +1,140 @@ +import { spawn, spawnSync, type ChildProcess } from "node:child_process"; +import { randomUUID } from "node:crypto"; +import { mkdir, readFile, writeFile } from "node:fs/promises"; +import { createServer, type AddressInfo } from "node:net"; +import { join } from "node:path"; +import { and, eq } from "drizzle-orm"; +import { afterEach, expect, it } from "vitest"; +import { activityLog, heartbeatRuns } from "@paperclipai/db"; +import { describeEmbeddedPostgres, nativeRestartFixture } from "../../__tests__/helpers/native-restart.js"; +import { NativePredecessorRetirementUnprovenError } from "./predecessor-retirement.js"; +import { retireSandboxRunnerPortHolders } from "./sandbox-runner-port-holders.js"; + +const HOLDER_SCRIPT = 'require("node:net").createServer().listen(Number(process.argv[1]), "127.0.0.1", () => process.stdout.write("ready\\n"));'; + +describeEmbeddedPostgres("a sandbox Runner port held by another session's Runner", () => { + const environment = nativeRestartFixture(); + const holders: ChildProcess[] = []; + afterEach(() => { for (const holder of holders.splice(0)) if (holder.exitCode === null) holder.kill("SIGKILL"); }); + + const confinedCommands: string[][] = []; + const confinedRunner = (base: string) => ({ execute: async (request: { command: string; args?: string[] }) => { + const operands = (request.args ?? []).slice(2); + confinedCommands.push(operands); + if (operands.some(operand => operand.startsWith("/") && !operand.startsWith(`${base}/`))) { + return { exitCode: 126, timedOut: false, stdout: "", stderr: "outside sandbox" }; + } + const result = spawnSync(request.command, request.args ?? [], { encoding: "utf8", timeout: 20_000, cwd: base }); + return { exitCode: result.status, timedOut: result.error !== undefined, stdout: result.stdout ?? "", stderr: result.stderr ?? "" }; + } }); + + async function freePort() { + const probe = createServer(); + await new Promise(done => probe.listen(0, "127.0.0.1", done)); + const port = (probe.address() as AddressInfo).port; + await new Promise(done => probe.close(() => done())); + return port; + } + + async function bind(port: number): Promise<"bound" | string> { + return new Promise(done => { + const contender = createServer(); + contender.once("error", (error: NodeJS.ErrnoException) => done(error.code ?? "error")); + contender.listen(port, "127.0.0.1", () => contender.close(() => done("bound"))); + }); + } + + async function startHolder(input: { runtimeRoot: string; session: string; runId: string; port: number; + lifecycle?: string; outbox?: readonly unknown[]; pendingProviderCleanup?: unknown }) { + const runnerInstanceId = `runner-${randomUUID()}`; + const holder = spawn(process.execPath, ["-e", HOLDER_SCRIPT, String(input.port), "--runner-id", runnerInstanceId], + { stdio: ["ignore", "pipe", "ignore"] }); + holders.push(holder); + await new Promise((done, fail) => { + holder.stdout!.once("data", () => done()); + holder.once("exit", code => fail(new Error(`holder exited ${code}`))); + }); + const directory = join(input.runtimeRoot, "sessions", input.session, "runner"); + await mkdir(directory, { recursive: true }); + await writeFile(join(directory, "runner-process.identity"), + `holdernonce\n${holder.pid}\n${new Date().toISOString()}\n${runnerInstanceId}\n${await processFingerprint(holder.pid!)}`); + await writeFile(join(directory, "runner-state.json"), JSON.stringify({ runnerInstanceId, runId: input.runId, + lifecycle: input.lifecycle ?? "ready", outbox: input.outbox ?? [], pendingTerminalDelivery: null, + pendingProviderCleanup: input.pendingProviderCleanup ?? null, warmTransition: null })); + return holder; + } + + async function processFingerprint(pid: number) { + if (process.platform !== "linux") return ""; + const stat = await readFile(`/proc/${pid}/stat`, "utf8"); + const ticks = stat.slice(stat.lastIndexOf(") ") + 2).trim().split(/\s+/)[19]; + const boot = (await readFile("/proc/sys/kernel/random/boot_id", "utf8")).trim(); + return `linux:${boot}:${ticks}\n`; + } + + async function exited(holder: ChildProcess) { + if (holder.exitCode !== null || holder.signalCode !== null) return true; + return new Promise(done => { + const timer = setTimeout(() => done(false), 5_000); + holder.once("exit", () => { clearTimeout(timer); done(true); }); + }); + } + + async function scenario(label: string, holderStatus: "succeeded" | "running" | "cancelled" | null, state: { lifecycle?: string; outbox?: readonly unknown[]; pendingProviderCleanup?: unknown } = {}) { + const predecessor = await environment.seedRun(`PORT-HOLDER-PREV-${label}`); + const current = await environment.seedRun(`PORT-HOLDER-NEXT-${label}`, predecessor.db); + if (holderStatus) await predecessor.db.update(heartbeatRuns).set({ status: holderStatus }).where(eq(heartbeatRuns.id, predecessor.runId)); + const base = join(environment.current.runtimeRoot, `sandbox-${label}`); + const runtimeRoot = join(base, ".paperclip-runtime", "paperclip-runner"); + const port = await freePort(); + const holder = await startHolder({ runtimeRoot, session: "a".repeat(64), runId: holderStatus ? predecessor.runId : randomUUID(), port, ...state }); + const retire = () => retireSandboxRunnerPortHolders({ db: current.db, companyId: environment.current.companyId, + agentId: environment.current.agentId, runId: current.runId, runner: confinedRunner(base) as never, runtimeRoot, + ownSessionRoot: join(runtimeRoot, "sessions", "b".repeat(64)), providerLeaseId: "sandbox-1", + retireInProcess: async () => false, waitMs: 3_000, pollMs: 100 }); + return { predecessor, current, holder, port, retire }; + } + + it("stops a finished run's idle Runner that holds the port, so the next Runner binds", async () => { + const { predecessor, current, holder, port, retire } = await scenario("finished", "succeeded"); + expect(await bind(port)).toBe("EADDRINUSE"); + await expect(retire()).resolves.toEqual([predecessor.runId]); + expect(await exited(holder)).toBe(true); + expect(await bind(port)).toBe("bound"); + expect(confinedCommands.flat().filter(operand => operand.startsWith("/")).every(operand => operand.startsWith(`${environment.current.runtimeRoot}/sandbox-finished/`))).toBe(true); + const [activity] = await current.db.select().from(activityLog).where(and( + eq(activityLog.runId, current.runId), eq(activityLog.action, "native.runner_port_holder_retired"))); + expect(activity?.details).toMatchObject({ holderRunId: predecessor.runId, holderStatus: "succeeded", providerLeaseId: "sandbox-1" }); + }); + + it.each([ + ["a live run", "running", {}, "runner_port_holder_owner_not_settled"], + ["a run with work in its outbox", "succeeded", { outbox: [{ seq: 1 }] }, "runner_port_holder_owner_not_settled"], + ["a run whose provider cleanup is pending", "cancelled", { pendingProviderCleanup: { reason: "fixture" } }, "runner_port_holder_owner_not_settled"], + ["no recorded run", null, {}, "runner_port_holder_unknown_owner"], + ] as const)("never stops a holder owned by %s and sends the run to a fresh sandbox", async (_name, status, state, reason) => { + const { holder, port, retire } = await scenario(`kept-${reason}-${status}`, status, state); + const failure = await retire().then(() => null, (error: unknown) => error); + expect(failure).toBeInstanceOf(NativePredecessorRetirementUnprovenError); + expect(((failure as Error).cause as Error).message).toBe(reason); + expect(holder.exitCode).toBeNull(); + expect(await bind(port)).toBe("EADDRINUSE"); + }); + + it("ignores this run's own session, moved-aside roots and holders that already exited", async () => { + const current = await environment.seedRun("PORT-HOLDER-OWN"); + const base = join(environment.current.runtimeRoot, "sandbox-own"); + const runtimeRoot = join(base, ".paperclip-runtime", "paperclip-runner"); + const own = await startHolder({ runtimeRoot, session: "b".repeat(64), runId: current.runId, port: await freePort() }); + const aside = await startHolder({ runtimeRoot, session: `${"c".repeat(64)}.bootstrap-restart-1`, runId: randomUUID(), port: await freePort() }); + const gone = await startHolder({ runtimeRoot, session: "d".repeat(64), runId: randomUUID(), port: await freePort() }); + gone.kill("SIGKILL"); + expect(await exited(gone)).toBe(true); + await expect(retireSandboxRunnerPortHolders({ db: current.db, companyId: environment.current.companyId, + agentId: environment.current.agentId, runId: current.runId, runner: confinedRunner(base) as never, runtimeRoot, + ownSessionRoot: join(runtimeRoot, "sessions", "b".repeat(64)), providerLeaseId: "sandbox-1", + retireInProcess: async () => false })).resolves.toEqual([]); + expect(own.exitCode).toBeNull(); + expect(aside.exitCode).toBeNull(); + }); +}); diff --git a/server/src/services/native-runtime/sandbox-runner-port-holders.ts b/server/src/services/native-runtime/sandbox-runner-port-holders.ts new file mode 100644 index 00000000000..69361d5dfbf --- /dev/null +++ b/server/src/services/native-runtime/sandbox-runner-port-holders.ts @@ -0,0 +1,163 @@ +import { and, eq } from "drizzle-orm"; +import { heartbeatRuns, type Db } from "@paperclipai/db"; +import type { CommandManagedRuntimeRunner } from "@paperclipai/adapter-utils/command-managed-runtime"; +import { persistActivity } from "../activity-log.js"; +import { NativePredecessorRetirementUnprovenError } from "./predecessor-retirement.js"; +import { REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT } from "./remote-runner-recovery-diagnostics.js"; + +export const SANDBOX_RUNNER_HOLDER_LIST_SCRIPT = [ + "set -eu", + 'root=$1; own=$2', + 'test -d "$root/sessions" && test ! -L "$root/sessions" || exit 0', + 'for session in "$root"/sessions/*; do', + ' test -d "$session" && test ! -L "$session" || continue', + ' case "${session##*/}" in *.*) continue;; esac', + ' test "$session" = "$own" && continue', + ' identity="$session/runner/runner-process.identity"; state="$session/runner/runner-state.json"', + ' test -f "$identity" && test ! -L "$identity" || continue', + ' encoded_state=""', + ' if test -f "$state" && test ! -L "$state"; then encoded_state=$(base64 < "$state" | tr -d "\\n"); fi', + ' printf "%s\\t%s\\t%s\\n" "$identity" "$(base64 < "$identity" | tr -d "\\n")" "$encoded_state"', + "done", +].join("\n"); + +const RETIRABLE_HOLDER_RUN_STATUSES = new Set(["succeeded", "cancelled"]); +const RUN_ID_PATTERN = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i; + +export type SandboxRunnerHolder = { + identityPath: string; + nonce: string; + pid: number; + runnerInstanceId: string; + processStartFingerprint: string | null; + runId: string | null; + idle: boolean; +}; + +function decode(value: string | undefined): string | null { + if (!value) return null; + try { return Buffer.from(value, "base64").toString("utf8"); } catch { return null; } +} + +function parseObject(value: string | null): Record | null { + if (!value) return null; + try { + const parsed: unknown = JSON.parse(value); + return parsed !== null && typeof parsed === "object" && !Array.isArray(parsed) ? parsed as Record : null; + } catch { return null; } +} + +export function parseSandboxRunnerHolders(stdout: string): SandboxRunnerHolder[] { + const holders: SandboxRunnerHolder[] = []; + for (const line of stdout.split("\n")) { + const [identityPath, encodedIdentity, encodedState] = line.split("\t"); + const identity = decode(encodedIdentity)?.trim().split("\n"); + if (!identityPath || !identity) continue; + const [nonce, rawPid, startedAt, runnerInstanceId, fingerprint] = identity; + const pid = Number(rawPid); + if (!nonce || !/^[a-zA-Z0-9_-]+$/.test(nonce) || !Number.isSafeInteger(pid) || pid <= 0 || !startedAt + || !runnerInstanceId || (fingerprint !== undefined && !/^linux:[0-9a-f-]+:[0-9]+$/.test(fingerprint))) continue; + const state = parseObject(decode(encodedState)); + const runId = typeof state?.runId === "string" && state.runnerInstanceId === runnerInstanceId ? state.runId : null; + const idle = state !== null && Array.isArray(state.outbox) && state.outbox.length === 0 + && state.pendingTerminalDelivery === null && state.pendingProviderCleanup == null && state.warmTransition == null; + holders.push({ identityPath, nonce, pid, runnerInstanceId, processStartFingerprint: fingerprint ?? null, runId, idle }); + } + return holders; +} + +type HolderProbe = "alive" | "gone" | "unproven"; + +async function probeHolder(runner: CommandManagedRuntimeRunner, holder: SandboxRunnerHolder, + signal?: "TERM" | "KILL"): Promise { + const result = await runner.execute({ + command: "sh", + args: ["-c", REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT + (signal ? '; kill -"$6" "$pid"' : ""), + "paperclip-runner-port-holder-check", holder.identityPath, holder.nonce, holder.runnerInstanceId, String(holder.pid), + holder.processStartFingerprint ?? "", ...(signal ? [signal] : [])], + bypassSession: true, + timeoutMs: 10_000, + }).catch(() => null); + if (!result || result.timedOut) return "unproven"; + if (result.exitCode === 0) return "alive"; + if (result.exitCode === 45 || result.exitCode === 47) return "gone"; + return "unproven"; +} + +async function waitUntilGone(runner: CommandManagedRuntimeRunner, holder: SandboxRunnerHolder, waitMs: number, pollMs: number) { + const deadline = Date.now() + waitMs; + for (;;) { + const probe = await probeHolder(runner, holder); + if (probe !== "alive") return probe; + if (Date.now() >= deadline) return "alive" as const; + await new Promise(done => setTimeout(done, pollMs)); + } +} + +export async function retireSandboxRunnerPortHolders(input: { + db: Db; + companyId: string; + agentId: string; + runId: string; + runner: CommandManagedRuntimeRunner; + runtimeRoot: string; + ownSessionRoot: string; + providerLeaseId: string | null; + adoptedRunId?: string | null; + retireInProcess: (holderRunId: string) => Promise; + onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise; + waitMs?: number; + pollMs?: number; +}): Promise { + const listed = await input.runner.execute({ + command: "sh", + args: ["-c", SANDBOX_RUNNER_HOLDER_LIST_SCRIPT, "paperclip-runner-port-holders", input.runtimeRoot, input.ownSessionRoot], + bypassSession: true, + timeoutMs: 15_000, + }).catch(() => null); + if (!listed || listed.exitCode !== 0 || listed.timedOut) { + throw new NativePredecessorRetirementUnprovenError(null, input.providerLeaseId, + new Error(`runner_port_holder_inventory_unavailable: exit=${listed?.exitCode ?? "none"} timedOut=${listed?.timedOut ?? false}`)); + } + const retired: string[] = []; + for (const holder of parseSandboxRunnerHolders(listed.stdout)) { + if (holder.runId === input.runId || (holder.runId !== null && holder.runId === input.adoptedRunId)) continue; + const initial = await probeHolder(input.runner, holder); + if (initial === "gone") continue; + const unproven = (reason: string) => new NativePredecessorRetirementUnprovenError(holder.runId, input.providerLeaseId, + new Error(`runner_port_holder_${reason}`)); + if (initial === "unproven") throw unproven("unverifiable"); + if (holder.runId && await input.retireInProcess(holder.runId)) { + if (await waitUntilGone(input.runner, holder, input.waitMs ?? 10_000, input.pollMs ?? 500) !== "gone") throw unproven("still_running"); + retired.push(holder.runId); + continue; + } + if (!holder.runId || !RUN_ID_PATTERN.test(holder.runId)) throw unproven("unknown_owner"); + const [owner] = await input.db.select({ status: heartbeatRuns.status, agentId: heartbeatRuns.agentId }).from(heartbeatRuns) + .where(and(eq(heartbeatRuns.companyId, input.companyId), eq(heartbeatRuns.id, holder.runId))).limit(1); + if (!owner) throw unproven("unknown_owner"); + if (!RETIRABLE_HOLDER_RUN_STATUSES.has(owner.status) || owner.agentId !== input.agentId || !holder.idle) { + throw unproven("owner_not_settled"); + } + let outcome: "stopped" | "killed" = "stopped"; + let probe = await probeHolder(input.runner, holder, "TERM"); + if (probe === "alive") probe = await waitUntilGone(input.runner, holder, input.waitMs ?? 10_000, input.pollMs ?? 500); + if (probe === "alive") { + outcome = "killed"; + probe = await probeHolder(input.runner, holder, "KILL"); + if (probe === "alive") probe = await waitUntilGone(input.runner, holder, input.waitMs ?? 10_000, input.pollMs ?? 500); + } + if (probe !== "gone") throw unproven("stop_unconfirmed"); + await persistActivity(input.db, { + companyId: input.companyId, actorType: "system", actorId: "native-runtime", agentId: input.agentId, + action: "native.runner_port_holder_retired", entityType: "heartbeat_run", entityId: input.runId, runId: input.runId, + details: { holderRunId: holder.runId, holderRunnerInstanceId: holder.runnerInstanceId, holderStatus: owner.status, outcome, + providerLeaseId: input.providerLeaseId }, + }); + await input.onLog?.("stdout", + `[paperclip-runner] Stopped the finished run ${holder.runId}'s idle Runner that still held this sandbox's Runner port (${outcome}).\n`) + .catch(() => undefined); + retired.push(holder.runId); + } + return retired; +}