mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
fix(runner): drain owned startup before certifying cleanup
Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
1 parent
3aaf1f8b1e
commit
463ebb6266
4 files changed
+213
-7
No files matched your search
@@ -146,3 +146,11 @@ retain existing execution on unrelated edits and use recorded run configuration
|
||||
for active/recovered runs. Explicit runner changes use revisions and session
|
||||
invalidation. Verify saved configuration, current versus historical run identity,
|
||||
restart and follow-up without granting session-reset authority to an agent.
|
||||
|
||||
Scoped source-only evidence: the
|
||||
[native transport](../../packages/paperclip-runner/src/live/runnerd-codex-transport.ts)
|
||||
joins owned startup before certifying cleanup. Unconfirmed preparation or
|
||||
registration retains recovery state; late registration may retire only its own
|
||||
route and cannot launch a provider after close. This does not grant an agent
|
||||
account-setup, cleanup-override or session-reset authority. Verify the no-provider
|
||||
ownership controls independently from installed/live provider acceptance.
|
||||
@@ -80,3 +80,11 @@ different observations, both required in the evidence.
|
||||
- The run log is instance database data; it is not the opt-out first-party Telemetry system or operator-configured OpenTelemetry tracing.
|
||||
- Authentication/model discovery does not establish tool, image, steering, or resumption capability parity.
|
||||
- Use [recovery](./recovery.md) for interrupted runs and [execution environments](./execution-environments.md) for local/SSH/sandbox differences.
|
||||
|
||||
Native setup retirement must join owned startup before cleanup can be confirmed.
|
||||
[Transport ownership controls](../../packages/paperclip-runner/src/live/runnerd-codex-transport.test.ts)
|
||||
hold credential preparation or registration without launching a provider, then
|
||||
verify bounded close, no later launch, and retention of unconfirmed state. A
|
||||
primary setup timeout remains a failure even when teardown also fails. These
|
||||
controls cover the lifecycle boundary; final packaged provider execution is
|
||||
separate evidence.
|
||||
@@ -69,6 +69,7 @@ import {
|
||||
codexSemanticToolSpecs,
|
||||
} from "../drivers/codex/codex-app-server-driver.js";
|
||||
import { releaseMaterializedNativeRuntimeSkills } from "../drivers/runtime-context-materializer.js";
|
||||
import * as runtimeMaterializer from "../drivers/runtime-context-materializer.js";
|
||||
import { RUNNERD_CANONICAL_ITEM } from "../drivers/codex/codex-driver-values.js";
|
||||
|
||||
import {
|
||||
@@ -2811,6 +2812,150 @@ function fakeCodexArgs(stateDirectory: string, ...args: string[]): string[] {
|
||||
];
|
||||
}
|
||||
|
||||
it("joins owned credential preparation before close returns and never launches after close", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-owned-startup-"));
|
||||
const artifact = join(root, "unexecuted-runner-artifact");
|
||||
await writeFile(artifact, "fixture artifact bytes; never executed");
|
||||
let release!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
let entered!: () => void;
|
||||
const started = new Promise<void>(resolve => { entered = resolve; });
|
||||
const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async ({ codexHome }) => {
|
||||
entered();
|
||||
await gate;
|
||||
await mkdir(codexHome, { recursive: true });
|
||||
await writeFile(join(codexHome, "owned-write.txt"), "completed owned preparation");
|
||||
});
|
||||
const registration = vi.fn(async () => ({ connectUrl: "ws://127.0.0.1:1", release: async () => {} }));
|
||||
const launch = vi.fn(() => { throw new Error("unexecuted late launch"); });
|
||||
const bundle = createCapabilityRunnerdCodexTransport({ stateDirectory: root, runnerBinary: artifact, codexCommand: artifact, provider: "codex", closeGraceMs: 1000,
|
||||
controlPlaneRegistration: registration, runnerProcessLauncher: launch });
|
||||
let opening: Promise<unknown> | undefined;
|
||||
let closing: Promise<void> | undefined;
|
||||
try {
|
||||
opening = bundle.transport.request("thread/start", { cwd: root, model: "gpt-test", dynamicTools: [] });
|
||||
void opening.catch(() => {});
|
||||
await started;
|
||||
let closeReturned = false;
|
||||
closing = bundle.transport.close("owned preparation test");
|
||||
void closing.then(() => { closeReturned = true; }, () => {});
|
||||
await new Promise(resolve => setTimeout(resolve, 25));
|
||||
expect(closeReturned).toBe(false);
|
||||
release();
|
||||
await expect(opening).rejects.toThrow(/closed/);
|
||||
await closing;
|
||||
expect(await readFile(join(root, "codex-home", "owned-write.txt"), "utf8")).toBe("completed owned preparation");
|
||||
expect(registration).not.toHaveBeenCalled();
|
||||
expect(launch).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
release();
|
||||
await Promise.allSettled([opening, closing]);
|
||||
preparation.mockRestore();
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("retains owned state when credential preparation outlives bounded close", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-retained-startup-"));
|
||||
const artifact = join(root, "unexecuted-artifact");
|
||||
await writeFile(artifact, "never executed");
|
||||
let release!: () => void, entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const started = new Promise<void>(resolve => { entered = resolve; });
|
||||
const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async () => { entered(); await gate; });
|
||||
const registration = vi.fn(async () => ({ connectUrl: "ws://127.0.0.1:1", release: async () => {} }));
|
||||
const launch = vi.fn(() => { throw new Error("unexpected launch"); });
|
||||
const bundle = createCapabilityRunnerdCodexTransport({ stateDirectory: root, runnerBinary: artifact, codexCommand: artifact, provider: "codex", closeGraceMs: 100,
|
||||
controlPlaneRegistration: registration, runnerProcessLauncher: launch });
|
||||
const opening = bundle.transport.request("thread/start", { cwd: root, model: "gpt-test", dynamicTools: [] });
|
||||
void opening.catch(() => {});
|
||||
try {
|
||||
await started;
|
||||
await expect(bundle.transport.close("bounded preparation")).rejects.toThrow("startup cleanup is unconfirmed");
|
||||
expect((await stat(root)).isDirectory()).toBe(true);
|
||||
release();
|
||||
await expect(opening).rejects.toThrow(/closed/);
|
||||
expect(registration).not.toHaveBeenCalled();
|
||||
expect(launch).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
release(); await opening.catch(() => {}); preparation.mockRestore();
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it.each(["settled", "late", "late-release-error"] as const)("retires owned startup registration without post-close launch (%s)", async mode => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-startup-registration-"));
|
||||
const artifact = join(root, "unexecuted-artifact");
|
||||
await writeFile(artifact, "never executed");
|
||||
let release!: () => void, entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const registered = new Promise<void>(resolve => { entered = resolve; });
|
||||
const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockResolvedValue(undefined);
|
||||
const retire = vi.fn(async () => { if (mode === "late-release-error") throw new Error("owned registration release failed"); });
|
||||
const registration = vi.fn(async () => { entered(); await gate; return { connectUrl: "ws://127.0.0.1:1", release: retire }; });
|
||||
const launch = vi.fn(() => { throw new Error("unexpected launch"); });
|
||||
const bundle = createCapabilityRunnerdCodexTransport({ stateDirectory: root, runnerBinary: artifact, codexCommand: artifact, provider: "codex", closeGraceMs: mode === "settled" ? 1000 : 100,
|
||||
controlPlaneRegistration: registration, runnerProcessLauncher: launch });
|
||||
const opening = bundle.transport.request("thread/start", { cwd: root, model: "gpt-test", dynamicTools: [] });
|
||||
void opening.catch(() => {});
|
||||
let closing: Promise<void> | undefined;
|
||||
try {
|
||||
await registered;
|
||||
let closeReturned = false;
|
||||
closing = bundle.transport.close("owned registration");
|
||||
void closing.then(() => { closeReturned = true; }, () => {});
|
||||
if (mode === "settled") {
|
||||
await new Promise(resolve => setTimeout(resolve, 25));
|
||||
expect(closeReturned).toBe(false);
|
||||
} else {
|
||||
await expect(closing).rejects.toThrow("startup cleanup is unconfirmed");
|
||||
expect((await stat(root)).isDirectory()).toBe(true);
|
||||
}
|
||||
release();
|
||||
await expect(opening).rejects.toThrow(mode === "late-release-error" ? "owned registration release failed" : /closed/);
|
||||
if (mode === "settled") await closing;
|
||||
expect(retire).toHaveBeenCalledTimes(1);
|
||||
expect(launch).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
release(); await Promise.allSettled([opening, closing]); preparation.mockRestore();
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
it("does not certify probe cleanup while owned credential preparation is unconfirmed", async () => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-probe-owned-preparation-"));
|
||||
const artifact = join(root, "unexecuted-artifact");
|
||||
await writeFile(artifact, "never executed");
|
||||
let release!: () => void, entered!: () => void;
|
||||
const gate = new Promise<void>(resolve => { release = resolve; });
|
||||
const started = new Promise<void>(resolve => { entered = resolve; });
|
||||
const preparation = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async () => { entered(); await gate; });
|
||||
const cleanup = vi.fn(async () => {});
|
||||
const launch = vi.fn(() => { throw new Error("unexpected launch"); });
|
||||
const probe = probeNativeRunnerEnvironment({ runtimeDirectory: root, provider: "codex", model: "gpt-test", timeoutMs: 100,
|
||||
environment: { OPENAI_API_KEY: "fixture" }, onCleanupConfirmed: cleanup,
|
||||
transportOptions: { runnerBinary: artifact, codexCommand: artifact, runnerProcessLauncher: launch,
|
||||
controlPlaneRegistration: async () => ({ connectUrl: "ws://127.0.0.1:1", release: async () => {} }) },
|
||||
});
|
||||
const result = probe.catch(error => error);
|
||||
try {
|
||||
await started;
|
||||
const error = await result;
|
||||
expect(cleanup).not.toHaveBeenCalled();
|
||||
expect(error).toBeInstanceOf(AggregateError);
|
||||
expect(error.message).toContain("hello probe timed out");
|
||||
expect(error.message).toContain("startup cleanup is unconfirmed");
|
||||
expect((await stat(root)).isDirectory()).toBe(true);
|
||||
release();
|
||||
await new Promise(resolve => setTimeout(resolve, 25));
|
||||
expect(launch).not.toHaveBeenCalled();
|
||||
} finally {
|
||||
release(); await result; preparation.mockRestore();
|
||||
await new Promise(resolve => setTimeout(resolve, 25));
|
||||
await rm(root, { recursive: true, force: true });
|
||||
}
|
||||
}, 15_000);
|
||||
|
||||
it.each(["complete", "api-key", "default-model", "mismatched-model", "model-refresh-error", "model-cleanup-error", "missing-model", "missing-runtime", "missing-completion", "late-bootstrap", "remote-complete", "remote-mismatched-model", "late-remote-preparation", "late-remote-release-error"] as const)("native Codex setup requires its daemon/app-server hello (%s)", async mode => {
|
||||
const root = await mkdtemp(join(tmpdir(), "runnerd-native-setup-"));
|
||||
const runtimeDirectory = join(root, "native-state");
|
||||
|
||||
@@ -3602,6 +3602,8 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
#runAttachTemplate: Record<string, unknown> | null = null;
|
||||
#closed = false;
|
||||
#closePromise: Promise<void> | null = null;
|
||||
#startupOperation: Promise<unknown> | null = null;
|
||||
#startupDrainTimedOut = false;
|
||||
#controllerDetachedForRestart = false;
|
||||
#failure: Error | null = null;
|
||||
readonly #failureSignal: Promise<never>;
|
||||
@@ -3715,7 +3717,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
throw new Error("native_runner_warm_transition_completion_pending");
|
||||
}
|
||||
if (method === "initialize") return { user: {} };
|
||||
if (method === "thread/start") return this.#start(params);
|
||||
if (method === "thread/start") return this.#trackStartup(this.#start(params));
|
||||
if (method === "collaborationMode/list") {
|
||||
// runnerd negotiates the real Codex preset or the provider-proxy-owned
|
||||
// planning contract during session.open. This transport-level mask
|
||||
@@ -3815,7 +3817,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
if (method === "thread/read") {
|
||||
if (this.#core === null) {
|
||||
this.#recoveryTurnBindingPending = true;
|
||||
await this.#resume();
|
||||
await this.#trackStartup(this.#resume());
|
||||
}
|
||||
// Ask the authenticated runner for its live provider snapshot rather
|
||||
// than reading its filesystem. This both supports remote process owners
|
||||
@@ -4455,8 +4457,33 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
);
|
||||
}
|
||||
|
||||
#trackStartup<T>(operation: Promise<T>): Promise<T> {
|
||||
this.#startupOperation = operation;
|
||||
return operation;
|
||||
}
|
||||
|
||||
#assertStartupOpen(): void {
|
||||
if (this.#closed) throw new Error("runnerd transport closed during provider startup");
|
||||
}
|
||||
|
||||
async #closeOnce(): Promise<void> {
|
||||
this.#closed = true;
|
||||
const { preparationDeadline, closeDeadline } = runnerCloseDeadlines(Date.now(), runnerCloseGraceMs(this.options));
|
||||
let startupConfirmed = true;
|
||||
if (this.#startupOperation !== null) {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
await Promise.race([
|
||||
this.#startupOperation.catch(() => undefined),
|
||||
new Promise<never>((_resolve, reject) => {
|
||||
timer = setTimeout(() => reject(new Error("Native startup drain timed out")), Math.max(0, preparationDeadline - Date.now()));
|
||||
}),
|
||||
]);
|
||||
} catch {
|
||||
startupConfirmed = false;
|
||||
this.#startupDrainTimedOut = true;
|
||||
} finally { clearTimeout(timer); }
|
||||
}
|
||||
this.#turnStartAdmission?.resolve(false);
|
||||
const adoptedRunner = this.options.adoptExistingRunner;
|
||||
// `settled` is a durable-state assertion, not merely the absence of a
|
||||
@@ -4481,10 +4508,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
// A terminal provider frame can become visible one control loop before
|
||||
// its durable provider suffix is ACKed. Drain it before suspension so a
|
||||
// fresh run authority never inherits the prior run's pending events.
|
||||
const { preparationDeadline, closeDeadline } = runnerCloseDeadlines(
|
||||
Date.now(),
|
||||
runnerCloseGraceMs(this.options),
|
||||
);
|
||||
|
||||
if (!(await this.#runnerHasExited())) {
|
||||
// Let an already-admitted tool result reach its original provider
|
||||
// before turn.stop can retire that tool-call identity. This shares the
|
||||
@@ -4611,7 +4635,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
// an incomplete or identity-conflicting state remains fail-closed.
|
||||
const finalProviderState = this.#providerDrainState();
|
||||
const runnerSettled =
|
||||
runnerSuspended &&
|
||||
startupConfirmed && runnerSuspended &&
|
||||
providerDrained &&
|
||||
this.#core?.semanticToolResultsSettled() === true &&
|
||||
(finalProviderState === null ||
|
||||
@@ -4663,6 +4687,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
throw new NativeSessionCloseUnrecoverableError(settlement);
|
||||
}
|
||||
if (!startupConfirmed) {
|
||||
throw new Error("Native runner startup cleanup is unconfirmed; recovery state retained.");
|
||||
}
|
||||
if (this.#ownsRoot && !adoptedRunner) {
|
||||
rmSync(this.#root, { recursive: true, force: true });
|
||||
}
|
||||
@@ -4756,6 +4783,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
nativeMcp: nativeMcpLaunchBinding(this.options.environment),
|
||||
});
|
||||
}
|
||||
this.#assertStartupOpen();
|
||||
const opencodeProxyPath =
|
||||
this.options.opencodeProxyPath ??
|
||||
(provider === "opencode" && !this.options.runnerFilesystemRoot
|
||||
@@ -5056,6 +5084,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#controlPlaneCheckpoint = registration.checkpoint ?? null;
|
||||
this.#controlPlaneRelease = registration.release;
|
||||
}
|
||||
if (this.#closed && this.#startupDrainTimedOut) {
|
||||
const release = this.#controlPlaneRelease;
|
||||
this.#controlPlaneRelease = null;
|
||||
await release?.();
|
||||
}
|
||||
this.#assertStartupOpen();
|
||||
const admissionDeadline = this.#coldAdmissionDeadline();
|
||||
const handle = spawnRunner({
|
||||
connection: registration?.connection ?? {
|
||||
@@ -5098,7 +5132,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
this.#handle = handle;
|
||||
this.#watchRunner(handle);
|
||||
await this.#publishSpawnedProcess(handle);
|
||||
this.#assertStartupOpen();
|
||||
await registration?.activate?.();
|
||||
this.#assertStartupOpen();
|
||||
if (registration?.failure) {
|
||||
void registration.failure.catch((error: unknown) => {
|
||||
this.#failTransport(
|
||||
@@ -5107,6 +5143,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
});
|
||||
}
|
||||
await this.#awaitRegistrationReady(registration?.ready);
|
||||
this.#assertStartupOpen();
|
||||
this.#evidence.runnerPid = handle.child.pid ?? null;
|
||||
this.#evidence.runnerProcessGroupId = handle.processGroupId ?? null;
|
||||
this.#publish();
|
||||
@@ -5633,6 +5670,12 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
: registration.release;
|
||||
}
|
||||
if (this.#closed && this.#startupDrainTimedOut) {
|
||||
const release = this.#controlPlaneRelease;
|
||||
this.#controlPlaneRelease = null;
|
||||
await release?.();
|
||||
}
|
||||
this.#assertStartupOpen();
|
||||
// Route and immutable endpoint admission precedes every launch-material
|
||||
// write. A refused transition leaves its retained home/context untouched.
|
||||
if (runtimeContext !== null) {
|
||||
@@ -5664,6 +5707,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
if (warmRecovery) {
|
||||
await this.options.authorizeWarmTransitionRecovery?.("before_bootstrap");
|
||||
}
|
||||
this.#assertStartupOpen();
|
||||
const bootstrapTicket = adoptedRunner
|
||||
? null
|
||||
: warmRecovery
|
||||
@@ -5679,6 +5723,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
await this.options.authorizeWarmTransitionRecovery?.("before_spawn");
|
||||
}
|
||||
// A replacement executor restores its cold provider before acknowledging recovery.
|
||||
this.#assertStartupOpen();
|
||||
const admissionDeadline = this.#coldAdmissionDeadline(adoptedRunner === undefined);
|
||||
const handle = adoptedRunner
|
||||
? null
|
||||
|
||||
Reference in new issue
Block a user