mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 19:28:46 +01:00
merge: reconcile canonical startup ownership repair
Preserve the source prerequisite repair and Claude effort/resume contracts. Eight no-process controls and runner noemit pass; native readiness remains a hosted CI gate. Co-Authored-By: Paperclip <noreply@paperclip.ing>
This commit is contained in:
commit
e3a8171679
2 files changed
+198
-8
No files matched your search
@@ -396,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);
|
||||
@@ -403,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);
|
||||
@@ -2177,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");
|
||||
@@ -8009,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({
|
||||
|
||||
@@ -3601,6 +3601,8 @@ 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;
|
||||
@@ -3641,6 +3643,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.
|
||||
@@ -3717,7 +3722,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.#trackStartup(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
|
||||
@@ -3816,8 +3821,10 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
}
|
||||
if (method === "thread/read") {
|
||||
if (this.#core === null) {
|
||||
this.#recoveryTurnBindingPending = true;
|
||||
await this.#trackStartup(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
|
||||
@@ -4431,6 +4438,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;
|
||||
@@ -4457,7 +4465,16 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
);
|
||||
}
|
||||
|
||||
#trackStartup<T>(operation: Promise<T>): Promise<T> {
|
||||
#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;
|
||||
}
|
||||
@@ -4468,6 +4485,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
|
||||
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) {
|
||||
@@ -5598,6 +5616,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(
|
||||
@@ -5616,6 +5635,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
"native_runner_warm_transition_registered_endpoint_mismatch",
|
||||
);
|
||||
}
|
||||
this.#assertStartupOpen();
|
||||
newTransitionRegistration = await this.options.controlPlaneRegistration(
|
||||
core,
|
||||
warmRecovery.proof.receipt.newIdentity,
|
||||
@@ -5691,6 +5711,7 @@ class DurablePrpCodexTransport implements CodexAppServerTransport {
|
||||
await releaseMaterializedNativeRuntimeSkills(
|
||||
resolve(localCodexHome, "skills"),
|
||||
);
|
||||
this.#assertStartupOpen();
|
||||
await prepareIsolatedCodexHome({
|
||||
context: sourceRuntimeContext,
|
||||
codexHome: localCodexHome,
|
||||
@@ -5769,10 +5790,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(
|
||||
@@ -5781,6 +5805,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;
|
||||
@@ -5797,6 +5822,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");
|
||||
}
|
||||
@@ -5811,11 +5837,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(
|
||||
@@ -6855,8 +6883,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,
|
||||
});
|
||||
@@ -6865,6 +6896,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`,
|
||||
@@ -7069,11 +7101,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();
|
||||
}
|
||||
|
||||
|
||||
Reference in new issue
Block a user