Integrate frozen work-folders candidate into isolated qualification preview
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
Dotta
commit
bf77f89cad
|
|
@ -272,6 +272,16 @@ export interface RunnerProcessHandle {
|
|||
kill(signal?: NodeJS.Signals | number): boolean;
|
||||
};
|
||||
completion: Promise<RunnerProcessResult>;
|
||||
/** Async launch identity AND launcher-side ownership persistence. Consumers
|
||||
* must await this before admission and must not publish ownership a second time.
|
||||
* Reject only after attempted containment and observe rejection immediately. */
|
||||
ready?: Promise<void>;
|
||||
/** Latch cancellation before async launch/readiness; prevent a later dispatch
|
||||
* or contain an already dispatched process through launcher-owned authority. */
|
||||
cancelPendingLaunch?(): void;
|
||||
/** Exact ready/completion rejection and the launcher's verified containment
|
||||
* outcome. Never infer containment from a rejected lifetime promise alone. */
|
||||
ownershipFailure?: { error: unknown; containment: "confirmed" | "unconfirmed" };
|
||||
processGroupId?: number | null;
|
||||
startedAt?: string;
|
||||
/** Relaunches the same immutable process specification with a fresh ticket. */
|
||||
|
|
@ -3513,14 +3523,23 @@ export function spawnRunner(options: {
|
|||
|
||||
const command = options.runnerBinaryPath ?? runnerBinary;
|
||||
const environment = runnerEnvironment(options.ticket, options.environment);
|
||||
const withRestart = (handle: RunnerProcessHandle): RunnerProcessHandle => ({
|
||||
...handle,
|
||||
// Remote launchers resolve process identity after returning the handle.
|
||||
get startedAt() {
|
||||
return handle.startedAt;
|
||||
},
|
||||
restart: (ticket) => spawnRunner({ ...options, ticket }),
|
||||
});
|
||||
const withRestart = (handle: RunnerProcessHandle): RunnerProcessHandle => {
|
||||
// Consumers await the original promises, but a remote launch can reject
|
||||
// before that await is installed. Observe without replacing their results.
|
||||
void handle.ready?.catch(() => undefined);
|
||||
void handle.completion.catch(() => undefined);
|
||||
return {
|
||||
...handle,
|
||||
// Remote launchers resolve process identity after returning the handle.
|
||||
get startedAt() {
|
||||
return handle.startedAt;
|
||||
},
|
||||
get ownershipFailure() {
|
||||
return handle.ownershipFailure;
|
||||
},
|
||||
restart: (ticket) => spawnRunner({ ...options, ticket }),
|
||||
};
|
||||
};
|
||||
if (options.processLauncher !== undefined) {
|
||||
return withRestart(
|
||||
options.processLauncher({ command, args, cwd: packageRoot, environment }),
|
||||
|
|
|
|||
|
|
@ -2085,9 +2085,178 @@ it("publishes spawned runner ownership before waiting for provider startup", asy
|
|||
}
|
||||
});
|
||||
|
||||
it.each([false, true])("persists each recovered runner identity before declaring recovery complete (save fails: %s)", async (failOwnershipSave) => {
|
||||
it.each([
|
||||
{ phase: "initial", rejectSave: true, stop: "none" },
|
||||
{ phase: "replacement", rejectSave: true, stop: "none" },
|
||||
{ phase: "initial", rejectSave: false, stop: "none" },
|
||||
{ phase: "initial", rejectSave: false, stop: "close" },
|
||||
{ phase: "replacement", rejectSave: false, stop: "deadline" },
|
||||
])("fences asynchronous $phase ownership admission (save rejects: $rejectSave, stop: $stop)", async ({ phase, rejectSave, stop }) => {
|
||||
const stateDirectory = await mkdtemp(join(tmpdir(), "runner-async-ownership-"));
|
||||
const handles: durableControlPlane.RunnerProcessHandle[] = [];
|
||||
let rejectOwnership!: () => void;
|
||||
const barrier = new Promise<void>((resolve) => { rejectOwnership = resolve; });
|
||||
let readyRejected = false;
|
||||
const launcherOwnershipSave = vi.fn();
|
||||
const transportOwnershipSave = vi.fn();
|
||||
const realSpawn = durableControlPlane.spawnRunner;
|
||||
const wrap = (handle: durableControlPlane.RunnerProcessHandle): durableControlPlane.RunnerProcessHandle => {
|
||||
handles.push(handle);
|
||||
const selected = handles.length === (phase === "initial" ? 1 : 2);
|
||||
if (!selected) return { ...handle, restart: handle.restart ? (ticket) => wrap(handle.restart!(ticket)) : undefined };
|
||||
let publishedPid: number | undefined;
|
||||
let ownershipFailure: durableControlPlane.RunnerProcessHandle["ownershipFailure"];
|
||||
let cancelLaunch!: (error: Error) => void;
|
||||
const cancelled = new Promise<never>((_resolve, reject) => { cancelLaunch = reject; });
|
||||
void cancelled.catch(() => undefined);
|
||||
const ready = (async () => {
|
||||
try { await Promise.race([barrier, cancelled]); }
|
||||
catch (error) {
|
||||
handle.child.kill("SIGKILL");
|
||||
await handle.completion;
|
||||
ownershipFailure = { error, containment: "confirmed" };
|
||||
throw error;
|
||||
}
|
||||
publishedPid = handle.child.pid;
|
||||
launcherOwnershipSave({ pid: publishedPid });
|
||||
if (!rejectSave) return;
|
||||
// Simulate remote ownership persistence rejecting after its own process
|
||||
// cleanup. The transport must independently retire its authority route.
|
||||
handle.child.kill("SIGKILL");
|
||||
await handle.completion;
|
||||
readyRejected = true;
|
||||
const error = new Error("fixture async ownership unavailable");
|
||||
ownershipFailure = { error, containment: "confirmed" };
|
||||
throw error;
|
||||
})();
|
||||
void ready.catch(() => undefined);
|
||||
const completion = ready.then(() => handle.completion);
|
||||
void completion.catch(() => undefined);
|
||||
return {
|
||||
...handle, ready, completion, processGroupId: null,
|
||||
...(stop === "deadline" ? { cancelPendingLaunch: () => cancelLaunch(new Error("fixture launch cancelled")) } : {}),
|
||||
get ownershipFailure() { return ownershipFailure; },
|
||||
child: { get pid() { return publishedPid; }, get exitCode() { return handle.child.exitCode; }, kill: (signal) => handle.child.kill(signal) },
|
||||
restart: handle.restart ? (ticket) => wrap(handle.restart!(ticket)) : undefined,
|
||||
};
|
||||
};
|
||||
const spawnSpy = vi.spyOn(durableControlPlane, "spawnRunner").mockImplementation((options) => wrap(realSpawn(options)));
|
||||
const diagnostics: string[] = [];
|
||||
const releaseRoute = vi.fn();
|
||||
const activate = vi.fn();
|
||||
let authority!: DurablePrpControlPlane;
|
||||
let connectionAttempts = 0;
|
||||
const bundle = createCapabilityRunnerdCodexTransport({
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex,
|
||||
codexArgs: fakeCodexArgs(stateDirectory), stateDirectory,
|
||||
lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 }, runnerReconnectGraceMs: stop === "deadline" ? 500 : 5_000,
|
||||
closeGraceMs: 1_000,
|
||||
onSpawn: transportOwnershipSave,
|
||||
onDiagnostic: (message) => diagnostics.push(message),
|
||||
controlPlaneRegistration: async (core) => {
|
||||
authority = core;
|
||||
const upgrade = core.handleUpgrade.bind(core);
|
||||
vi.spyOn(core, "handleUpgrade").mockImplementation((...args) => { connectionAttempts += 1; return upgrade(...args); });
|
||||
await core.start(); return { activate, release: releaseRoute };
|
||||
},
|
||||
});
|
||||
let openingSettled = false;
|
||||
const opening = bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs() });
|
||||
void opening.then(() => { openingSettled = true; }, () => { openingSettled = true; });
|
||||
try {
|
||||
if (phase === "replacement") {
|
||||
await opening;
|
||||
handles[0]!.child.kill("SIGKILL");
|
||||
authority.queueCommand("runner.drain", {}, "pending-before-ownership");
|
||||
}
|
||||
const expectedHandles = phase === "initial" ? 1 : 2;
|
||||
await vi.waitFor(() => expect(handles).toHaveLength(expectedHandles));
|
||||
// Real runner handshakes arrive while PID publication/persistence is
|
||||
// pending. They must not consume bootstrap credentials or replay work.
|
||||
await vi.waitFor(() => expect(connectionAttempts).toBeGreaterThanOrEqual(expectedHandles));
|
||||
await new Promise((resolve) => setTimeout(resolve, 75));
|
||||
expect(readyRejected).toBe(false);
|
||||
expect(authority.store.state.connectionCount).toBe(expectedHandles - 1);
|
||||
if (phase === "replacement") expect(authority.store.state.commands.find((command) => command.commandId === "pending-before-ownership")?.status).toBe("pending");
|
||||
else expect(authority.store.state.commands.every((command) => command.status === "pending")).toBe(true);
|
||||
if (phase === "initial") { expect(openingSettled).toBe(false); expect(activate).not.toHaveBeenCalled(); }
|
||||
expect(diagnostics).not.toContain("runner process restored its durable PRP session");
|
||||
if (stop !== "none") {
|
||||
if (stop === "close") {
|
||||
const closing = bundle.transport.close().catch(() => undefined);
|
||||
await vi.waitFor(() => expect(openingSettled).toBe(true), { timeout: 1_000 });
|
||||
await closing;
|
||||
} else {
|
||||
await vi.waitFor(() => expect(diagnostics.some((message) => message.includes("runner_ownership_admission_deadline_exceeded"))).toBe(true), { timeout: 4_000 });
|
||||
}
|
||||
rejectOwnership(); // A late successful readiness must not revive admission.
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
expect(activate).toHaveBeenCalledTimes(phase === "initial" ? 0 : 1);
|
||||
expect(authority.activeRunnerConnectionCount()).toBe(0);
|
||||
expect(diagnostics).not.toContain("runner process restored its durable PRP session");
|
||||
expect(handles).toHaveLength(expectedHandles);
|
||||
if (stop === "deadline") expect(diagnostics.some((message) => message.includes("cleanup incomplete"))).toBe(false);
|
||||
expect(() => process.kill(handles[expectedHandles - 1]!.child.pid!, 0)).toThrow();
|
||||
return;
|
||||
}
|
||||
rejectOwnership();
|
||||
if (!rejectSave) {
|
||||
await opening;
|
||||
expect(launcherOwnershipSave).toHaveBeenCalledExactlyOnceWith({ pid: handles[0]!.child.pid });
|
||||
expect(transportOwnershipSave).not.toHaveBeenCalled();
|
||||
expect(activate).toHaveBeenCalledOnce();
|
||||
expect(authority.activeRunnerConnectionCount()).toBe(1);
|
||||
return;
|
||||
}
|
||||
await vi.waitFor(() => expect(diagnostics.some((message) => message.includes("native_runner_process_ownership_failed: fixture async ownership unavailable"))).toBe(true));
|
||||
expect(launcherOwnershipSave).toHaveBeenCalledOnce();
|
||||
expect(transportOwnershipSave).toHaveBeenCalledTimes(phase === "initial" ? 0 : 1);
|
||||
expect(diagnostics).toContain("native_runner_process_ownership_failed: fixture async ownership unavailable");
|
||||
expect(diagnostics.some((message) => message.includes("cleanup incomplete"))).toBe(false);
|
||||
expect(authority.activeRunnerConnectionCount()).toBe(0);
|
||||
expect(() => authority.connectUrl).toThrow("not listening");
|
||||
expect(releaseRoute).toHaveBeenCalledOnce();
|
||||
expect(handles).toHaveLength(expectedHandles);
|
||||
expect((await stat(join(stateDirectory, "control-plane"))).isDirectory()).toBe(true);
|
||||
await expect(bundle.transport.request("thread/start", { cwd: tmpdir() })).rejects.toThrow("native_runner_process_ownership_failed");
|
||||
} finally {
|
||||
rejectOwnership();
|
||||
await opening.catch(() => undefined);
|
||||
await bundle.transport.close().catch(() => undefined);
|
||||
for (const handle of handles) {
|
||||
if (process.platform !== "win32" && handle.processGroupId) { try { process.kill(-handle.processGroupId, "SIGKILL"); } catch {} }
|
||||
else handle.child.kill("SIGKILL");
|
||||
await handle.completion.catch(() => undefined);
|
||||
}
|
||||
spawnSpy.mockRestore();
|
||||
await rm(stateDirectory, { recursive: true, force: true });
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it.each([
|
||||
{ failOwnershipSave: false, parentExitsBeforeSave: false },
|
||||
{ failOwnershipSave: true, parentExitsBeforeSave: false },
|
||||
{ failOwnershipSave: true, parentExitsBeforeSave: true },
|
||||
])("persists each recovered runner identity before declaring recovery complete (save fails: $failOwnershipSave, parent exited: $parentExitsBeforeSave)", async ({ failOwnershipSave, parentExitsBeforeSave }) => {
|
||||
const stateDirectory = await mkdtemp(join(tmpdir(), "runner-replacement-ownership-"));
|
||||
const handles: durableControlPlane.RunnerProcessHandle[] = [];
|
||||
const descendantFile = join(stateDirectory, "runner-descendants.txt");
|
||||
const runnerWrapper = join(stateDirectory, "runner-with-descendant.sh");
|
||||
const shellQuote = (value: string) => "'" + value.replaceAll("'", "'\\''") + "'";
|
||||
if (process.platform !== "win32") {
|
||||
await writeFile(runnerWrapper, `#!/bin/sh
|
||||
sleep 600 &
|
||||
printf '%s:%s\\n' "$$" "$!" >> ${shellQuote(descendantFile)}
|
||||
exec ${shellQuote(defaultCapabilityRunnerdBinary())} "$@"
|
||||
`, { mode: 0o700 });
|
||||
}
|
||||
const descendantRunning = (pid: number) => {
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
const state = execFileSync("ps", ["-o", "stat=", "-p", String(pid)], { encoding: "utf8" }).trim();
|
||||
return state.length > 0 && !state.startsWith("Z");
|
||||
} catch { return false; }
|
||||
};
|
||||
const realSpawn = durableControlPlane.spawnRunner;
|
||||
const wrap = (handle: durableControlPlane.RunnerProcessHandle): durableControlPlane.RunnerProcessHandle => {
|
||||
const wrapped = {
|
||||
|
|
@ -2098,7 +2267,10 @@ it.each([false, true])("persists each recovered runner identity before declaring
|
|||
handles.push(wrapped);
|
||||
return wrapped;
|
||||
};
|
||||
const spawnSpy = vi.spyOn(durableControlPlane, "spawnRunner").mockImplementation((options) => wrap(realSpawn(options)));
|
||||
const spawnSpy = vi.spyOn(durableControlPlane, "spawnRunner").mockImplementation((options) => wrap(realSpawn({
|
||||
...options,
|
||||
...(process.platform !== "win32" ? { runnerBinaryPath: runnerWrapper } : {}),
|
||||
})));
|
||||
let releaseOwnership!: () => void;
|
||||
const ownershipBarrier = new Promise<void>((resolve) => { releaseOwnership = resolve; });
|
||||
let durableOwner: { pid: number; processGroupId: number | null; startedAt: string } | null = null;
|
||||
|
|
@ -2110,12 +2282,19 @@ it.each([false, true])("persists each recovered runner identity before declaring
|
|||
durableOwner = structuredClone(owner);
|
||||
});
|
||||
const diagnostics: string[] = [];
|
||||
let recoveryAuthority!: DurablePrpControlPlane;
|
||||
const releaseRoute = vi.fn();
|
||||
const bundle = createCapabilityRunnerdCodexTransport({
|
||||
runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex,
|
||||
codexArgs: fakeCodexArgs(stateDirectory), stateDirectory,
|
||||
lifecyclePolicy: { mode: "warm", idleTimeoutMs: 60_000 },
|
||||
runnerReconnectGraceMs: 5_000, onSpawn,
|
||||
onDiagnostic: (message) => diagnostics.push(message),
|
||||
controlPlaneRegistration: async (authority) => {
|
||||
recoveryAuthority = authority;
|
||||
await authority.start();
|
||||
return { release: releaseRoute };
|
||||
},
|
||||
});
|
||||
try {
|
||||
await bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs() });
|
||||
|
|
@ -2124,6 +2303,23 @@ it.each([false, true])("persists each recovered runner identity before declaring
|
|||
await vi.waitFor(() => expect(onSpawn).toHaveBeenCalledTimes(2));
|
||||
expect(durableOwner).toEqual(originalOwner);
|
||||
expect(diagnostics).not.toContain("runner process restored its durable PRP session");
|
||||
let replacementDescendant: number | null = null;
|
||||
if (process.platform !== "win32") {
|
||||
await vi.waitFor(async () => {
|
||||
const rows = (await readFile(descendantFile, "utf8")).trim().split("\n");
|
||||
const row = rows.find((line) => line.startsWith(`${handles[1]!.child.pid}:`));
|
||||
expect(row).toBeDefined();
|
||||
replacementDescendant = Number(row!.split(":")[1]);
|
||||
expect(descendantRunning(replacementDescendant)).toBe(true);
|
||||
});
|
||||
}
|
||||
if (parentExitsBeforeSave) {
|
||||
handles[1]!.child.kill("SIGKILL");
|
||||
await handles[1]!.completion;
|
||||
// The detached group can outlive its leader. Rejection must still fence
|
||||
// that group's descendant even though completion is already resolved.
|
||||
if (replacementDescendant !== null) expect(descendantRunning(replacementDescendant)).toBe(true);
|
||||
}
|
||||
releaseOwnership();
|
||||
if (failOwnershipSave) {
|
||||
await vi.waitFor(() => expect(diagnostics).toContain("native_runner_process_ownership_failed: fixture durable ownership unavailable"));
|
||||
|
|
@ -2131,6 +2327,14 @@ it.each([false, true])("persists each recovered runner identity before declaring
|
|||
expect(diagnostics).not.toContain("runner process restored its durable PRP session");
|
||||
await expect(bundle.transport.request("thread/start", { cwd: tmpdir() })).rejects.toThrow("native_runner_process_ownership_failed");
|
||||
expect(handles).toHaveLength(2);
|
||||
// Assert fencing before the test's finally calls transport.close(): a
|
||||
// terminal failure alone must not leave an unowned runner authorized.
|
||||
expect(() => process.kill(handles[1]!.child.pid!, 0)).toThrow();
|
||||
if (replacementDescendant !== null) expect(descendantRunning(replacementDescendant)).toBe(false);
|
||||
expect(recoveryAuthority.activeRunnerConnectionCount()).toBe(0);
|
||||
expect(() => recoveryAuthority.connectUrl).toThrow("not listening");
|
||||
expect(releaseRoute).toHaveBeenCalledOnce();
|
||||
expect((await stat(join(stateDirectory, "control-plane"))).isDirectory()).toBe(true);
|
||||
return;
|
||||
}
|
||||
await vi.waitFor(() => expect(diagnostics).toContain("runner process restored its durable PRP session"));
|
||||
|
|
@ -2148,7 +2352,9 @@ it.each([false, true])("persists each recovered runner identity before declaring
|
|||
releaseOwnership();
|
||||
await bundle.transport.close().catch(() => undefined);
|
||||
for (const handle of handles) {
|
||||
if (handle.child.exitCode === null) handle.child.kill("SIGKILL");
|
||||
if (process.platform !== "win32" && handle.processGroupId) {
|
||||
try { process.kill(-handle.processGroupId, "SIGKILL"); } catch { /* Already exited. */ }
|
||||
} else if (handle.child.exitCode === null) handle.child.kill("SIGKILL");
|
||||
await handle.completion.catch(() => undefined);
|
||||
}
|
||||
spawnSpy.mockRestore();
|
||||
|
|
@ -2156,6 +2362,25 @@ it.each([false, true])("persists each recovered runner identity before declaring
|
|||
}
|
||||
}, 15_000);
|
||||
|
||||
it("delegates unpersisted custom runner containment without host-signalling its reported process group", () => {
|
||||
const kill = vi.fn(() => true);
|
||||
const hostSignal = vi.spyOn(process, "kill").mockImplementation(() => {
|
||||
throw new Error("remote PID must not be signalled on this host");
|
||||
});
|
||||
try {
|
||||
const handle = {
|
||||
child: { pid: 12345, kill },
|
||||
processGroupId: 12345,
|
||||
completion: Promise.resolve({ code: 0, signal: null }),
|
||||
} as unknown as durableControlPlane.RunnerProcessHandle;
|
||||
runnerdRecoveryInternals.signalUnpersistedRunner(handle, true);
|
||||
expect(kill).toHaveBeenCalledExactlyOnceWith("SIGKILL");
|
||||
expect(hostSignal).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
hostSignal.mockRestore();
|
||||
}
|
||||
});
|
||||
|
||||
it("launches runnerd with its production durable outbox limits", () => {
|
||||
expect(runnerdLaunchProfileInternals.maxOutboxBytes).toBe(16 * 1024 * 1024);
|
||||
expect(runnerdLaunchProfileInternals.p0ReserveBytes).toBe(1024 * 1024);
|
||||
|
|
|
|||
|
|
@ -3337,6 +3337,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
readonly #failureSignal: Promise<never>;
|
||||
#rejectFailureSignal!: (error: Error) => void;
|
||||
#runnerRecoveryInProgress = false;
|
||||
#runnerOwnershipAdmission: Promise<void> | null = null;
|
||||
#runnerOwnershipPendingHandle: RunnerProcessHandle | null = null;
|
||||
readonly #runnerOwnershipAbort = new AbortController();
|
||||
readonly #cancelledRunnerLaunches = new WeakSet<RunnerProcessHandle>();
|
||||
#startupComplete = false;
|
||||
#startupFailureCode = "native_runner_process_exited";
|
||||
#controlPlaneCheckpoint:
|
||||
|
|
@ -3935,20 +3939,158 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
};
|
||||
}
|
||||
|
||||
async #publishSpawnedProcess(handle: RunnerProcessHandle): Promise<void> {
|
||||
this.#evidence.runnerPid = handle.child.pid ?? null;
|
||||
this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null;
|
||||
this.#publish();
|
||||
if (handle.child.pid !== undefined) {
|
||||
await this.options.onSpawn?.({
|
||||
pid: handle.child.pid,
|
||||
processGroupId: handle.processGroupId ?? null,
|
||||
// Persist the same original process identity exposed to recovery.
|
||||
startedAt: this.processInfo().startedAt,
|
||||
});
|
||||
async #awaitRunnerOwnershipAdmission(core: DurablePrpControlPlane): Promise<boolean> {
|
||||
try {
|
||||
await this.#runnerOwnershipAdmission;
|
||||
return true;
|
||||
} catch {
|
||||
// Close pending handshakes before returning to the core's post-admission
|
||||
// credential recheck. Do not await the outer fence here: it joins these
|
||||
// callbacks after stopping ingress and would otherwise wait on itself.
|
||||
await core.stop();
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
#cancelPendingRunnerLaunch(handle: RunnerProcessHandle): void {
|
||||
if (this.#cancelledRunnerLaunches.has(handle)) return;
|
||||
this.#cancelledRunnerLaunches.add(handle);
|
||||
if (handle.cancelPendingLaunch) {
|
||||
handle.cancelPendingLaunch();
|
||||
} else if (handle.ready) {
|
||||
// Compatibility for custom launchers without a cancellation latch: a late
|
||||
// PID must still be signalled through its own handle, never on this host.
|
||||
void handle.ready.then(() => {
|
||||
if (!this.#controllerDetachedForRestart) signalUnpersistedRunner(handle, this.options.runnerProcessLauncher !== undefined);
|
||||
}).catch(() => undefined);
|
||||
}
|
||||
}
|
||||
|
||||
async #publishSpawnedProcess(handle: RunnerProcessHandle, deadline?: number): Promise<void> {
|
||||
void handle.completion.catch(() => undefined);
|
||||
this.#runnerOwnershipPendingHandle = handle;
|
||||
const assertActive = () => {
|
||||
if (this.#closed) throw new Error("PRP Codex transport is closed");
|
||||
this.#throwIfFailed();
|
||||
if (deadline !== undefined && Date.now() >= deadline) throw new Error("runner_ownership_admission_deadline_exceeded");
|
||||
};
|
||||
let stopped!: () => void;
|
||||
const closeSignal = new Promise<never>((_resolve, reject) => {
|
||||
stopped = () => reject(new Error("PRP Codex transport is closed"));
|
||||
this.#runnerOwnershipAbort.signal.addEventListener("abort", stopped, { once: true });
|
||||
if (this.#runnerOwnershipAbort.signal.aborted) stopped();
|
||||
});
|
||||
let deadlineTimer: NodeJS.Timeout | undefined;
|
||||
const deadlineSignal = new Promise<never>((_resolve, reject) => {
|
||||
if (deadline !== undefined) deadlineTimer = setTimeout(() => reject(new Error("runner_ownership_admission_deadline_exceeded")), Math.max(0, deadline - Date.now()));
|
||||
});
|
||||
const ownership = (async () => {
|
||||
await handle.ready;
|
||||
assertActive();
|
||||
if (handle.ready && handle.child.pid === undefined) {
|
||||
throw new Error("runner_ready_without_process_identity");
|
||||
}
|
||||
this.#evidence.runnerPid = handle.child.pid ?? null;
|
||||
this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null;
|
||||
this.#publish();
|
||||
if (handle.child.pid !== undefined && handle.ready === undefined) {
|
||||
await this.options.onSpawn?.({
|
||||
pid: handle.child.pid,
|
||||
processGroupId: handle.processGroupId ?? null,
|
||||
startedAt: this.processInfo().startedAt,
|
||||
});
|
||||
}
|
||||
assertActive();
|
||||
})();
|
||||
const admission = Promise.race([ownership, closeSignal, deadlineSignal, this.#failureSignal]);
|
||||
this.#runnerOwnershipAdmission = admission;
|
||||
try {
|
||||
await admission;
|
||||
assertActive();
|
||||
} catch (error) {
|
||||
if (this.#closed) throw error; // close/detach owns route and process disposition.
|
||||
throw await this.#fenceFailedRunnerOwnership(handle, error);
|
||||
} finally {
|
||||
this.#runnerOwnershipAbort.signal.removeEventListener("abort", stopped);
|
||||
if (deadlineTimer !== undefined) clearTimeout(deadlineTimer);
|
||||
if (this.#runnerOwnershipPendingHandle === handle) this.#runnerOwnershipPendingHandle = null;
|
||||
}
|
||||
}
|
||||
|
||||
async #fenceFailedRunnerOwnership(handle: RunnerProcessHandle, error: unknown): Promise<Error> {
|
||||
// The process already holds a valid bootstrap ticket. Fence its
|
||||
// route and process before exposing failure; caller.close() may never
|
||||
// run. Keep the durable files/checkpoint and sandbox lease for recovery.
|
||||
const core = this.#core;
|
||||
core?.disconnectActiveRunner();
|
||||
const releaseRoute = this.#controlPlaneRelease;
|
||||
const cleanup = Promise.allSettled([
|
||||
Promise.resolve().then(async () => {
|
||||
this.#cancelPendingRunnerLaunch(handle);
|
||||
const ownershipFailure = handle.ownershipFailure;
|
||||
if (handle.ready && ownershipFailure && ownershipFailure.error === error) {
|
||||
if (ownershipFailure.containment !== "confirmed") throw error;
|
||||
// The launcher already joined verified remote containment before
|
||||
// rejecting readiness/lifetime. Do not signal again or misclassify
|
||||
// that same expected rejection as a new cleanup failure.
|
||||
this.#evidence.runnerExited = true;
|
||||
this.#evidence.runnerExitCode = handle.child.exitCode;
|
||||
this.#evidence.runnerSignal = handle.child.signalCode ?? null;
|
||||
return;
|
||||
}
|
||||
// Default local launches own a detached process group. Custom
|
||||
// launchers (including sandboxes) own their signalling callback;
|
||||
// their reported PIDs must never be signalled on this host.
|
||||
signalUnpersistedRunner(handle, this.options.runnerProcessLauncher !== undefined);
|
||||
// Signal dispatch is not exit proof. Join the original handle's
|
||||
// completion before reporting the replacement itself as exited.
|
||||
let result;
|
||||
try {
|
||||
result = await handle.completion;
|
||||
} catch (completionError) {
|
||||
// Cancellation can finish containment after the initial getter read.
|
||||
// Preserve the original admission error while recognizing only this
|
||||
// exact producer-confirmed lifetime rejection as successful cleanup.
|
||||
const completedFailure = handle.ownershipFailure;
|
||||
if (!handle.ready || !completedFailure || completedFailure.error !== completionError || completedFailure.containment !== "confirmed") throw completionError;
|
||||
this.#evidence.runnerExited = true;
|
||||
this.#evidence.runnerExitCode = handle.child.exitCode;
|
||||
this.#evidence.runnerSignal = handle.child.signalCode ?? null;
|
||||
return;
|
||||
}
|
||||
this.#evidence.runnerExited = true;
|
||||
this.#evidence.runnerExitCode = result.code;
|
||||
this.#evidence.runnerSignal = result.signal as NodeJS.Signals | null;
|
||||
}),
|
||||
Promise.resolve().then(async () => {
|
||||
await releaseRoute?.();
|
||||
if (this.#controlPlaneRelease === releaseRoute) this.#controlPlaneRelease = null;
|
||||
}),
|
||||
Promise.resolve().then(() => core?.stop()).then(() => core?.drainPendingConnectionProcessing()),
|
||||
]);
|
||||
let cleanupDetail = "";
|
||||
let cleanupTimer: NodeJS.Timeout | undefined;
|
||||
try {
|
||||
const results = await Promise.race([
|
||||
cleanup,
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
cleanupTimer = setTimeout(() => reject(new Error("ownership failure cleanup timed out")), 2_000);
|
||||
}),
|
||||
]);
|
||||
const rejected = results.find((result) => result.status === "rejected");
|
||||
if (rejected?.status === "rejected") throw rejected.reason;
|
||||
} catch (cleanupError) {
|
||||
cleanupDetail = `; cleanup incomplete: ${cleanupError instanceof Error ? cleanupError.message : String(cleanupError)}`;
|
||||
} finally {
|
||||
if (cleanupTimer !== undefined) clearTimeout(cleanupTimer);
|
||||
}
|
||||
const failure = new Error(
|
||||
`native_runner_process_ownership_failed: ${error instanceof Error ? error.message : String(error)}${cleanupDetail}`,
|
||||
);
|
||||
this.#failTransport(failure);
|
||||
return failure;
|
||||
}
|
||||
|
||||
async #readDurableRunnerState(): Promise<Record<string, unknown>> {
|
||||
if (this.options.readRunnerState) return this.options.readRunnerState();
|
||||
return record(
|
||||
|
|
@ -4100,6 +4242,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
if (this.#closed) return;
|
||||
this.#controllerDetachedForRestart = true;
|
||||
this.#closed = true;
|
||||
this.#runnerOwnershipAbort.abort();
|
||||
this.#turnStartAdmission?.resolve(false);
|
||||
if (this.#pump !== null) clearInterval(this.#pump);
|
||||
this.#pump = null;
|
||||
|
|
@ -4128,6 +4271,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
|
||||
async #closeOnce(): Promise<void> {
|
||||
this.#closed = true;
|
||||
this.#runnerOwnershipAbort.abort();
|
||||
if (this.#runnerOwnershipPendingHandle) {
|
||||
try { this.#cancelPendingRunnerLaunch(this.#runnerOwnershipPendingHandle); }
|
||||
catch (error) { this.#evidence.diagnostics.push(`pending launch cancellation failed: ${error instanceof Error ? error.message : String(error)}`); }
|
||||
}
|
||||
this.#turnStartAdmission?.resolve(false);
|
||||
const adoptedRunner = this.options.adoptExistingRunner;
|
||||
// `settled` is a durable-state assertion, not merely the absence of a
|
||||
|
|
@ -4330,6 +4478,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
identity,
|
||||
expectedRunnerVersion: runnerArtifact.version,
|
||||
expectedRunnerDigest: runnerArtifact.digest,
|
||||
beforeAuthenticatedConnection: async () => { await this.#awaitRunnerOwnershipAdmission(core); },
|
||||
onProtocolIntegrityError: (error) => this.#failTransport(error),
|
||||
onSemanticToolInput: (call) => this.#handleSemanticToolInput(call),
|
||||
connectionLeaseTtlMs: 60 * 60 * 1_000,
|
||||
|
|
@ -4706,8 +4855,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
processLauncher: this.options.runnerProcessLauncher,
|
||||
});
|
||||
this.#handle = handle;
|
||||
this.#watchRunner(handle);
|
||||
await this.#publishSpawnedProcess(handle);
|
||||
if (this.#closed) throw new Error("PRP Codex transport is closed");
|
||||
this.#throwIfFailed();
|
||||
this.#watchRunner(handle);
|
||||
await registration?.activate?.();
|
||||
if (registration?.failure) {
|
||||
void registration.failure.catch((error: unknown) => {
|
||||
|
|
@ -5056,25 +5207,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
expectedRunnerDigest: runnerArtifact.digest,
|
||||
onProtocolIntegrityError: (error) => this.#failTransport(error),
|
||||
onSemanticToolInput: (call) => this.#handleSemanticToolInput(call),
|
||||
...(warmRecovery
|
||||
? {
|
||||
beforeAuthenticatedConnection: async (admission) => {
|
||||
// Core policy already validated the current lease and tuple.
|
||||
// Receipt replay still needs the recovery claim, including the
|
||||
// completed-core/lost-final-ACK window. Ordinary post-ACK lease
|
||||
// reconnects no longer depend on that historical claim.
|
||||
if (
|
||||
core.store.state.warmTransition?.receipt.transitionId ===
|
||||
warmRecovery.transitionId ||
|
||||
admission.warmTransitionId === warmRecovery.transitionId
|
||||
) {
|
||||
await this.options.authorizeWarmTransitionRecovery?.(
|
||||
"before_authentication",
|
||||
);
|
||||
}
|
||||
},
|
||||
}
|
||||
: {}),
|
||||
beforeAuthenticatedConnection: async (admission) => {
|
||||
if (!await this.#awaitRunnerOwnershipAdmission(core)) return;
|
||||
// Preserve warm-transition recovery admission after process ownership.
|
||||
if (warmRecovery && (
|
||||
core.store.state.warmTransition?.receipt.transitionId === warmRecovery.transitionId ||
|
||||
admission.warmTransitionId === warmRecovery.transitionId
|
||||
)) {
|
||||
await this.options.authorizeWarmTransitionRecovery?.("before_authentication");
|
||||
}
|
||||
},
|
||||
connectionLeaseTtlMs: 60 * 60 * 1_000,
|
||||
});
|
||||
this.#core = core;
|
||||
|
|
@ -5337,8 +5479,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
});
|
||||
if (handle) {
|
||||
this.#handle = handle;
|
||||
this.#watchRunner(handle);
|
||||
await this.#publishSpawnedProcess(handle);
|
||||
if (this.#closed) throw new Error("PRP Codex transport is closed");
|
||||
this.#throwIfFailed();
|
||||
this.#watchRunner(handle);
|
||||
}
|
||||
if (oldTransitionRegistration && newTransitionRegistration) {
|
||||
await oldTransitionRegistration.activate?.();
|
||||
|
|
@ -6486,11 +6630,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
|||
},
|
||||
);
|
||||
try {
|
||||
await this.#publishSpawnedProcess(recoveredHandle);
|
||||
await this.#publishSpawnedProcess(recoveredHandle, deadline);
|
||||
} catch (error) {
|
||||
this.#failTransport(new Error(
|
||||
`native_runner_process_ownership_failed: ${error instanceof Error ? error.message : String(error)}`,
|
||||
));
|
||||
// Publication fenced the route and process before rejecting. A
|
||||
// failed ownership save is terminal, never a fresh recovery attempt.
|
||||
return;
|
||||
}
|
||||
if (this.#closed || this.#failure !== null) return;
|
||||
|
|
@ -6608,7 +6751,29 @@ export const runnerdLaunchProfileInternals = Object.freeze({
|
|||
p0ReserveBytes: RUNNERD_P0_RESERVE_BYTES,
|
||||
});
|
||||
|
||||
// A locally spawned detached group can outlive its runner. Unlike waiting for
|
||||
// completion with a timeout, unconditional containment also reaches descendants
|
||||
// when the group leader exited while ownership persistence was pending.
|
||||
function signalUnpersistedRunner(handle: RunnerProcessHandle, customLauncher: boolean): void {
|
||||
const groupId = handle.processGroupId;
|
||||
if (
|
||||
!customLauncher && process.platform !== "win32" &&
|
||||
Number.isSafeInteger(groupId) && (groupId ?? 0) > 0 && groupId === handle.child.pid
|
||||
) {
|
||||
try {
|
||||
process.kill(-groupId!, "SIGKILL");
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== "ESRCH") throw error;
|
||||
}
|
||||
return;
|
||||
}
|
||||
// A custom handle may refer to a remote PID. Only its launcher-provided
|
||||
// callback has authority to signal it; never address that PID on this host.
|
||||
handle.child.kill("SIGKILL");
|
||||
}
|
||||
|
||||
export const runnerdRecoveryInternals = Object.freeze({
|
||||
signalUnpersistedRunner,
|
||||
completedMaintenanceTerminalReceipt,
|
||||
completedMaintenanceTerminalReplayMatches,
|
||||
awaitProviderDrainBarrier,
|
||||
|
|
|
|||
|
|
@ -315,7 +315,8 @@ describeEmbeddedPostgres("native runner restart recovery with real processes", (
|
|||
const [run] = await fixture.db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId));
|
||||
expect(run).toMatchObject({ nativePhase: phase, errorCode: code });
|
||||
const [issue] = await fixture.db.select().from(issues).where(eq(issues.id, fixture.issueId));
|
||||
expect(issue!.status).toBe(phase === "terminal_failure" ? "in_review" : "in_progress");
|
||||
// Exhausted recovery has no completed result for human review.
|
||||
expect(issue!.status).toBe(phase === "terminal_failure" ? "blocked" : "in_progress");
|
||||
expect(onSpawn).not.toHaveBeenCalled();
|
||||
expect(execute).toHaveBeenCalledTimes(scenario === "missing-host-state" ? 0 : 1);
|
||||
if (scenario !== "missing-host-state") expect(await readFile(journalPath, "utf8")).toBe(journal);
|
||||
|
|
|
|||
|
|
@ -299,6 +299,116 @@ beforeEach(() => {
|
|||
});
|
||||
|
||||
describe("remote runner process supervision", () => {
|
||||
it.runIf(process.platform === "linux" || !!process.env.PAPERCLIP_REMOTE_PROCESS_TEST_CONTAINER).each([
|
||||
"live-parent", "exited-parent", "replaced-marker", "unverified-member", "unavailable-containment",
|
||||
"cancel-pending-ownership", "cancel-pending-ownership-late-rejection",
|
||||
])("contains ownership-save failure without signalling unrelated remote processes (%s)", async (scenario) => {
|
||||
const root = await mkdtemp(join(process.env.PAPERCLIP_REMOTE_PROCESS_TEST_ROOT ?? tmpdir(), "remote-ownership-"));
|
||||
const container = process.env.PAPERCLIP_REMOTE_PROCESS_TEST_CONTAINER;
|
||||
const execute = async (command: { command?: string; args?: string[]; cwd?: string; timeoutMs?: number }) => {
|
||||
const result = spawnSync(container ? "docker" : command.command!, container
|
||||
? ["exec", container, command.command!, ...(command.args ?? [])]
|
||||
: command.args ?? [], { cwd: command.cwd, timeout: command.timeoutMs ?? 10_000, encoding: "utf8" });
|
||||
let stdout = result.stdout ?? "";
|
||||
if (scenario === "unavailable-containment" && command.args?.[2] === "paperclip-runner-process-identity") {
|
||||
stdout = stdout.split("\n").slice(0, 4).join("\n") + "\n";
|
||||
}
|
||||
return { exitCode: result.status, signal: result.signal, timedOut: result.error?.message.includes("ETIMEDOUT") ?? false,
|
||||
stdout, stderr: result.stderr ?? "" };
|
||||
};
|
||||
const script = join(root, "runner.sh"), pidsPath = join(root, "pids"), unrelatedPath = join(root, "unrelated");
|
||||
await writeFile(script, `#!/bin/sh
|
||||
trap '' TERM
|
||||
${scenario === "unverified-member" ? "env -u PAPERCLIP_RUNNER_PROCESS_NONCE " : ""}sh -c 'trap "" TERM; while :; do sleep 1; done' --runner-id owned-child &
|
||||
printf '%s %s\\n' "$$" "$!" > '${pidsPath}'
|
||||
while :; do sleep 1; done
|
||||
`);
|
||||
const marker = join(root, "runner-process.identity"), checkpoint = join(root, "runner-state.json");
|
||||
await writeFile(checkpoint, '{"keep":"original-checkpoint"}');
|
||||
let pids: number[] = [], unrelated = 0, expectedMarker = "";
|
||||
const cancelling = scenario.startsWith("cancel-pending-ownership");
|
||||
const confirmed = scenario === "live-parent" || scenario === "exited-parent" || cancelling;
|
||||
let cancelLaunch!: () => void;
|
||||
let resolveOwnership!: () => void, rejectOwnership!: (error: Error) => void;
|
||||
const pendingOwnership = new Promise<void>((resolve, reject) => { resolveOwnership = resolve; rejectOwnership = reject; });
|
||||
try {
|
||||
await execute({ command: "sh", args: ["-c", `nohup setsid sh -c 'echo $$ > "${unrelatedPath}"; exec sleep 300' </dev/null >/dev/null 2>&1 &`] });
|
||||
await vi.waitFor(async () => { unrelated = Number((await readFile(unrelatedPath, "utf8")).trim()); expect(unrelated).toBeGreaterThan(1); });
|
||||
const saveFailure = new Error("ownership-save-rejected");
|
||||
const onSpawn = vi.fn(async () => {
|
||||
await vi.waitFor(async () => { pids = (await readFile(pidsPath, "utf8")).trim().split(/\s+/).map(Number); expect(pids).toHaveLength(2); });
|
||||
expectedMarker = await readFile(marker, "utf8");
|
||||
if (scenario === "exited-parent") await execute({ command: "sh", args: ["-c", `kill -KILL ${pids[0]}`] });
|
||||
if (scenario === "replaced-marker") {
|
||||
const lines = expectedMarker.split("\n"); lines[1] = String(unrelated); expectedMarker = lines.join("\n");
|
||||
await writeFile(marker, expectedMarker);
|
||||
}
|
||||
if (cancelling) {
|
||||
cancelLaunch();
|
||||
await pendingOwnership;
|
||||
return;
|
||||
}
|
||||
throw saveFailure;
|
||||
});
|
||||
const launcher = createRemoteRunnerProcessLauncher({
|
||||
target: { kind: "remote", transport: "sandbox", environmentId: "test", leaseId: "test", remoteCwd: root },
|
||||
runner: { execute } as never, remoteBinary: "/bin/sh", processIdentityPath: marker,
|
||||
stateDirectory: root, diagnosticsDirectory: join(root, "diagnostics"), runnerInstanceId: "runner-ownership-test", onSpawn,
|
||||
});
|
||||
const handle = launcher({ command: "/bin/sh", args: [script, "--runner-id", "runner-ownership-test"], cwd: root, environment: {} });
|
||||
cancelLaunch = () => handle.cancelPendingLaunch?.();
|
||||
const failure = await handle.completion.catch(error => error);
|
||||
expect(failure).toBeInstanceOf(Error);
|
||||
if (cancelling) expect(failure.message).toBe("runner_remote_process_launch_cancelled");
|
||||
else if (confirmed) expect(failure).toBe(saveFailure);
|
||||
else expect(failure.message).toContain("cleanup incomplete");
|
||||
await expect(handle.ready).rejects.toBe(failure);
|
||||
expect(handle.ownershipFailure).toEqual({ error: failure, containment: confirmed ? "confirmed" : "unconfirmed" });
|
||||
expect(onSpawn).toHaveBeenCalledOnce();
|
||||
if (scenario === "cancel-pending-ownership-late-rejection") rejectOwnership(new Error("late ownership save failure"));
|
||||
else resolveOwnership();
|
||||
await new Promise(resolve => setTimeout(resolve, 0));
|
||||
await expect(handle.ready).rejects.toBe(failure);
|
||||
handle.cancelPendingLaunch?.(); // A classified failure is immutable.
|
||||
expect(handle.ownershipFailure).toEqual({ error: failure, containment: confirmed ? "confirmed" : "unconfirmed" });
|
||||
for (const pid of pids) {
|
||||
const observed = await execute({ command: "sh", args: ["-c", `test ! -r /proc/${pid}/stat || test "$(awk '{print $3}' /proc/${pid}/stat)" = Z`] });
|
||||
expect(observed.exitCode, `remote owned process ${pid}`).toBe(confirmed ? 0 : 1);
|
||||
}
|
||||
expect((await execute({ command: "sh", args: ["-c", `kill -0 ${unrelated}`] })).exitCode).toBe(0);
|
||||
expect(await readFile(checkpoint, "utf8")).toBe('{"keep":"original-checkpoint"}');
|
||||
expect(await readFile(marker, "utf8")).toBe(expectedMarker);
|
||||
} finally {
|
||||
for (const pid of [...pids, unrelated].filter(pid => Number.isSafeInteger(pid) && pid > 1)) {
|
||||
await execute({ command: "sh", args: ["-c", `kill -KILL ${pid} 2>/dev/null || true`] });
|
||||
}
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
}, 30_000);
|
||||
it.each([false, true])("does not dispatch after cancellation while staging is pending (trace=%s)", async (traced) => {
|
||||
let finishStaging!: () => void;
|
||||
const staging = new Promise<void>(resolve => { finishStaging = resolve; });
|
||||
const execute = vi.fn(), onSpawn = vi.fn(), onRunnerProcessSpawned = vi.fn();
|
||||
const launcher = createRemoteRunnerProcessLauncher({
|
||||
target: { kind: "remote", transport: "sandbox", environmentId: "test", leaseId: "test", remoteCwd: "/workspace" },
|
||||
runner: { execute } as never, remoteBinary: "/runtime/runnerd", processIdentityPath: "/runtime/identity",
|
||||
stateDirectory: "/runtime", diagnosticsDirectory: "/runtime/diagnostics", runnerInstanceId: "runner-cancel",
|
||||
ensureArtifact: () => staging, onSpawn, onRunnerProcessSpawned,
|
||||
trace: traced ? { measure: (_name: string, action: () => Promise<void>) => action() } as never : undefined,
|
||||
});
|
||||
const handle = launcher({ command: "/runtime/runnerd", args: [], cwd: "/workspace", environment: {} });
|
||||
handle.cancelPendingLaunch?.();
|
||||
const failure = await handle.completion.catch(error => error);
|
||||
expect(failure.message).toBe("runner_remote_process_launch_cancelled");
|
||||
await expect(handle.ready).rejects.toBe(failure);
|
||||
expect(handle.ownershipFailure).toEqual({ error: failure, containment: "confirmed" });
|
||||
finishStaging();
|
||||
await new Promise(resolve => setTimeout(resolve, 0));
|
||||
expect(execute).not.toHaveBeenCalled();
|
||||
expect(onSpawn).not.toHaveBeenCalled();
|
||||
expect(onRunnerProcessSpawned).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it.each([false, true])("supervises detached runnerd and observes signal failures (%s)", async (signalFails) => {
|
||||
let launchNonce = "";
|
||||
const execute = vi.fn(
|
||||
|
|
@ -405,6 +515,10 @@ describe("remote runner process supervision", () => {
|
|||
stderr: "paperclip-runnerd: provider transport closed",
|
||||
});
|
||||
|
||||
await expect(handle.ready).resolves.toBeUndefined();
|
||||
handle.cancelPendingLaunch?.();
|
||||
expect(handle.ownershipFailure).toBeUndefined();
|
||||
|
||||
const launch = execute.mock.calls.find(
|
||||
([input]) => input.args?.[2] === "paperclip-runner-launch",
|
||||
)?.[0];
|
||||
|
|
|
|||
|
|
@ -9729,7 +9729,101 @@ const REMOTE_RUNNER_IDENTITY_CHECK_SCRIPT =
|
|||
'set -eu; identity_path=$1; expected_nonce=$2; expected_runner_id=$3; expected_pid=$4; test -f "$identity_path" && test ! -L "$identity_path" || exit 3; { IFS= read -r nonce; IFS= read -r pid; IFS= read -r started_at; IFS= read -r runner_id; } < "$identity_path"; test "$nonce" = "$expected_nonce" && test "$runner_id" = "$expected_runner_id" && test "$pid" = "$expected_pid" && test -n "$started_at" || exit 4; kill -0 "$pid" 2>/dev/null || exit 3; if test -r "/proc/$pid/cmdline"; then command_line=$(tr "\\000" "\\n" < "/proc/$pid/cmdline"); printf "%s\\n" "$command_line" | grep -Fqx -- "--runner-id" || exit 4; printf "%s\\n" "$command_line" | grep -Fqx -- "$expected_runner_id" || exit 4; fi';
|
||||
|
||||
const REMOTE_RUNNER_CHILD_LAUNCH_SCRIPT =
|
||||
'set -eu; identity_path=$1; identity_nonce=$2; runner_instance_id=$3; diagnostics_directory=$4; shift 4; umask 077; test ! -L "$diagnostics_directory"; if test -e "$diagnostics_directory"; then test -d "$diagnostics_directory"; else mkdir -p -- "$diagnostics_directory"; fi; chmod 0700 "$diagnostics_directory"; started_at=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ"); identity_tmp="${identity_path}.tmp.$$"; printf "%s\\n%s\\n%s\\n%s\\n" "$identity_nonce" "$$" "$started_at" "$runner_instance_id" > "$identity_tmp"; chmod 0600 "$identity_tmp"; mv -f -- "$identity_tmp" "$identity_path"; exec "$@"';
|
||||
'set -eu; identity_path=$1; identity_nonce=$2; runner_instance_id=$3; diagnostics_directory=$4; shift 4; export PAPERCLIP_RUNNER_PROCESS_NONCE="$identity_nonce"; umask 077; test ! -L "$diagnostics_directory"; if test -e "$diagnostics_directory"; then test -d "$diagnostics_directory"; else mkdir -p -- "$diagnostics_directory"; fi; chmod 0700 "$diagnostics_directory"; started_at=$(date -u +"%Y-%m-%dT%H:%M:%S.%3NZ"); identity_tmp="${identity_path}.tmp.$$"; printf "%s\\n%s\\n%s\\n%s\\n" "$identity_nonce" "$$" "$started_at" "$runner_instance_id" > "$identity_tmp"; chmod 0600 "$identity_tmp"; mv -f -- "$identity_tmp" "$identity_path"; exec "$@"';
|
||||
|
||||
// Linux proof is optional: older/non-Linux transports still launch normally.
|
||||
// Capture it in the existing marker read, before publishing durable ownership.
|
||||
type RemoteRunnerContainment = { pid: number; startTicks: string; sessionId: number };
|
||||
const REMOTE_RUNNER_CONTAINMENT_PROBE = String.raw`
|
||||
import json, os, pathlib, sys
|
||||
try:
|
||||
marker = pathlib.Path(sys.argv[1]).read_text().splitlines()
|
||||
pid = int(marker[1])
|
||||
fields = pathlib.Path(f"/proc/{pid}/stat").read_text().rsplit(")", 1)[1].split()
|
||||
command = pathlib.Path(f"/proc/{pid}/cmdline").read_bytes().split(b"\0")
|
||||
environment = pathlib.Path(f"/proc/{pid}/environ").read_bytes().split(b"\0")
|
||||
expected = ("PAPERCLIP_RUNNER_PROCESS_NONCE=" + marker[0]).encode()
|
||||
if expected not in environment or not any(command[i:i+2] == [b"--runner-id", marker[3].encode()] for i in range(len(command))):
|
||||
print(json.dumps({"pending": True}))
|
||||
elif int(fields[2]) == pid and int(fields[3]) == pid:
|
||||
print(json.dumps({"pid": pid, "startTicks": fields[19], "sessionId": pid}))
|
||||
else:
|
||||
print("null")
|
||||
except (OSError, ValueError, IndexError):
|
||||
print("null")
|
||||
`;
|
||||
|
||||
// Open pidfds before checking process birth/nonce. Signals then address those
|
||||
// kernel process objects, never a numeric PID that may have been recycled.
|
||||
// An exited leader is safe only while every surviving session member carries
|
||||
// this launch's inherited nonce. Unknown members preserve all recovery files.
|
||||
const REMOTE_RUNNER_OWNERSHIP_FAILURE_CLEANUP = String.raw`
|
||||
import json, os, pathlib, signal, sys, time
|
||||
marker_path, nonce, runner_id, raw_pid, started_at, raw_proof = sys.argv[1:]
|
||||
proof = json.loads(raw_proof)
|
||||
pid = int(raw_pid)
|
||||
expected_marker = [nonce, str(pid), started_at, runner_id]
|
||||
expected_environment = ("PAPERCLIP_RUNNER_PROCESS_NONCE=" + nonce).encode()
|
||||
if not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"):
|
||||
sys.exit(4)
|
||||
def check_marker():
|
||||
if os.path.islink(marker_path) or pathlib.Path(marker_path).read_text().splitlines() != expected_marker:
|
||||
raise RuntimeError("runner marker changed")
|
||||
def members():
|
||||
found = []
|
||||
try:
|
||||
check_marker()
|
||||
for entry in pathlib.Path("/proc").iterdir():
|
||||
if not entry.name.isdigit():
|
||||
continue
|
||||
try:
|
||||
fields = (entry / "stat").read_text().rsplit(")", 1)[1].split()
|
||||
if int(fields[3]) != proof["sessionId"] or fields[0] == "Z":
|
||||
continue
|
||||
member_pid = int(entry.name)
|
||||
fd = os.pidfd_open(member_pid)
|
||||
found.append(fd)
|
||||
current = (entry / "stat").read_text().rsplit(")", 1)[1].split()
|
||||
if fields[19] != current[19] or int(current[3]) != proof["sessionId"]:
|
||||
raise RuntimeError("process identity changed")
|
||||
if member_pid == pid and current[19] != proof["startTicks"]:
|
||||
raise RuntimeError("runner pid reused")
|
||||
if expected_environment not in (entry / "environ").read_bytes().split(b"\0"):
|
||||
raise RuntimeError("unverified session member")
|
||||
except (FileNotFoundError, ProcessLookupError):
|
||||
continue
|
||||
return found
|
||||
except BaseException:
|
||||
for fd in found:
|
||||
os.close(fd)
|
||||
raise
|
||||
try:
|
||||
if proof["pid"] != pid or proof["sessionId"] != pid:
|
||||
raise RuntimeError("invalid containment")
|
||||
deadline = time.monotonic() + 8
|
||||
kill_after = time.monotonic() + 2
|
||||
while True:
|
||||
fds = members()
|
||||
if not fds:
|
||||
check_marker()
|
||||
sys.exit(0)
|
||||
try:
|
||||
requested = signal.SIGKILL if time.monotonic() >= kill_after else signal.SIGTERM
|
||||
for fd in fds:
|
||||
try:
|
||||
signal.pidfd_send_signal(fd, requested)
|
||||
except ProcessLookupError:
|
||||
pass
|
||||
finally:
|
||||
for fd in fds:
|
||||
os.close(fd)
|
||||
if time.monotonic() >= deadline:
|
||||
sys.exit(5)
|
||||
time.sleep(0.1)
|
||||
except (OSError, ValueError, KeyError, RuntimeError) as error:
|
||||
print(str(error), file=sys.stderr)
|
||||
sys.exit(4)
|
||||
`;
|
||||
|
||||
const REMOTE_RUNNER_FAILED_IDENTITY_CLEANUP_SCRIPT =
|
||||
'set -eu; identity_path=$1; expected_nonce=$2; expected_runner_id=$3; marker_wait=0; while { test ! -f "$identity_path" || test -L "$identity_path"; } && test "$marker_wait" -lt 50; do marker_wait=$((marker_wait + 1)); sleep 0.1; done; test -f "$identity_path" && test ! -L "$identity_path" || exit 3; { IFS= read -r nonce; IFS= read -r pid; IFS= read -r started_at; IFS= read -r runner_id; } < "$identity_path"; test "$nonce" = "$expected_nonce" && test "$runner_id" = "$expected_runner_id" && test -n "$started_at" || exit 4; case "$pid" in ""|*[!0-9]*) exit 4 ;; esac; test "$pid" -gt 0 || exit 4; if kill -0 "$pid" 2>/dev/null; then if test -r "/proc/$pid/cmdline"; then command_line=$(tr "\\000" "\\n" < "/proc/$pid/cmdline"); printf "%s\\n" "$command_line" | grep -Fqx -- "--runner-id" || exit 4; printf "%s\\n" "$command_line" | grep -Fqx -- "$expected_runner_id" || exit 4; fi; signal_target=$pid; if command -v ps >/dev/null 2>&1; then session_id=$(ps -o sid= -p "$pid" 2>/dev/null | tr -d " ") || true; if test "$session_id" = "$pid"; then signal_target="-$pid"; fi; fi; kill -TERM -- "$signal_target" 2>/dev/null || kill -TERM "$pid" 2>/dev/null || true; term_wait=0; while kill -0 "$pid" 2>/dev/null && test "$term_wait" -lt 50; do term_wait=$((term_wait + 1)); sleep 0.1; done; if kill -0 "$pid" 2>/dev/null; then kill -KILL -- "$signal_target" 2>/dev/null || kill -KILL "$pid" 2>/dev/null || true; kill_wait=0; while kill -0 "$pid" 2>/dev/null && test "$kill_wait" -lt 50; do kill_wait=$((kill_wait + 1)); sleep 0.1; done; fi; kill -0 "$pid" 2>/dev/null && exit 5; fi; test -f "$identity_path" && test ! -L "$identity_path" || exit 4; { IFS= read -r final_nonce; IFS= read -r final_pid; IFS= read -r final_started_at; IFS= read -r final_runner_id; } < "$identity_path"; test "$final_nonce" = "$nonce" && test "$final_pid" = "$pid" && test "$final_started_at" = "$started_at" && test "$final_runner_id" = "$runner_id" || exit 4; rm -f -- "$identity_path"';
|
||||
|
|
@ -9761,7 +9855,7 @@ async function waitForRemoteRunnerProcessIdentity(input: {
|
|||
identityPath: string;
|
||||
nonce: string;
|
||||
runnerInstanceId: string;
|
||||
}): Promise<{ pid: number; startedAt: string }> {
|
||||
}): Promise<{ pid: number; startedAt: string; containment: RemoteRunnerContainment | null }> {
|
||||
const deadline = Date.now() + REMOTE_RUNNER_PROCESS_IDENTITY_WAIT_MS;
|
||||
while (Date.now() < deadline) {
|
||||
const result = await input.runner
|
||||
|
|
@ -9769,19 +9863,30 @@ async function waitForRemoteRunnerProcessIdentity(input: {
|
|||
command: "sh",
|
||||
args: [
|
||||
"-c",
|
||||
'test -f "$1" && test ! -L "$1" && cat -- "$1"',
|
||||
'test -f "$1" && test ! -L "$1" && cat -- "$1"; if command -v python3 >/dev/null 2>&1; then python3 -c "$2" "$1"; fi',
|
||||
"paperclip-runner-process-identity",
|
||||
input.identityPath,
|
||||
REMOTE_RUNNER_CONTAINMENT_PROBE,
|
||||
],
|
||||
bypassSession: true,
|
||||
timeoutMs: 2_000,
|
||||
})
|
||||
.catch(() => null);
|
||||
const lines = result?.stdout.trim().split("\n") ?? [];
|
||||
const identity =
|
||||
result && result.exitCode === 0 && !result.timedOut
|
||||
? parseRemoteRunnerProcessIdentity(result.stdout, input)
|
||||
? parseRemoteRunnerProcessIdentity(lines.slice(0, 4).join("\n"), input)
|
||||
: null;
|
||||
if (identity) return identity;
|
||||
if (identity) {
|
||||
let proof: Record<string, unknown> | null = null;
|
||||
try { proof = JSON.parse(lines[4] ?? "null"); } catch { /* optional platform proof */ }
|
||||
if (proof?.pending !== true) {
|
||||
const containment = proof?.pid === identity.pid && proof?.sessionId === identity.pid
|
||||
&& typeof proof.startTicks === "string" && /^\d+$/.test(proof.startTicks)
|
||||
? proof as RemoteRunnerContainment : null;
|
||||
return { ...identity, containment };
|
||||
}
|
||||
}
|
||||
await new Promise<void>((resolve) => setTimeout(resolve, 100));
|
||||
}
|
||||
throw new Error("runner_remote_process_identity_unavailable");
|
||||
|
|
@ -9835,7 +9940,25 @@ export function createRemoteRunnerProcessLauncher(input: {
|
|||
nonce: string;
|
||||
pid: number;
|
||||
startedAt: string;
|
||||
containment: RemoteRunnerContainment | null;
|
||||
} | null = null;
|
||||
let ownershipFailure: { error: unknown; containment: "confirmed" | "unconfirmed" } | undefined;
|
||||
let resolveReady!: () => void;
|
||||
let rejectReady!: (error: unknown) => void;
|
||||
const ready = new Promise<void>((resolve, reject) => { resolveReady = resolve; rejectReady = reject; });
|
||||
void ready.catch(() => undefined);
|
||||
let launchCancelled = false;
|
||||
let launchDispatched = false;
|
||||
let readySucceeded = false;
|
||||
const cancelledError = new Error("runner_remote_process_launch_cancelled");
|
||||
let rejectCancellation!: (error: Error) => void;
|
||||
const cancellation = new Promise<never>((_resolve, reject) => { rejectCancellation = reject; });
|
||||
void cancellation.catch(() => undefined);
|
||||
const cancelPendingLaunch = () => {
|
||||
if (readySucceeded || ownershipFailure || launchCancelled) return;
|
||||
launchCancelled = true;
|
||||
rejectCancellation(cancelledError);
|
||||
};
|
||||
const child: RunnerProcessHandle["child"] = {
|
||||
pid: undefined,
|
||||
exitCode: null,
|
||||
|
|
@ -9885,15 +10008,16 @@ export function createRemoteRunnerProcessLauncher(input: {
|
|||
const completion = (async () => {
|
||||
if (input.ensureArtifact) {
|
||||
if (input.trace) {
|
||||
await input.trace.measure(
|
||||
await Promise.race([input.trace.measure(
|
||||
"runner.runtime.stage",
|
||||
input.ensureArtifact,
|
||||
{ parentName: "runner.session.startup" },
|
||||
);
|
||||
), cancellation]);
|
||||
} else {
|
||||
await input.ensureArtifact();
|
||||
await Promise.race([input.ensureArtifact(), cancellation]);
|
||||
}
|
||||
}
|
||||
if (launchCancelled) throw cancelledError;
|
||||
const launchStartedAtMs = Date.now();
|
||||
// The provider's onSpawn callback is optional and some sandbox command
|
||||
// runners cannot report a remote pid until after the command has begun
|
||||
|
|
@ -9924,6 +10048,10 @@ export function createRemoteRunnerProcessLauncher(input: {
|
|||
// into its own session instead; its own bounded diagnostics directory and
|
||||
// durable PRP state remain the authorities, and the controller monitors
|
||||
// the exact persisted process identity below.
|
||||
if (launchCancelled) throw cancelledError;
|
||||
launchDispatched = true;
|
||||
// After dispatch, cancellation must still adopt the exact identity before
|
||||
// containment. Abandoning the RPC here could strand an unknown process.
|
||||
const launchResult = await runner.execute({
|
||||
command: "sh",
|
||||
args: [
|
||||
|
|
@ -9951,7 +10079,7 @@ export function createRemoteRunnerProcessLauncher(input: {
|
|||
: "runner_remote_process_launch_failed",
|
||||
);
|
||||
}
|
||||
let identity: { pid: number; startedAt: string };
|
||||
let identity: { pid: number; startedAt: string; containment: RemoteRunnerContainment | null };
|
||||
try {
|
||||
identity = await waitForRemoteRunnerProcessIdentity({
|
||||
runner,
|
||||
|
|
@ -9983,11 +10111,36 @@ export function createRemoteRunnerProcessLauncher(input: {
|
|||
}
|
||||
launchedIdentity = { nonce: identityNonce, ...identity };
|
||||
child.pid = identity.pid;
|
||||
await input.onSpawn?.({
|
||||
pid: identity.pid,
|
||||
processGroupId: null,
|
||||
startedAt: identity.startedAt,
|
||||
});
|
||||
try {
|
||||
if (launchCancelled) throw cancelledError;
|
||||
await Promise.race([Promise.resolve().then(() => {
|
||||
if (launchCancelled) throw cancelledError;
|
||||
return input.onSpawn?.({
|
||||
pid: identity.pid,
|
||||
processGroupId: null,
|
||||
startedAt: identity.startedAt,
|
||||
});
|
||||
}), cancellation]);
|
||||
if (launchCancelled) throw cancelledError;
|
||||
} catch (error) {
|
||||
const cleanup = identity.containment ? await runner.execute({
|
||||
command: "python3",
|
||||
args: ["-c", REMOTE_RUNNER_OWNERSHIP_FAILURE_CLEANUP, input.processIdentityPath,
|
||||
identityNonce, input.runnerInstanceId, String(identity.pid), identity.startedAt,
|
||||
JSON.stringify(identity.containment)],
|
||||
bypassSession: true,
|
||||
timeoutMs: 12_000,
|
||||
}).catch(() => null) : null;
|
||||
if (cleanup?.exitCode !== 0 || cleanup.timedOut) {
|
||||
const failure = new Error(`${error instanceof Error ? error.message : String(error)}; cleanup incomplete: remote runner ownership unverified`, { cause: error });
|
||||
ownershipFailure = { error: failure, containment: "unconfirmed" };
|
||||
throw failure;
|
||||
}
|
||||
ownershipFailure = { error, containment: "confirmed" };
|
||||
throw error;
|
||||
}
|
||||
readySucceeded = true;
|
||||
resolveReady();
|
||||
await input.trace?.record({
|
||||
name: "runner.process.launch",
|
||||
parentName: "runner.session.startup",
|
||||
|
|
@ -10067,9 +10220,18 @@ export function createRemoteRunnerProcessLauncher(input: {
|
|||
};
|
||||
}
|
||||
})();
|
||||
void completion.catch(error => {
|
||||
if (!launchDispatched && error === cancelledError) {
|
||||
ownershipFailure = { error, containment: "confirmed" };
|
||||
}
|
||||
rejectReady(error);
|
||||
});
|
||||
return {
|
||||
child,
|
||||
completion,
|
||||
ready,
|
||||
cancelPendingLaunch,
|
||||
get ownershipFailure() { return ownershipFailure; },
|
||||
get startedAt() { return launchedIdentity?.startedAt; },
|
||||
};
|
||||
};
|
||||
|
|
|
|||
Loading…
Reference in New Issue