fix(runner): retain startup ownership through cancellation

Reuse the shared startup repair with duplicate admission, shutdown-aware readiness, and resume fences. Preserve absolute cleanup deadlines and unconfirmed recovery state.

Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
DottaandPaperclip committed 2026-10-12 19:59:20 -05:00
1 parent 5a9ce11c49
commit ce23716503
2 files changed
+392 -12

No files matched your search

@@ -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 {
@@ -395,6 +396,9 @@ it("publishes spawned runner ownership before waiting for provider startup", asy
});
const opening = bundle.transport.request("thread/start", { cwd: tmpdir(), dynamicTools: codexSemanticToolSpecs() });
try {
// Same-tick recovery must not mutate an already admitted fresh startup.
await expect(bundle.transport.request("thread/read", { threadId: "rejected-duplicate" }))
.rejects.toThrow("PRP provider thread is already started");
await vi.waitFor(() => expect(onSpawn).toHaveBeenCalledOnce());
expect(onSpawn).toHaveBeenCalledWith({ pid: expect.any(Number), processGroupId: expect.any(Number), startedAt: expect.any(String) });
expect(bundle.transport.processInfo?.().pid).toBeGreaterThan(0);
@@ -402,13 +406,17 @@ it("publishes spawned runner ownership before waiting for provider startup", asy
releaseOwnership();
await opening;
expect(activate).toHaveBeenCalledOnce();
const turnStarted = expectTurnStarted(bundle.transport.notifications()[Symbol.asyncIterator]());
void turnStarted.catch(() => undefined);
await bundle.transport.request("turn/start", { input: [{ type: "text", text: "Verify fixture turn delivery." }] });
await turnStarted;
} finally {
releaseOwnership();
await opening.catch(() => undefined);
await bundle.transport.close();
await rm(stateDirectory, { recursive: true, force: true });
}
});
}, 30_000);
it("launches runnerd with its production durable outbox limits", () => {
expect(runnerdLaunchProfileInternals.maxOutboxBytes).toBe(16 * 1024 * 1024);
@@ -2176,6 +2184,115 @@ it.each(["installed", "explicit"] as const)("records the selected %s Codex comma
}
});
it("does not invoke deferred startup preparation after immediate close", async () => {
const prepare = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockResolvedValue(undefined);
const registration = vi.fn(async () => ({ connectUrl: "ws://127.0.0.1:1", release: () => undefined }));
const launch = vi.fn(() => { throw new Error("unexpected launch"); });
const { transport } = createCapabilityRunnerdCodexTransport({
runnerBinary: resolve(import.meta.dirname, "../../package.json"), codexCommand: "/unused-fixture-command", sourceCodexHome: "",
controlPlaneRegistration: registration, runnerProcessLauncher: launch,
});
const opening = transport.request("thread/start", { cwd: tmpdir(), dynamicTools: [] });
void opening.catch(() => undefined);
try {
await transport.close();
await expect(opening).rejects.toThrow(/closed/);
expect(prepare).not.toHaveBeenCalled();
expect(registration).not.toHaveBeenCalled();
expect(launch).not.toHaveBeenCalled();
} finally {
await opening.catch(() => undefined);
await transport.close().catch(() => undefined);
prepare.mockRestore();
}
});
it("owns blocked Codex home preparation through driver abort without later registration or launch", async () => {
const root = await mkdtemp(join(tmpdir(), "runnerd-blocked-home-preparation-"));
const runnerBinary = join(root, "runnerd");
const command = join(root, "codex");
await writeFile(runnerBinary, "unexecuted fixture artifact", { mode: 0o700 });
await writeFile(command, "#!/bin/sh\nexit 99\n", { mode: 0o700 });
let releasePreparation!: () => void;
const preparationReleased = new Promise<void>(resolve => { releasePreparation = resolve; });
let markPreparationStarted!: () => void;
const preparationStarted = new Promise<void>(resolve => { markPreparationStarted = resolve; });
let closeSettled = false;
let preparationContinuedAfterClose = false;
const prepare = vi.spyOn(runtimeMaterializer, "prepareIsolatedCodexHome").mockImplementation(async input => {
markPreparationStarted();
await preparationReleased;
preparationContinuedAfterClose = closeSettled;
// A synthetic, credential-free write makes the owned preparation visible.
await mkdir(input.codexHome, { mode: 0o700 });
await writeFile(join(input.codexHome, "preparation-marker"), "fixture preparation drained\n", { mode: 0o600 });
});
const registration = vi.fn(async () => ({
connection: { mode: "connect" as const, connectUrl: "ws://127.0.0.1:1/fixture-never-connected" },
release: () => undefined,
}));
const launch = vi.fn(() => { throw new Error("fixture_stop_before_runner_launch"); });
const { transport } = createCapabilityRunnerdCodexTransport({
provider: "codex", runnerBinary, codexCommand: command, stateDirectory: root,
sourceCodexHome: "", environment: {}, controlPlaneRegistration: registration, runnerProcessLauncher: launch,
});
const request = transport.request.bind(transport);
let bootstrapOperation: Promise<Record<string, unknown>> | undefined;
const requestSpy = vi.spyOn(transport, "request").mockImplementation((method, params) => {
const operation = request(method, params);
if (method === "thread/start") bootstrapOperation = operation;
return operation;
});
const closeTransport = transport.close.bind(transport);
let closeOperation: Promise<void> | undefined;
const closeSpy = vi.spyOn(transport, "close").mockImplementation(reason => {
closeOperation ??= closeTransport(reason).then(() => { closeSettled = true; });
return closeOperation;
});
const controller = new AbortController();
const cancelled = new Error("fixture Codex bootstrap cancelled");
const driver = new CodexAppServerDriver({
taskEnvelope: {
schema: "paperclip.skillless_task.v1", objective: "Credential-free bootstrap ownership fixture.",
completionContract: { revision: "blocked-preparation-v1", criteria: [] },
constraints: [], expectedResultSchema: "paperclip.run_result.v1",
},
conversationMode: "direct", environment: {}, transportFactory: () => transport,
});
let openingSettled = false;
const opening = driver.openSession({ runId: "fixture-blocked-home", normalizedSessionId: "fixture-blocked-home", workingDirectory: root, signal: controller.signal })
.then(value => { openingSettled = true; return { value }; }, error => { openingSettled = true; return { error }; });
try {
await preparationStarted;
// A rejected duplicate must not replace the first admitted startup owner.
await expect(transport.request("thread/start", { cwd: root, dynamicTools: [] }))
.rejects.toThrow("PRP provider thread is already started");
controller.abort(cancelled);
await new Promise<void>(resolve => setImmediate(resolve));
expect.soft(closeOperation).toBeDefined();
expect.soft(closeSettled).toBe(false);
expect.soft(openingSettled).toBe(false);
releasePreparation();
const outcome = await opening;
await closeOperation;
await bootstrapOperation?.catch(() => undefined);
expect.soft(preparationContinuedAfterClose).toBe(false);
expect.soft(registration.mock.calls.length).toBe(0);
expect.soft(launch.mock.calls.length).toBe(0);
expect.soft(outcome).toHaveProperty("error");
expect.soft(outcome).toHaveProperty("error", cancelled);
} finally {
releasePreparation();
await opening;
await bootstrapOperation?.catch(() => undefined);
await transport.close().catch(() => undefined);
requestSpy.mockRestore();
closeSpy.mockRestore();
prepare.mockRestore();
await rm(root, { recursive: true, force: true });
}
});
it("rejects remote Codex without a guest executable before resolving controller dependencies", async () => {
const root = await mkdtemp(join(tmpdir(), "paperclip-runner-remote-codex-command-"));
const runnerBinary = join(root, "runnerd");
@@ -2811,6 +2928,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");
@@ -7864,6 +8125,41 @@ it("keeps an adopted runner alive when the detached controller finalizer closes"
it("adopts an opening session without a checkpoint instead of bootstrapping a duplicate provider", () => verifyLiveRunnerAdoption(false, false, false, false, true), 30_000);
it("cancels startup readiness before releasing the route that settles its ready callback", async () => {
const root = await mkdtemp(join(tmpdir(), "runnerd-ready-release-cycle-"));
let releaseReady!: () => void;
const readyReleased = new Promise<void>(resolve => { releaseReady = resolve; });
let markReadyStarted!: () => void;
const readyStarted = new Promise<void>(resolve => { markReadyStarted = resolve; });
const release = vi.fn(async () => { releaseReady(); });
let core!: DurablePrpControlPlane;
const bundle = createCapabilityRunnerdCodexTransport({
runnerBinary: defaultCapabilityRunnerdBinary(), codexCommand: fakeCodex,
codexArgs: fakeCodexArgs(root), stateDirectory: root, sourceCodexHome: "", environment: {}, closeGraceMs: 2_000,
controlPlaneRegistration: async authority => {
core = authority;
await authority.start();
return { ready: () => { markReadyStarted(); return readyReleased; }, release };
},
});
const opening = bundle.transport.request("thread/start", { cwd: root, dynamicTools: [] });
void opening.catch(() => undefined);
try {
await readyStarted;
await vi.waitFor(() => expect(core.store.state.commands.find(command => command.type === "session.open")?.status).toBe("completed"));
await expect(bundle.transport.close("cancel pending ingress authentication")).resolves.toBeUndefined();
await expect(opening).rejects.toThrow(/closed/);
expect(release).toHaveBeenCalledOnce();
expect(bundle.evidence().runnerExited).toBe(true);
expect(bundle.evidence().diagnostics.join("\n")).not.toContain("startup cleanup is unconfirmed");
} finally {
releaseReady();
await opening.catch(() => undefined);
await bundle.transport.close().catch(() => undefined);
await rm(root, { recursive: true, force: true });
}
}, 30_000);
it("surfaces a runner exit while provider-ingress readiness is still pending", async () => {
const neverReady = new Promise<void>(() => undefined);
const bundle = createCapabilityRunnerdCodexTransport({
@@ -3598,7 +3598,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
#authorizedTools: Record<string, unknown> | null = null;
#runAttachTemplate: Record<string, unknown> | null = null;
#closed = false;
readonly #closedSignal: Promise<void>;
#resolveClosedSignal!: () => void;
#closePromise: Promise<void> | null = null;
#startupOperation: Promise<unknown> | null = null;
#startupDrainTimedOut = false;
#controllerDetachedForRestart = false;
#failure: Error | null = null;
readonly #failureSignal: Promise<never>;
@@ -3635,6 +3639,9 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
this.#failureSignal = new Promise<never>((_resolve, reject) => {
this.#rejectFailureSignal = reject;
});
this.#closedSignal = new Promise<void>(resolve => {
this.#resolveClosedSignal = resolve;
});
// Failure is also observed by request/notification paths. Register an
// internal handler so a process exit after the owner has closed the
// session cannot become an unhandled process-level rejection.
@@ -3711,7 +3718,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
@@ -3810,8 +3817,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
}
if (method === "thread/read") {
if (this.#core === null) {
this.#recoveryTurnBindingPending = true;
await this.#resume();
await this.#trackStartup(() => {
this.#recoveryTurnBindingPending = true;
return this.#resume();
});
}
// Ask the authenticated runner for its live provider snapshot rather
// than reading its filesystem. This both supports remote process owners
@@ -4425,6 +4434,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
if (this.#closed) return;
this.#controllerDetachedForRestart = true;
this.#closed = true;
this.#resolveClosedSignal();
this.#turnStartAdmission?.resolve(false);
if (this.#pump !== null) clearInterval(this.#pump);
this.#pump = null;
@@ -4451,8 +4461,43 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
);
}
#trackStartup<T>(start: () => Promise<T>): Promise<T> {
// Admit one owner before invoking startup. A duplicate request's rejection
// must never replace the preparation that close is responsible for joining.
if (this.#startupOperation !== null) {
return Promise.reject(new Error("PRP provider thread is already started"));
}
const operation = Promise.resolve().then(() => {
this.#assertStartupOpen();
return start();
});
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;
this.#resolveClosedSignal();
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
@@ -4477,10 +4522,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
@@ -4607,7 +4649,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 ||
@@ -4659,6 +4701,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 });
}
@@ -4752,6 +4797,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
nativeMcp: nativeMcpLaunchBinding(this.options.environment),
});
}
this.#assertStartupOpen();
const opencodeProxyPath =
this.options.opencodeProxyPath ??
(provider === "opencode" && !this.options.runnerFilesystemRoot
@@ -5051,6 +5097,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 ?? {
@@ -5093,7 +5145,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(
@@ -5102,6 +5156,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();
@@ -5556,6 +5611,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
>;
let oldTransitionRegistration: RecoveryRegistration | null = null;
let newTransitionRegistration: RecoveryRegistration | null = null;
this.#assertStartupOpen();
if (warmRecovery && this.options.controlPlaneRegistration) {
try {
oldTransitionRegistration = await this.options.controlPlaneRegistration(
@@ -5574,6 +5630,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
"native_runner_warm_transition_registered_endpoint_mismatch",
);
}
this.#assertStartupOpen();
newTransitionRegistration = await this.options.controlPlaneRegistration(
core,
warmRecovery.proof.receipt.newIdentity,
@@ -5628,6 +5685,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) {
@@ -5643,6 +5706,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
await releaseMaterializedNativeRuntimeSkills(
resolve(localCodexHome, "skills"),
);
this.#assertStartupOpen();
await prepareIsolatedCodexHome({
context: sourceRuntimeContext,
codexHome: localCodexHome,
@@ -5659,6 +5723,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
if (warmRecovery) {
await this.options.authorizeWarmTransitionRecovery?.("before_bootstrap");
}
this.#assertStartupOpen();
const bootstrapTicket = adoptedRunner
? null
: warmRecovery
@@ -5674,6 +5739,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
@@ -5719,10 +5785,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
this.#watchRunner(handle);
await this.#publishSpawnedProcess(handle);
}
this.#assertStartupOpen();
if (oldTransitionRegistration && newTransitionRegistration) {
await oldTransitionRegistration.activate?.();
this.#assertStartupOpen();
await newTransitionRegistration.activate?.();
} else await registration?.activate?.();
this.#assertStartupOpen();
if (registration?.failure) {
void registration.failure.catch((error: unknown) => {
this.#failTransport(
@@ -5731,6 +5800,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
});
}
if (!adoptedRunner) await this.#awaitRegistrationReady(registration?.ready);
this.#assertStartupOpen();
if (adoptedRunner) {
this.#evidence.runnerPid = adoptedRunner.pid;
this.#evidence.runnerProcessGroupId = adoptedRunner.processGroupId;
@@ -5747,6 +5817,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
!recoveryIdentityMatches(core.store.state.identity, desiredIdentity) ||
core.store.state.warmTransition !== undefined
) {
this.#assertStartupOpen();
if (Date.now() >= deadline || (await this.#runnerHasExited())) {
throw new Error("native_runner_warm_transition_recovery_pending");
}
@@ -5761,11 +5832,13 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
this.#deferredTurnStartEvents = [];
if (oldTransitionRegistration && newTransitionRegistration) {
await oldTransitionRegistration.release();
this.#assertStartupOpen();
this.#controlPlaneRelease = newTransitionRegistration.release;
this.#controlPlaneCheckpoint =
newTransitionRegistration.checkpoint ?? null;
}
}
this.#assertStartupOpen();
this.#pump = setInterval(() => this.#pumpEventsSafely(), 5);
if (adoptedRunner) {
await this.#awaitAdoptedRunnerConnection(
@@ -6803,8 +6876,11 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
await awaitAdoptedRunnerAuthentication({
activeConnectionCount: () => core.activeRunnerConnectionCount(),
isAlive: () => adoptedRunner.isAlive(),
throwIfFailed: () => this.#throwIfFailed(),
failure: this.#failureSignal,
throwIfFailed: () => { this.#assertStartupOpen(); this.#throwIfFailed(); },
failure: Promise.race([
this.#failureSignal,
this.#closedSignal.then(() => { throw new Error("runnerd transport closed during provider startup"); }),
]),
ready,
timeoutMs: this.options.runnerReconnectGraceMs ?? 30_000,
});
@@ -6813,6 +6889,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
this.#failTransport(failure);
throw failure;
}
this.#assertStartupOpen();
this.#adoptedRunnerAuthenticated = true;
this.#diagnostic(
`adopted runner ${adoptedRunner.pid} authenticated to its durable PRP authority`,
@@ -7017,11 +7094,18 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
async #awaitRegistrationReady(
ready: (() => Promise<void>) | undefined,
): Promise<void> {
this.#assertStartupOpen();
if (ready === undefined) {
this.#throwIfFailed();
return;
}
await Promise.race([ready(), this.#failureSignal]);
await Promise.race([
ready(), this.#failureSignal,
// Readiness is a notification wait, not launch-material preparation.
// Its route owner is released by close; joining it first would cycle.
this.#closedSignal.then(() => { this.#assertStartupOpen(); }),
]);
this.#assertStartupOpen();
this.#throwIfFailed();
}