RomanNum3ral
/
paperclip
mirror of https://github.com/paperclipai/paperclip.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/master' into fix/runner-paid-matrix-integrity 

* origin/master:
  fix(runner): retain aborted admission cleanup (#12755)

# Conflicts:
#	packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts
This commit is contained in:
Dotta 2026-09-03 03:26:17 -05:00
parent 1f5e252238 f1d9206c4a
commit 7462d142dd
2 changed files with 298 additions and 107 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

333
packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts
View File

@ -11,6 +11,7 @@ import type {
VerifiedAcpxInstallation,
} from "./installation-integrity.js";
import { resolveQualifiedAcpxProfile } from "./qualified-profiles.js";
import { prepareAcpxRuntimeSandbox } from "./runtime-sandbox.js";
import {
AcpxRuntimeHost,
type AcpxRuntimeHostDependencies,
@ -20,8 +21,18 @@ import {
} from "./runtime-host.js";
const temporaryDirectories: string[] = [];
const admissionControllers: AbortController[] = [];
const pendingAdmissionOpenings = new Set<Promise<void>>();
const pendingAdmissionCleanups = new Set<Promise<void>>();
afterEach(async () => {
for (const controller of admissionControllers.splice(0)) {
if (!controller.signal.aborted) {
controller.abort(new Error("ACPX runtime host test cleanup"));
}
}
await Promise.all([...pendingAdmissionOpenings]);
await Promise.all([...pendingAdmissionCleanups]);
await Promise.all(
temporaryDirectories
.splice(0)
@ -32,7 +43,7 @@ afterEach(async () => {
describe("ACPX runtime host", () => {
it("rejects a pre-aborted admission before acquiring provider resources", async () => {
const fixture = await hostFixture();
const controller = new AbortController();
const controller = trackedAdmissionController();
const cancellation = new Error("admission cancelled before start");
controller.abort(cancellation);
const openRuntime = vi.fn(async () => runtimePort());
@ -61,9 +72,115 @@ describe("ACPX runtime host", () => {
expect(fixture.commandClose).not.toHaveBeenCalled();
});
it("retains aborted sandbox preparation until its filesystem work settles", async () => {
const fixture = await hostFixture();
const controller = trackedAdmissionController();
const cancellation = new Error("sandbox admission cancelled");
const sandboxStarted = deferred<void>();
const finishSandbox = deferred<void>();
const retainedCleanups: Promise<void>[] = [];
const stageCredential = vi.fn(async () => {
throw new Error("credential staging must not start");
});
const openRuntime = vi.fn(async () => runtimePort());
const dependencies = fixture.dependencies({ openRuntime });
dependencies.stageCredential = stageCredential;
dependencies.prepareSandbox = async (input) => {
sandboxStarted.resolve(undefined);
await finishSandbox.promise;
return await prepareAcpxRuntimeSandbox(input);
};
dependencies.retainAdmissionCleanup = (cleanup) => {
retainedCleanups.push(cleanup);
trackAdmissionCleanup(cleanup);
};
const opening = trackAdmissionOpening(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
signal: controller.signal,
},
dependencies,
),
);
await sandboxStarted.promise;
controller.abort(cancellation);
await expect(opening).rejects.toBe(cancellation);
expect(retainedCleanups).toHaveLength(2);
let sandboxCleanupSettled = false;
void retainedCleanups[0]!.then(
() => {
sandboxCleanupSettled = true;
},
() => {
sandboxCleanupSettled = true;
},
);
await Promise.resolve();
expect(sandboxCleanupSettled).toBe(false);
finishSandbox.resolve(undefined);
await expect(retainedCleanups[0]).resolves.toBeUndefined();
expect(sandboxCleanupSettled).toBe(true);
expect(stageCredential).not.toHaveBeenCalled();
expect(openRuntime).not.toHaveBeenCalled();
expect(fixture.commandClose).not.toHaveBeenCalled();
});
it("settles retained admission work when aborted sandbox preparation rejects", async () => {
const fixture = await hostFixture();
const controller = trackedAdmissionController();
const cancellation = new Error("sandbox admission cancelled");
const sandboxStarted = deferred<void>();
const finishSandbox = deferred<void>();
const retainedCleanups: Promise<void>[] = [];
const stageCredential = vi.fn(async () => {
throw new Error("credential staging must not start");
});
const openRuntime = vi.fn(async () => runtimePort());
const dependencies = fixture.dependencies({ openRuntime });
dependencies.stageCredential = stageCredential;
dependencies.prepareSandbox = async (input) => {
sandboxStarted.resolve(undefined);
await finishSandbox.promise;
return await prepareAcpxRuntimeSandbox(input);
};
dependencies.retainAdmissionCleanup = (cleanup) => {
retainedCleanups.push(cleanup);
trackAdmissionCleanup(cleanup);
};
const opening = trackAdmissionOpening(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
signal: controller.signal,
},
dependencies,
),
);
await sandboxStarted.promise;
controller.abort(cancellation);
await expect(opening).rejects.toBe(cancellation);
expect(retainedCleanups).toHaveLength(2);
finishSandbox.reject(new Error("sandbox preparation failed after abort"));
await expect(retainedCleanups[0]).resolves.toBeUndefined();
expect(stageCredential).not.toHaveBeenCalled();
expect(openRuntime).not.toHaveBeenCalled();
expect(fixture.commandClose).not.toHaveBeenCalled();
});
it("scrubs credentials when abort wins before the adapter body starts", async () => {
const fixture = await hostFixture();
const controller = new AbortController();
const controller = trackedAdmissionController();
const cancellation = new Error("runtime admission cancelled before entry");
const createRuntime = vi.fn();
let credentialHome = "";
@ -919,32 +1036,39 @@ describe("ACPX runtime host", () => {
it("closes a command lease that resolves after admission is aborted", async () => {
const fixture = await hostFixture();
const commandAdmission = deferred<VerifiedAcpxCommandLease>();
const commandAdmissionStarted = deferred<void>();
const lateCommandClose = vi.fn(async () => undefined);
const openCommand = vi.fn(() => commandAdmission.promise);
const openCommand = vi.fn(() => {
commandAdmissionStarted.resolve(undefined);
return commandAdmission.promise;
});
const openRuntime = vi.fn(async () => runtimePort());
const controller = new AbortController();
const controller = trackedAdmissionController();
const cancellation = new Error("command admission cancelled");
const profile = resolveQualifiedAcpxProfile("claude", "claude-sonnet-5");
const opening = AcpxRuntimeHost.open(
{
...fixture.options,
agent: "claude",
model: "claude-sonnet-5",
permissionMode: "deny-all",
signal: controller.signal,
},
{
verifyInstallation: async () => ({
commandDigest: profile.commandDigest,
agentServerPackageJsonPath: join(fixture.root, "package.json"),
agentRuntimePackageJsonPath: null,
openCommand,
}),
openRuntime,
reportRetainedCleanupFailure: vi.fn(),
},
const opening = trackAdmissionOpening(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "claude",
model: "claude-sonnet-5",
permissionMode: "deny-all",
signal: controller.signal,
},
{
verifyInstallation: async () => ({
commandDigest: profile.commandDigest,
agentServerPackageJsonPath: join(fixture.root, "package.json"),
agentRuntimePackageJsonPath: null,
openCommand,
}),
openRuntime,
retainAdmissionCleanup: trackAdmissionCleanup,
reportRetainedCleanupFailure: vi.fn(),
},
),
);
await vi.waitFor(() => expect(openCommand).toHaveBeenCalledOnce());
await commandAdmissionStarted.promise;
controller.abort(cancellation);
await expect(opening).rejects.toBe(cancellation);
@ -978,29 +1102,33 @@ describe("ACPX runtime host", () => {
await rm(lateCredentialPath);
});
const reportRetainedCleanupFailure = vi.fn();
const stageCredential = vi.fn(() => credentialAdmission.promise);
const openRuntime = vi.fn(async () => runtimePort());
const controller = new AbortController();
const cancellation = new Error("credential admission cancelled");
const opening = AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
signal: controller.signal,
},
{
...fixture.dependencies({
openRuntime,
reportRetainedCleanupFailure,
}),
stageCredential,
},
);
await vi.waitFor(() => expect(stageCredential).toHaveBeenCalledOnce(), {
timeout: 5_000,
const credentialAdmissionStarted = deferred<void>();
const stageCredential = vi.fn(() => {
credentialAdmissionStarted.resolve(undefined);
return credentialAdmission.promise;
});
const openRuntime = vi.fn(async () => runtimePort());
const controller = trackedAdmissionController();
const cancellation = new Error("credential admission cancelled");
const opening = trackAdmissionOpening(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
signal: controller.signal,
},
{
...fixture.dependencies({
openRuntime,
reportRetainedCleanupFailure,
}),
stageCredential,
},
),
);
await credentialAdmissionStarted.promise;
controller.abort(cancellation);
await expect(opening).rejects.toBe(cancellation);
@ -1033,6 +1161,7 @@ describe("ACPX runtime host", () => {
it("retains managed credentials until an aborted late runtime is closed", async () => {
const fixture = await hostFixture();
const runtimeAdmission = deferred<AcpxRuntimePort>();
const runtimeAdmissionStarted = deferred<void>();
const retryClose = deferred<void>();
const lateRuntime = runtimePort({
onClose: vi
@ -1047,33 +1176,31 @@ describe("ACPX runtime host", () => {
receivedSignal = options.signal;
credentialHome = options.launchEnvironment.CODEX_HOME!;
bridgeUrl = options.mcpServers[0]!.url;
runtimeAdmissionStarted.resolve(undefined);
return runtimeAdmission.promise;
});
const controller = new AbortController();
const controller = trackedAdmissionController();
const cancellation = new Error("runtime admission cancelled");
const opening = AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
environment: {
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}",
const opening = trackAdmissionOpening(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
environment: {
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}",
},
signal: controller.signal,
semanticTools: {
tools: [],
handler: async () => ({ ok: true }),
},
},
signal: controller.signal,
semanticTools: {
tools: [],
handler: async () => ({ ok: true }),
},
},
fixture.dependencies({ openRuntime }),
fixture.dependencies({ openRuntime }),
),
);
// Real credential staging and tool-bridge startup can exceed Vitest's
// one-second waitFor default on a busy CI runner. Keep the admission live
// until it reaches the boundary this cancellation test is exercising.
await vi.waitFor(() => expect(openRuntime).toHaveBeenCalledOnce(), {
timeout: 5_000,
});
await runtimeAdmissionStarted.promise;
expect(receivedSignal).toBe(controller.signal);
controller.abort(cancellation);
@ -1128,34 +1255,38 @@ describe("ACPX runtime host", () => {
it("scrubs credentials after rejected runtime cleanup is proven", async () => {
const fixture = await hostFixture();
const runtimeAdmission = deferred<AcpxRuntimePort>();
const runtimeAdmissionStarted = deferred<void>();
const providerCleanup = deferred<void>();
const credentialClose = vi.fn(async () => undefined);
const controller = new AbortController();
const controller = trackedAdmissionController();
const cancellation = new Error("runtime admission cancelled");
const openRuntime = vi.fn((options: AcpxRuntimePortOpenOptions) => {
options.retainFailedAdmissionCleanup(providerCleanup.promise);
runtimeAdmissionStarted.resolve(undefined);
return runtimeAdmission.promise;
});
const opening = AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
signal: controller.signal,
},
{
...fixture.dependencies({ openRuntime }),
stageCredential: async () => ({
path: join(fixture.root, "auth.json"),
mode: "inline_json",
lifetimeFenceFds: [42, 43],
activateLifetimeOwner: async () => undefined,
close: credentialClose,
}),
},
const opening = trackAdmissionOpening(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "deny-all",
signal: controller.signal,
},
{
...fixture.dependencies({ openRuntime }),
stageCredential: async () => ({
path: join(fixture.root, "auth.json"),
mode: "inline_json",
lifetimeFenceFds: [42, 43],
activateLifetimeOwner: async () => undefined,
close: credentialClose,
}),
},
),
);
await vi.waitFor(() => expect(openRuntime).toHaveBeenCalledOnce());
await runtimeAdmissionStarted.promise;
controller.abort(cancellation);
await expect(opening).rejects.toBe(cancellation);
@ -1252,6 +1383,7 @@ async function hostFixture() {
openCommand: async () => command,
}) satisfies VerifiedAcpxInstallation,
openRuntime: input.openRuntime,
retainAdmissionCleanup: trackAdmissionCleanup,
reportRetainedCleanupFailure:
input.reportRetainedCleanupFailure ?? vi.fn(),
};
@ -1259,6 +1391,33 @@ async function hostFixture() {
};
}
function trackedAdmissionController(): AbortController {
const controller = new AbortController();
admissionControllers.push(controller);
return controller;
}
function trackAdmissionOpening<T>(opening: Promise<T>): Promise<T> {
trackSettledPromise(pendingAdmissionOpenings, opening);
return opening;
}
function trackAdmissionCleanup(cleanup: Promise<void>): void {
trackSettledPromise(pendingAdmissionCleanups, cleanup);
}
function trackSettledPromise<T>(
pending: Set<Promise<void>>,
promise: Promise<T>,
): void {
const observed = promise.then(
() => undefined,
() => undefined,
);
pending.add(observed);
void observed.finally(() => pending.delete(observed));
}
function deferred<T>(): {
promise: Promise<T>;
resolve(value: T): void;

72
packages/paperclip-runner/src/drivers/acpx/runtime-host.ts
View File

@ -140,6 +140,8 @@ export interface AcpxRuntimeHostDependencies {
/** Internal test seam for aborting credential acquisition. */
stageCredential?: typeof stageManagedCodexCredential;
openRuntime(options: AcpxRuntimePortOpenOptions): Promise<AcpxRuntimePort>;
/** Internal test seam for deterministic sandbox-admission scheduling. */
prepareSandbox?: typeof prepareAcpxRuntimeSandbox;
/** Internal test seam for the post-handshake admission deadline. */
admissionVerificationTimeoutMs?: number;
/** Internal test seam for failed-admission cleanup. */
@ -249,15 +251,18 @@ export class AcpxRuntimeHost {
);
}
const profile = resolveQualifiedAcpxProfile(options.agent, options.model);
const binding = await runAbortableAdmissionStage(options.signal, () =>
createAcpxRecoveryBinding({
runtimeDirectory: options.runtimeDirectory,
normalizedSessionId: options.normalizedSessionId,
workingDirectory: options.workingDirectory,
profile,
requestedModel: options.model,
permissionMode: options.permissionMode,
}),
const binding = await runAbortableAdmissionStage(
options.signal,
() =>
createAcpxRecoveryBinding({
runtimeDirectory: options.runtimeDirectory,
normalizedSessionId: options.normalizedSessionId,
workingDirectory: options.workingDirectory,
profile,
requestedModel: options.model,
permissionMode: options.permissionMode,
}),
dependencies.retainAdmissionCleanup,
);
if (options.expectedIdentity) {
verifyExpectedAcpxIdentity(options.expectedIdentity, binding, null);
@ -272,10 +277,13 @@ export class AcpxRuntimeHost {
);
}
const installation = await runAbortableAdmissionStage(options.signal, () =>
(dependencies.verifyInstallation ?? verifyQualifiedAcpxInstallation)(
profile,
),
const installation = await runAbortableAdmissionStage(
options.signal,
() =>
(dependencies.verifyInstallation ?? verifyQualifiedAcpxInstallation)(
profile,
),
dependencies.retainAdmissionCleanup,
);
if (installation.commandDigest !== profile.commandDigest) {
throw new Error("Verified ACPX installation does not match its profile");
@ -322,12 +330,15 @@ export class AcpxRuntimeHost {
retainRuntimeHostCleanup(ownedCleanup);
};
try {
const sandbox = await runAbortableAdmissionStage(options.signal, () =>
prepareAcpxRuntimeSandbox({
binding,
agent: options.agent,
environment: options.environment,
}),
const sandbox = await runAbortableAdmissionStage(
options.signal,
() =>
(dependencies.prepareSandbox ?? prepareAcpxRuntimeSandbox)({
binding,
agent: options.agent,
environment: options.environment,
}),
dependencies.retainAdmissionCleanup,
);
if (options.agent === "codex") {
credential = await acquireAbortableAdmissionResource({
@ -443,6 +454,7 @@ export class AcpxRuntimeHost {
profile,
admissionVerificationTimeoutMs,
),
dependencies.retainAdmissionCleanup,
);
const observedIdentity: AcpxExpectedSessionIdentity = {
kind: "acpx",
@ -629,11 +641,31 @@ export class AcpxRuntimeHost {
async function runAbortableAdmissionStage<T>(
signal: AbortSignal | undefined,
operation: () => Promise<T>,
retainCleanup: ((cleanup: Promise<void>) => void) | undefined,
): Promise<T> {
if (signal === undefined) return await operation();
signal.throwIfAborted();
const pending = Promise.resolve().then(operation);
return await raceAdmissionWithAbort(pending, signal);
try {
return await raceAdmissionWithAbort(pending, signal);
} catch (error) {
if (signal.aborted) {
// Abort may win while sandbox preparation or another non-resource stage
// still owns asynchronous work. Keep that exact operation observed and
// expose it to the embedding lifecycle so filesystem teardown cannot
// remove its session root while it is still making durable writes.
// The aborted opening is already authoritative, and this stage owns no
// provider resource. Its retained promise represents settlement only;
// a late stage rejection must not masquerade as failed resource cleanup.
const cleanup = pending.then(
() => undefined,
() => undefined,
);
retainRuntimeHostCleanup(cleanup);
retainCleanup?.(cleanup);
}
throw error;
}
}
async function acquireAbortableAdmissionResource<T>(input: {