mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-13 07:12:00 +01:00
fix(native): retire a finished run's idle Runner that still holds the sandbox Runner port before launching (#20995)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Native runs execute through a Runner (`paperclip-runnerd`) inside a remote sandbox. Daytona reaches every Runner through one fixed preview port (43127), so a sandbox serves one listening Runner at a time. > - A warm sandbox is reused by the next run of the same task. That run often opens a new session, so it neither adopts the previous run's warm Runner nor retires it. Retirement of another session's warm owner works only through the in-memory warm registry of the server that launched it. > - After a server restart (or on another server), that registry is empty. The previous run's idle warm Runner keeps the port, and every new launch fails with `runner_ingress_bind_conflict` until the bounded retry runs out. > - This pull request finds such a holder on the sandbox before launch and proves from durable records whether it may be stopped. A finished run's idle Runner is stopped through its verified process identity. Any other holder sends the run to the fresh-sandbox fallback from #20863 and is never stopped. > - The benefit is that a follow-up run after a restart starts on its reused sandbox instead of colliding with its predecessor and needing Retry. ## Linked Issues or Issue Description **What happened?** Restart-chaos wave `w-20261111T230537-s6402` (Claude through ACPX, Daytona warm sandboxes, `--ephemeral-server`, K1/K2/K15 with `--k15-fresh-home`, seed 6402, master, no injected fault) needed Retry on 4 of 6 tasks. Each failing run's first Runner failed with `runner_ingress_bind_conflict: failed to bind listener: Address already in use`. The harness's sandbox captures name the process holding port 43127 at the time. In every case it was the previous, already `succeeded` run of the same task, started under the server generation before the restart: | Task | Failing run | Port 43127 held by | Holder run status | | --- | --- | --- | --- | | t01 | b4769dfd | 395b202d (Runner started 23:11:30) | succeeded 23:12:19 | | t05 | b6d466fd | 0969ad46 (23:11:29) | succeeded 23:12:18 | | t02 | 4097fc1f | 527a1659 (23:11:28) | succeeded 23:12:17 | | t03 | 60d821af | 1c06bf75 (23:11:28) | succeeded 23:12:25 | **Root cause** Warm owners are tracked only in the process-local `warmNativeSessions` map in `server/src/services/native-runtime/native-session-executor.ts`. `claimWarmNativeInstructionCopy`, `reserveWarmNativeInstructionDirectory` and the same-session admission path retire a predecessor only when it is in that map. A restarted server has an empty map. The follow-up run uses a new session, so `inspectRemoteWarmSessionRecovery` (same session only) does not adopt it either. The launch then meets the old Runner on the fixed port. `stopStaleRemoteRunnerBeforeBootstrap` stops only the run's own Runner, so the same-run retries collide again. **Expected behavior** A finished predecessor's idle warm Runner is retired before the next Runner launches on the same sandbox, wherever it was launched. A holder that cannot be proven safe to stop is never stopped, and the run moves to a fresh sandbox. ## What Changed - New `server/src/services/native-runtime/sandbox-runner-port-holders.ts`: `retireSandboxRunnerPortHolders` lists every other session's Runner on the sandbox (session roots with a `runner-process.identity`; this run's own root and moved-aside roots are skipped). It checks each one with the existing verified process check (`REMOTE_RECOVERY_PROCESS_CHECK_SCRIPT`: pid, runner id in the command line and, when recorded, the boot generation). Then, for a live holder: - Registered in this server's warm registry: it is retired through `retireWarmPredecessor`. - Its company-scoped run row is `succeeded` or `cancelled`, for the same agent, and its Runner state is idle (empty outbox, no pending terminal delivery, and both saved fences `pendingProviderCleanup` and `warmTransition` clear): it is stopped through the verified identity (`TERM`, then `KILL`, each with a bounded wait). An activity log entry `native.runner_port_holder_retired` is written. - When the inventory command itself cannot run, the holder state is unproven, so the run takes the fresh-sandbox fallback. - Anything else (a live run, a failed or interrupted run that may still be resumed, an unknown owner, work in the outbox, an unverifiable or changed identity marker, or a stop that is not confirmed): `NativePredecessorRetirementUnprovenError`. Before any process start, the existing #20863 path marks the lease replace-before-reuse and re-queues the run in the uncharged lane on a fresh sandbox. - `executePaperclipNativeSessionAttempt` runs this check before `native.process_start_requested`, and only for a resumed (reused) sandbox launched through a shell runner without a warm session to reuse and without a reattach recovery. A created or replacement sandbox cannot hold another session's Runner, so it adds no command there. ## Verification - `pnpm --filter @paperclipai/server exec vitest run src/services/native-runtime/sandbox-runner-port-holders.integration.test.ts` (embedded Postgres, real processes): two Runners contend for one real port. The second bind first fails with `EADDRINUSE`, the same failure as staging. After `retireSandboxRunnerPortHolders` the finished run's holder has exited and the bind succeeds, with the activity entry recorded. The commands run through a confined runner that refuses any path outside the sandbox workspace base. A holder owned by a live run, by a finished run with work in its outbox, by a cancelled run with pending provider cleanup, or by no recorded run is never stopped: the call raises `NativePredecessorRetirementUnprovenError` (`runner_port_holder_owner_not_settled` / `runner_port_holder_unknown_owner`) and the port stays held. This run's own session, moved-aside roots and exited holders are ignored. - `native-warm-checkpoint-ownership.integration.test.ts`: its fixture sandbox runner now answers the inventory command with an empty list (that sandbox has no other Runner); all 9 cells pass. - `native-session-executor.test.ts`: 1406 of 1406 pass. Server `tsc --noEmit` passes. - Chaos results are in a comment below. ## Risks - A finished run's idle warm Runner is stopped earlier than its idle timeout. It held no unsaved work (its run is `succeeded` or `cancelled` and its outbox is empty), and SIGTERM is the same stop as its idle close. Only that warm reuse is lost. - Failed and interrupted holders are not stopped, because their run may still reattach. Such a run goes to a fresh sandbox instead. - One extra sandbox command before launch on a reused sandbox for a cold launch, no wait and no file changes. A warm reuse and a reattach skip it. - Not changed: a bind conflict from a holder that is not a Runner (no identity file) still uses the existing same-run bootstrap budget. ## Model Used - Claude Opus 5.5 (`claude-opus-5-5`), Anthropic, through Claude Code with tool use (shell, file edit, test runs). Default context window, extended reasoning on. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Refs #20863, #20953, #20972, #20992. 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
DottaandClaude Opus 5.5 authored and GitHub committed
1 parent
7d9973405c
commit
776e900b48
4 files changed
+342
-1
No files matched your search
@@ -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<boolean> {
|
||||
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<typeof closeWarmNativeSession>[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",
|
||||
|
||||
+3
-1
@@ -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;
|
||||
|
||||
@@ -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<void>(done => probe.listen(0, "127.0.0.1", done));
|
||||
const port = (probe.address() as AddressInfo).port;
|
||||
await new Promise<void>(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<void>((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<boolean>(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();
|
||||
});
|
||||
});
|
||||
@@ -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<string, unknown> | null {
|
||||
if (!value) return null;
|
||||
try {
|
||||
const parsed: unknown = JSON.parse(value);
|
||||
return parsed !== null && typeof parsed === "object" && !Array.isArray(parsed) ? parsed as Record<string, unknown> : 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<HolderProbe> {
|
||||
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<boolean>;
|
||||
onLog?: (stream: "stdout" | "stderr", chunk: string) => Promise<void>;
|
||||
waitMs?: number;
|
||||
pollMs?: number;
|
||||
}): Promise<string[]> {
|
||||
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;
|
||||
}
|
||||
Reference in new issue
Block a user