feat(runner): add Codex ACPX sidecar (#12410)

## Thinking Path

> - Paperclip is the open source app people use to manage AI agents for
work.
> - The runner needs a bounded process boundary for each qualified
provider runtime.
> - The ACPX contract, Codex profile, structured questions, and recovery
rules now exist in the package.
> - The package does not yet provide an executable that applies those
rules to a real Codex ACPX host.
> - Provider admission must retain process, credential, and cleanup
ownership on every failure path.
> - A production selector must not depend on an unreviewed
provider-generic sidecar.
> - This pull request adds one installable Codex-only ACPX sidecar and
leaves it unselected.
> - The benefit is a testable package boundary for later runnerd
integration without changing current execution selection.

## Linked Issues or Issue Description

Refs #12409

Refs #12386

This pull request implements the Codex-only executable for the ACPX
sidecar contract merged in #12386. It builds on the question conformance
gate merged in #12409. Runnerd and the server do not select this
executable in this pull request.

## What Changed

- Publish the `paperclip-runner-acpx-sidecar` package binary and
document its current boundary.
- Add a versioned stdin/stdout sidecar that admits only the qualified
Codex ACPX profile and exact initialized model.
- Support atomic session open, run attachment, turn start and
cancellation, tool and input resolution, session read and snapshot, safe
suspension, close, and recovery identity checks.
- Bind runtime directory, workspace, permission mode, provider identity,
run identity, and semantic tool catalog before use.
- Validate completion and blocked results against the PRP result
contract. Bound pending tools, pending inputs, messages, events, usage,
diagnostics, and errors.
- Redact provider output and convert file locations to bounded
workspace-relative display data. Do not treat displayed paths as
file-access authority.
- Harden verified executable loading, module resolution, launch
environment filtering, process-group guardianship, and provider
termination.
- Retain managed credentials and every failed-admission resource until
the exact provider cleanup proves ownership was released.
- Keep failed-admission cleanup alive with bounded backoff until the
provider exits. Do not scrub credentials or admit a replacement while
cleanup still owns the provider.
- Use one runtime-host cleanup-owner registry and preserve sequential
cleanup retries across command timeouts and shutdown.
- Arm a credential-free same-group watchdog before provider admission so
guardian death reaps even a stopped provider; retain an independent
kernel EOF proof before releasing credentials.
- Add sidecar process, lifecycle, location, package, driver, credential,
installation, runtime-adapter, and runtime-host regression tests.
- Add `tsx` as a package test-only development dependency for the real
TypeScript sidecar process test.

## Verification

- Replay base: `b93ad538b63c81a1e3d24bbb54c02f8effdea787` (`master`
after #12409 merged).
- Exact replay head: `ca7e93f4385c37289c82b360ca8def0d88c1bd00`.
- Stable patch ID for the resolved 18-file delta:
`d293717a1e9f2485b61c553c58ea37690fc0a4fa`.
- The intended pull request delta contains exactly these 18 files:
  - `packages/paperclip-runner/README.md`
  - `packages/paperclip-runner/package.json`
  - `packages/paperclip-runner/src/cli/acpx-runtime-sidecar.ts`
  - `packages/paperclip-runner/src/cli/acpx-runtime-sidecar.test.ts`
  - `packages/paperclip-runner/src/cli/acpx-sidecar-lifecycle.ts`
  - `packages/paperclip-runner/src/cli/acpx-sidecar-locations.ts`
  - `packages/paperclip-runner/src/cli/acpx-sidecar-locations.test.ts`
  - `packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.ts`
- `packages/paperclip-runner/src/drivers/acpx/codex-acpx-driver.test.ts`
  - `packages/paperclip-runner/src/drivers/acpx/codex-credentials.ts`
- `packages/paperclip-runner/src/drivers/acpx/codex-credentials.test.ts`
- `packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts`
-
`packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts`
- `packages/paperclip-runner/src/drivers/acpx/installation-integrity.ts`
-
`packages/paperclip-runner/src/drivers/acpx/installation-integrity.test.ts`
  - `packages/paperclip-runner/src/drivers/acpx/runtime-host.ts`
  - `packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts`
- `packages/paperclip-runner/test/acpx-codex-package-contract.test.mjs`
- The resolved combined delta is 4,725 additions and 665 deletions; it
preserves the lower-PR runtime-close semantics, repairs stale
successful-admission fixtures, deterministically observes renewed
reconciliation, accepts authoritative same-host cleanup recovery without
dropping pending owners, ignores superseded cleanup failures after a
newer owner recovers, and requires both guardian exit and independent
provider-lifetime EOF before releasing cleanup or credential ownership.
A readiness-gated, credential-free watchdog also reaps a stopped
provider if its guardian is externally killed.
- This change updates the runner package manifest, README, and test-only
dependencies. It does not change `pnpm-lock.yaml`, a workflow,
migration, server route, UI path, runnerd selection, or current
direct-adapter behavior.
- Focused GitHub verification: **PASSED** for the sidecar process,
lifecycle, location, package-contract, driver, credential,
installation-integrity, runtime-adapter, and runtime-host suites on the
replayed head.
- Package verification: **PASSED** for the clean tarball, Node shebang,
and exact binary mapping on the replayed head.
- GitHub Actions and security checks: **PASSED** for the replayed exact
head; full CI run `33357846557` completed 23/23 jobs successfully, and
Superagent, Socket, Snyk, supply-chain, and contributor-trust checks are
green. Storybook was intentionally skipped because this PR does not
touch its paths.
- Greptile: **5/5** on the exact head with zero unresolved review
threads.
- No local test result is claimed. GitHub Actions is the authoritative
verification environment for the replayed revision.

## Risks

This change has medium security and lifecycle risk because the new
executable crosses a process, credential, filesystem, and provider
boundary. The sidecar fails closed on unsupported providers, models,
permissions, identities, catalogs, forms, commands, and persistent-state
deletion. A cleanup owner can remain alive until a stubborn provider
exits. Its retries use bounded backoff, and admission stays closed while
ownership remains. Command and shutdown waits remain bounded without
abandoning the underlying cleanup. The new `tsx` dependency is
development-only. The package exposes a new binary, but no runnerd,
server, UI, or direct-adapter path starts it in this pull request.

> For core feature work, check [`ROADMAP.md`](ROADMAP.md) first and
discuss it in `#dev` before opening the PR. Feature PRs that overlap
with planned core work may need to be redirected — check the roadmap
first. See `CONTRIBUTING.md`.

## Model Used

OpenAI Codex with GPT-5.6, extended reasoning, repository tool use, and
code execution.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [ ] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge
This commit is contained in:
Dotta 2026-08-30 23:49:01 -05:00 committed by GitHub
parent b93ad538b6
commit 9ad8dbffa0
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
18 changed files with 4732 additions and 672 deletions

View File

@ -49,6 +49,12 @@ an installed server does not depend on a separate system Rust installation or
a manually copied binary. `pnpm-lock.yaml` remains under the repository's
existing lockfile process.
The package also builds `paperclip-runner-acpx-sidecar`. This bounded v2
stdin/stdout bridge admits the qualified Codex ACPX profile only. It validates
the exact model, session identity, tool catalog, structured input, and terminal
settlement at the process boundary. Runnerd and the server do not select this
sidecar in this slice. Other ACPX agents remain unavailable.
Run the complete contract gate with:
```sh

View File

@ -7,6 +7,9 @@
"node": ">=24.11.0"
},
"type": "module",
"bin": {
"paperclip-runner-acpx-sidecar": "./dist/cli/acpx-runtime-sidecar.js"
},
"exports": {
".": {
"types": "./dist/index.d.ts",
@ -59,6 +62,7 @@
},
"devDependencies": {
"@types/node": "^24.0.0",
"tsx": "^4.23.12",
"typescript": "^7.0.2",
"vitest": "^4.1.10"
}

View File

@ -0,0 +1,563 @@
import { spawn, type ChildProcessWithoutNullStreams } from "node:child_process";
import { fileURLToPath } from "node:url";
import { afterEach, describe, expect, it, vi } from "vitest";
import { ACPX_SIDECAR_PROTOCOL_VERSION } from "../drivers/acpx/sidecar-protocol.js";
import {
awaitSidecarCleanupWithin,
closeActiveSidecarHostWithin,
closeSidecarHostForCommand,
combineSidecarAdmissionCleanups,
combineSidecarHostCleanups,
hasSidecarSessionOwnership,
observeSidecarCleanupWithin,
parseAcpxRunAttachment,
readSidecarHostStatusWithin,
recoverAndCombineSidecarHostCleanup,
recoverSidecarHostCleanup,
reportAuthoritativeSidecarHostCleanupFailure,
requireSidecarCommandHost,
verifyOpenedAcpxSidecarHost,
} from "./acpx-sidecar-lifecycle.js";
const children = new Set<SidecarProcess>();
afterEach(async () => {
await Promise.all([...children].map((child) => child.close()));
children.clear();
});
describe("Codex ACPX runtime sidecar", () => {
it("keeps session admission closed while any cleanup owner remains", () => {
const cleanup = Promise.resolve();
expect(hasSidecarSessionOwnership(null, null, null)).toBe(false);
expect(hasSidecarSessionOwnership({}, null, null)).toBe(true);
expect(hasSidecarSessionOwnership(null, cleanup, null)).toBe(true);
expect(hasSidecarSessionOwnership(null, null, cleanup)).toBe(true);
});
it("allows only an explicit cleanup retry to reach a retained host", () => {
const host = { identity: () => ({ kind: "acpx" }) };
const cleanup = new Promise<void>(() => undefined);
expect(() => requireSidecarCommandHost(host, cleanup)).toThrow(
"cleanup is in progress",
);
expect(
requireSidecarCommandHost(host, cleanup, { allowCleanupRetry: true }),
).toBe(host);
expect(() =>
requireSidecarCommandHost(null, cleanup, { allowCleanupRetry: true }),
).toThrow("session is not open");
});
it("closes an opened host when post-open verification fails", async () => {
const close = vi.fn().mockResolvedValue(undefined);
const host = {
identity: () => ({ kind: "acpx" }),
status: vi.fn().mockRejectedValue(new Error("status failed")),
close,
};
await expect(verifyOpenedAcpxSidecarHost(host, () => ({}))).rejects.toThrow(
"status failed",
);
expect(close).toHaveBeenCalledOnce();
expect(close).toHaveBeenCalledWith({
reason: "ACPX session open verification failed",
});
});
it("bounds failed-admission cleanup when the host does not settle", async () => {
let finishCleanup!: () => void;
const cleanup = new Promise<void>((resolve) => {
finishCleanup = resolve;
});
const close = vi.fn(() => cleanup);
const retainCleanup = vi.fn();
const host = {
identity: () => ({ kind: "acpx" }),
status: vi.fn().mockRejectedValue(new Error("status failed")),
close,
};
await expect(
verifyOpenedAcpxSidecarHost(host, () => ({}), 1, retainCleanup),
).rejects.toThrow("verification and provider cleanup failed");
expect(close).toHaveBeenCalledOnce();
expect(retainCleanup).toHaveBeenCalledWith(cleanup);
finishCleanup();
await cleanup;
});
it("bounds shutdown waiting without releasing retained cleanup ownership", async () => {
let finishCleanup!: () => void;
const cleanup = new Promise<void>((resolve) => {
finishCleanup = resolve;
});
await expect(awaitSidecarCleanupWithin(cleanup, 1)).resolves.toBe(
"deferred",
);
let settled = false;
void cleanup.then(() => {
settled = true;
});
expect(settled).toBe(false);
finishCleanup();
await cleanup;
expect(settled).toBe(true);
await expect(awaitSidecarCleanupWithin(cleanup, 1)).resolves.toBe(
"settled",
);
});
it("preserves retained cleanup failure for shutdown accounting", async () => {
const failure = new Error("provider cleanup failed");
await expect(
observeSidecarCleanupWithin(Promise.reject(failure), 1),
).resolves.toEqual({ status: "failed", error: failure });
await expect(
observeSidecarCleanupWithin(new Promise<void>(() => undefined), 1),
).resolves.toEqual({ status: "deferred" });
});
it("preserves failed-admission rejection until every cleanup settles", async () => {
let finishCleanup!: () => void;
const pending = new Promise<void>((resolve) => {
finishCleanup = resolve;
});
const retained = combineSidecarAdmissionCleanups([
Promise.reject(new Error("provider survived termination")),
pending,
]);
let settled = false;
void retained.then(
() => {
settled = true;
},
() => {
settled = true;
},
);
await Promise.resolve();
expect(settled).toBe(false);
finishCleanup();
await expect(retained).rejects.toThrow(
"did not release provider ownership",
);
});
it("bounds active-host cleanup during sidecar shutdown", async () => {
const cleanup = new Promise<void>(() => undefined);
const close = vi.fn(() => cleanup);
const retainCleanup = vi.fn();
await expect(
closeActiveSidecarHostWithin({ close }, "SIGTERM", 1, retainCleanup),
).resolves.toBe("deferred");
expect(close).toHaveBeenCalledWith({ reason: "SIGTERM" });
expect(retainCleanup).toHaveBeenCalledWith(cleanup);
});
it("bounds command cleanup without replacing its exact owner", async () => {
let finishCleanup!: () => void;
const cleanup = new Promise<void>((resolve) => {
finishCleanup = resolve;
});
const close = vi.fn(() => cleanup);
const retainCleanup = vi.fn();
await expect(
closeSidecarHostForCommand({ close }, "session close", 1, retainCleanup),
).rejects.toThrow("cleanup exceeded its command timeout");
expect(close).toHaveBeenCalledOnce();
expect(retainCleanup).toHaveBeenCalledWith(cleanup);
finishCleanup();
await cleanup;
});
it("preserves a settled command cleanup failure", async () => {
const cleanup = Promise.reject(new Error("runtime close failed"));
await expect(
closeSidecarHostForCommand({ close: () => cleanup }, "session close", 10),
).rejects.toThrow("runtime close failed");
});
it("recovers a rejected active-host cleanup sequentially", async () => {
const close = vi
.fn<() => Promise<void>>()
.mockRejectedValueOnce(new Error("first close failed"))
.mockResolvedValue(undefined);
const host = { close };
const initialCleanup = host.close();
await expect(
recoverSidecarHostCleanup(host, initialCleanup),
).resolves.toBeUndefined();
expect(close).toHaveBeenCalledTimes(2);
});
it("bounds repeated active-host cleanup failures", async () => {
const close = vi
.fn<() => Promise<void>>()
.mockRejectedValue(new Error("close failed"));
const host = { close };
const initialCleanup = host.close();
await expect(
recoverSidecarHostCleanup(host, initialCleanup),
).rejects.toThrow("close failed");
expect(close).toHaveBeenCalledTimes(4);
});
it("retains a pending cleanup owner after a command retry succeeds", async () => {
let finishPending!: () => void;
const pending = new Promise<void>((resolve) => {
finishPending = resolve;
});
const successfulRetry = Promise.resolve();
const owner = combineSidecarHostCleanups([pending, successfulRetry]);
let settled = false;
void owner.then(() => {
settled = true;
});
await Promise.resolve();
expect(settled).toBe(false);
finishPending();
await expect(owner).resolves.toBeUndefined();
});
it("accepts a coalesced rejection after sequential recovery succeeds", async () => {
let rejectCoalesced!: (error: unknown) => void;
const coalesced = new Promise<void>((_resolve, reject) => {
rejectCoalesced = reject;
});
const close = vi
.fn<() => Promise<void>>()
.mockReturnValueOnce(coalesced)
.mockReturnValueOnce(coalesced)
.mockResolvedValue(undefined);
const host = { close };
const recoveredPrior = recoverSidecarHostCleanup(host, host.close());
const owner = recoverAndCombineSidecarHostCleanup(
host,
host.close(),
recoveredPrior,
);
let settled = false;
void owner
.finally(() => {
settled = true;
})
.catch(() => undefined);
await Promise.resolve();
expect(settled).toBe(false);
rejectCoalesced(new Error("coalesced close failed before recovery"));
await expect(owner).resolves.toBeUndefined();
expect(close).toHaveBeenCalledTimes(4);
});
it("rejects when every active-host cleanup owner fails", async () => {
const owner = combineSidecarHostCleanups([
Promise.reject(new Error("recovery exhausted")),
Promise.reject(new Error("retry failed")),
]);
await expect(owner).rejects.toThrow("did not release provider ownership");
});
it("accepts a later recovery after an older owner exhausts", async () => {
await expect(
combineSidecarHostCleanups([
Promise.reject(new Error("older recovery exhausted")),
Promise.resolve(),
]),
).resolves.toBeUndefined();
});
it("does not escalate a superseded cleanup owner failure", async () => {
let rejectOlder!: (error: unknown) => void;
const older = new Promise<void>((_resolve, reject) => {
rejectOlder = reject;
});
const replacement = combineSidecarHostCleanups([
older,
Promise.resolve(),
]);
const reportFailure = vi.fn();
void older.catch((error: unknown) => {
reportAuthoritativeSidecarHostCleanupFailure(
false,
replacement,
older,
error,
reportFailure,
);
});
rejectOlder(new Error("older recovery exhausted"));
await expect(replacement).resolves.toBeUndefined();
expect(reportFailure).not.toHaveBeenCalled();
});
it("escalates only an authoritative cleanup owner's terminal failure", async () => {
const owner = combineSidecarHostCleanups([
Promise.reject(new Error("older recovery exhausted")),
Promise.reject(new Error("replacement recovery exhausted")),
]);
const reportFailure = vi.fn();
await owner.catch((error: unknown) => {
reportAuthoritativeSidecarHostCleanupFailure(
false,
owner,
owner,
error,
reportFailure,
);
});
expect(reportFailure).toHaveBeenCalledOnce();
expect(reportFailure.mock.calls[0]?.[0]).toBeInstanceOf(AggregateError);
reportAuthoritativeSidecarHostCleanupFailure(
true,
owner,
owner,
new Error("shutdown cleanup failed"),
reportFailure,
);
expect(reportFailure).toHaveBeenCalledOnce();
});
it("bounds status verification before cleaning up the opened host", async () => {
const close = vi.fn().mockResolvedValue(undefined);
const host = {
identity: () => ({ kind: "acpx" }),
status: vi.fn(() => new Promise<never>(() => undefined)),
close,
};
await expect(
verifyOpenedAcpxSidecarHost(host, () => ({}), 1),
).rejects.toThrow("status read exceeded its timeout");
expect(close).toHaveBeenCalledOnce();
});
it("bounds ordinary status reads so serialized shutdown can proceed", async () => {
const host = {
status: vi.fn(() => new Promise<never>(() => undefined)),
};
await expect(readSidecarHostStatusWithin(host, 1)).rejects.toThrow(
"status read exceeded its timeout",
);
});
it("validates a complete run attachment before it can be committed", () => {
let attachedRunId: string | null = null;
const attach = (params: Record<string, unknown>) => {
const attachment = parseAcpxRunAttachment(params);
attachedRunId = attachment.runId;
return attachment;
};
expect(() => attach({ runId: "run-1", catalogRevision: 0 })).toThrow(
"catalogRevision must be a positive integer",
);
expect(attachedRunId).toBeNull();
expect(attach({ runId: "run-1", catalogRevision: 2 })).toEqual({
runId: "run-1",
catalogRevision: 2,
});
expect(attachedRunId).toBe("run-1");
});
it("recovers after malformed input and reports its qualified Codex profile", async () => {
const sidecar = startSidecar();
sidecar.write({
protocolVersion: ACPX_SIDECAR_PROTOCOL_VERSION,
id: 1,
command: "initialize",
params: {},
unexpected: true,
});
await expect(
sidecar.next((frame) => frame.eventType === "runtime.diagnostic"),
).resolves.toMatchObject({
protocolVersion: ACPX_SIDECAR_PROTOCOL_VERSION,
eventType: "runtime.diagnostic",
payload: { code: "malformed_frame" },
});
sidecar.write(initializeRequest(2, "codex"));
await expect(
sidecar.next((frame) => frame.id === 2),
).resolves.toMatchObject({
protocolVersion: ACPX_SIDECAR_PROTOCOL_VERSION,
id: 2,
ok: true,
result: {
profile: {
agent: "codex",
qualificationModel: "gpt-5.6-sol",
},
capabilities: {
persistentSessions: true,
exactModelVerification: true,
structuredInput: "paperclip.question_set.v1",
},
},
});
expect(sidecar.stderr()).toContain("malformed_frame");
sidecar.write(initializeRequest(3, "codex"));
await expect(
sidecar.next((frame) => frame.id === 3),
).resolves.toMatchObject({
id: 3,
ok: false,
error: { message: "ACPX sidecar is already initialized" },
});
});
it("fails closed after an unsupported provider bootstrap", async () => {
const sidecar = startSidecar();
sidecar.write(initializeRequest(1, "pi"));
await expect(
sidecar.next((frame) => frame.id === 1),
).resolves.toMatchObject({
id: 1,
ok: false,
error: {
code: "acpx_sidecar_command_failed",
message: "This production ACPX sidecar supports Codex only",
retryable: false,
},
});
sidecar.write(initializeRequest(2, "codex"));
await expect(
sidecar.next((frame) => frame.id === 2),
).resolves.toMatchObject({
id: 2,
ok: false,
error: {
message: expect.stringContaining(
"ACPX provider bootstrap failed before initialize",
),
retryable: false,
},
});
});
});
function initializeRequest(id: number, agent: string): Record<string, unknown> {
return {
protocolVersion: ACPX_SIDECAR_PROTOCOL_VERSION,
id,
command: "initialize",
params: { agent, model: "gpt-5.6-sol" },
};
}
function startSidecar(): SidecarProcess {
const sidecar = new SidecarProcess();
children.add(sidecar);
return sidecar;
}
class SidecarProcess {
readonly #child: ChildProcessWithoutNullStreams;
readonly #frames: Array<Record<string, unknown>> = [];
readonly #signals: Array<() => void> = [];
#stderr = "";
#closed = false;
constructor() {
this.#child = spawn(
fileURLToPath(new URL("../../node_modules/.bin/tsx", import.meta.url)),
[fileURLToPath(new URL("./acpx-runtime-sidecar.ts", import.meta.url))],
{ stdio: ["pipe", "pipe", "pipe"] },
);
let stdout = "";
this.#child.stdout.setEncoding("utf8");
this.#child.stdout.on("data", (chunk: string) => {
stdout += chunk;
for (;;) {
const newline = stdout.indexOf("\n");
if (newline < 0) break;
const line = stdout.slice(0, newline);
stdout = stdout.slice(newline + 1);
if (!line.trim()) continue;
this.#frames.push(JSON.parse(line) as Record<string, unknown>);
for (const signal of this.#signals.splice(0)) signal();
}
});
this.#child.stderr.setEncoding("utf8");
this.#child.stderr.on("data", (chunk: string) => {
this.#stderr += chunk;
});
}
write(value: Record<string, unknown>): void {
this.#child.stdin.write(`${JSON.stringify(value)}\n`);
}
stderr(): string {
return this.#stderr;
}
async next(
predicate: (frame: Record<string, unknown>) => boolean,
): Promise<Record<string, unknown>> {
const deadline = Date.now() + 5_000;
for (;;) {
const index = this.#frames.findIndex(predicate);
if (index >= 0) return this.#frames.splice(index, 1)[0]!;
const remaining = deadline - Date.now();
if (remaining <= 0) {
throw new Error(
`Timed out waiting for sidecar frame. stderr=${JSON.stringify(this.#stderr)}`,
);
}
await new Promise<void>((resolve, reject) => {
const timer = setTimeout(() => {
const index = this.#signals.indexOf(signal);
if (index >= 0) this.#signals.splice(index, 1);
reject(new Error("Timed out waiting for sidecar output"));
}, remaining);
const signal = () => {
clearTimeout(timer);
resolve();
};
this.#signals.push(signal);
});
}
}
async close(): Promise<void> {
if (this.#closed) return;
this.#closed = true;
this.#child.stdin.end();
const exit = new Promise<void>((resolve) => {
this.#child.once("exit", () => resolve());
});
const timeout = new Promise<void>((resolve) => {
setTimeout(() => {
if (this.#child.exitCode === null) this.#child.kill("SIGKILL");
resolve();
}, 2_000).unref();
});
await Promise.race([exit, timeout]);
}
}

File diff suppressed because it is too large Load Diff

View File

@ -0,0 +1,286 @@
export interface OpenedAcpxSidecarHost {
identity(): unknown;
status(): Promise<unknown>;
close(options: { reason: string }): Promise<void>;
}
const FAILED_ADMISSION_CLOSE_TIMEOUT_MS = 8_000;
const ACTIVE_HOST_CLEANUP_ATTEMPTS = 4;
export function hasSidecarSessionOwnership(
host: unknown,
activeHostCleanup: Promise<void> | null,
failedAdmissionCleanup: Promise<void> | null,
): boolean {
return Boolean(host || activeHostCleanup || failedAdmissionCleanup);
}
/**
* Ordinary commands cannot observe a host while cleanup owns it. An explicit
* cleanup retry may reuse that same host so its close can supersede a stale
* owner; admission remains guarded separately by hasSidecarSessionOwnership.
*/
export function requireSidecarCommandHost<T>(
host: T | null,
activeHostCleanup: Promise<void> | null,
options: { allowCleanupRetry?: boolean } = {},
): T {
if (!host) throw new Error("ACPX session is not open");
if (activeHostCleanup && options.allowCleanupRetry !== true) {
throw new Error("ACPX session cleanup is in progress");
}
return host;
}
export async function readSidecarHostStatusWithin(
host: Pick<OpenedAcpxSidecarHost, "status">,
timeoutMs = FAILED_ADMISSION_CLOSE_TIMEOUT_MS,
): Promise<unknown> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
host.status(),
new Promise<never>((_resolve, reject) => {
timer = setTimeout(
() =>
reject(new Error("ACPX session status read exceeded its timeout")),
timeoutMs,
);
timer.unref();
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
export async function awaitSidecarCleanupWithin(
cleanup: Promise<void>,
timeoutMs = FAILED_ADMISSION_CLOSE_TIMEOUT_MS,
): Promise<"settled" | "deferred"> {
const outcome = await observeSidecarCleanupWithin(cleanup, timeoutMs);
return outcome.status === "deferred" ? "deferred" : "settled";
}
export type SidecarCleanupOutcome =
| { status: "settled" }
| { status: "deferred" }
| { status: "failed"; error: unknown };
export async function observeSidecarCleanupWithin(
cleanup: Promise<void>,
timeoutMs = FAILED_ADMISSION_CLOSE_TIMEOUT_MS,
): Promise<SidecarCleanupOutcome> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
cleanup.then(
() => ({ status: "settled" as const }),
(error: unknown) => ({ status: "failed" as const, error }),
),
new Promise<SidecarCleanupOutcome>((resolve) => {
timer = setTimeout(() => resolve({ status: "deferred" }), timeoutMs);
timer.unref();
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
export async function combineSidecarAdmissionCleanups(
cleanups: readonly Promise<void>[],
): Promise<void> {
const outcomes = await Promise.allSettled(cleanups);
const errors = outcomes.flatMap((outcome) =>
outcome.status === "rejected" ? [outcome.reason as unknown] : [],
);
if (errors.length > 0) {
throw new AggregateError(
errors,
"ACPX failed-admission cleanup did not release provider ownership",
);
}
}
/**
* Keep ownership until every cleanup started for the same host settles. Once
* all observers are terminal, one successful close proves that host released
* its provider resources; an intermediate coalesced rejection must not erase
* that proof. Reject only when every cleanup owner failed.
*/
export async function combineSidecarHostCleanups(
cleanups: readonly [Promise<void>, Promise<void>],
): Promise<void> {
const outcomes = await Promise.allSettled(cleanups);
if (outcomes.some((outcome) => outcome.status === "fulfilled")) return;
const errors = outcomes.flatMap((outcome) =>
outcome.status === "rejected" ? [outcome.reason as unknown] : [],
);
throw new AggregateError(
errors,
"ACPX active-host cleanup did not release provider ownership",
);
}
export function recoverAndCombineSidecarHostCleanup(
host: Pick<OpenedAcpxSidecarHost, "close">,
cleanup: Promise<void>,
prior: Promise<void> | null,
): Promise<void> {
const recovered = recoverSidecarHostCleanup(host, cleanup);
return prior
? combineSidecarHostCleanups([prior, recovered])
: recovered;
}
export function reportAuthoritativeSidecarHostCleanupFailure(
closing: boolean,
activeCleanup: Promise<void> | null,
failedCleanup: Promise<void>,
error: unknown,
reportFailure: (error: unknown) => void,
): void {
if (!closing && activeCleanup === failedCleanup) reportFailure(error);
}
export async function closeActiveSidecarHostWithin(
host: Pick<OpenedAcpxSidecarHost, "close">,
reason: string,
timeoutMs = FAILED_ADMISSION_CLOSE_TIMEOUT_MS,
retainCleanup: (cleanup: Promise<void>) => void = () => undefined,
): Promise<"settled" | "deferred"> {
const cleanup = host.close({ reason });
retainCleanup(cleanup);
return await awaitSidecarCleanupWithin(cleanup, timeoutMs);
}
export async function closeSidecarHostForCommand(
host: Pick<OpenedAcpxSidecarHost, "close">,
reason: string,
timeoutMs = FAILED_ADMISSION_CLOSE_TIMEOUT_MS,
retainCleanup: (cleanup: Promise<void>) => void = () => undefined,
): Promise<void> {
const cleanup = host.close({ reason });
retainCleanup(cleanup);
const disposition = await awaitSidecarCleanupWithin(cleanup, timeoutMs);
if (disposition === "deferred") {
throw new Error("ACPX session cleanup exceeded its command timeout");
}
// The bounded wait only reports settlement; preserve the exact close error
// for the command response and keep the host available for a later retry.
await cleanup;
}
export async function recoverSidecarHostCleanup(
host: Pick<OpenedAcpxSidecarHost, "close">,
initialCleanup: Promise<void>,
maxAttempts = ACTIVE_HOST_CLEANUP_ATTEMPTS,
): Promise<void> {
let cleanup = initialCleanup;
for (let attempt = 1; ; attempt += 1) {
try {
await cleanup;
return;
} catch (error) {
if (attempt >= maxAttempts) throw error;
// AcpxRuntimeHost releases its failed close promise before propagating
// the rejection, so this starts a new sequential cleanup attempt rather
// than reusing or overlapping the rejected operation.
cleanup = host.close({ reason: "Paperclip cleanup recovery" });
}
}
}
export async function verifyOpenedAcpxSidecarHost(
host: OpenedAcpxSidecarHost,
sanitizeStatus: (value: unknown) => Record<string, unknown>,
closeTimeoutMs = FAILED_ADMISSION_CLOSE_TIMEOUT_MS,
retainCleanup: (cleanup: Promise<void>) => void = () => undefined,
): Promise<{ identity: unknown; status: Record<string, unknown> }> {
try {
const identity = host.identity();
const status = sanitizeStatus(
await readSidecarHostStatusWithin(host, closeTimeoutMs),
);
return { identity, status };
} catch (error) {
const cleanup = host.close({
reason: "ACPX session open verification failed",
});
// The admission timeout bounds the command response, not ownership. The
// sidecar retains this exact close operation so shutdown can still await
// provider termination after the bounded verification path returns.
retainCleanup(cleanup);
const cleanupError = await boundedFailedAdmissionClose(
cleanup,
closeTimeoutMs,
);
if (cleanupError) {
throw new AggregateError(
[error, cleanupError],
"ACPX session verification and provider cleanup failed",
);
}
throw error;
}
}
async function boundedFailedAdmissionClose(
close: Promise<void>,
timeoutMs: number,
): Promise<unknown | null> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
close.then(
() => null,
(error: unknown) => error,
),
new Promise<Error>((resolve) => {
timer = setTimeout(
() =>
resolve(
new Error(
"ACPX failed-admission cleanup exceeded its shutdown timeout",
),
),
timeoutMs,
);
timer.unref();
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
export interface AcpxRunAttachment {
runId: string;
catalogRevision: number;
}
export function parseAcpxRunAttachment(
params: Record<string, unknown>,
): AcpxRunAttachment {
return {
runId: boundedIdentity(params.runId, "runId"),
catalogRevision: positiveInteger(params.catalogRevision, "catalogRevision"),
};
}
export function boundedIdentity(value: unknown, field: string): string {
const result = typeof value === "string" ? value.trim() : "";
if (!result) throw new Error(`${field} is required`);
if (result.length > 240 || /[\u0000-\u001f\u007f]/.test(result)) {
throw new Error(`${field} is invalid`);
}
return result;
}
function positiveInteger(value: unknown, field: string): number {
if (!Number.isSafeInteger(value) || Number(value) < 1) {
throw new Error(`${field} must be a positive integer`);
}
return Number(value);
}

View File

@ -0,0 +1,59 @@
import { describe, expect, it } from "vitest";
import { safeAcpxLocations } from "./acpx-sidecar-locations.js";
describe("ACPX sidecar locations", () => {
it("preserves valid host-relative display names without admitting escape", () => {
expect(
safeAcpxLocations(
[
{ path: "src/main.ts", line: 4 },
{ path: "reports/100%/summary.txt" },
{ path: "../outside.txt" },
{ path: "/etc/passwd" },
{ uri: "https://example.test/private" },
{ path: "bad\0name" },
],
"/workspace/project",
),
).toEqual([
{
path: "src/main.ts",
line: 4,
pathBoundary: "paperclip.workspace_relative_display.v1",
},
{
path: "reports/100%/summary.txt",
line: null,
pathBoundary: "paperclip.workspace_relative_display.v1",
},
]);
});
it.runIf(process.platform !== "win32")(
"preserves POSIX literal colon and backslash filename characters",
() => {
expect(
safeAcpxLocations(
[{ path: "src:main.ts" }, { path: String.raw`folder\literal` }],
"/workspace/project",
),
).toEqual([
{
path: "src:main.ts",
line: null,
pathBoundary: "paperclip.workspace_relative_display.v1",
},
{
path: String.raw`folder\literal`,
line: null,
pathBoundary: "paperclip.workspace_relative_display.v1",
},
]);
},
);
it("omits every location until the session working directory is bound", () => {
expect(safeAcpxLocations([{ path: "src/main.ts" }], undefined)).toEqual([]);
});
});

View File

@ -0,0 +1,49 @@
import { isAbsolute, relative, resolve, sep } from "node:path";
export const ACPX_WORKSPACE_RELATIVE_DISPLAY_BOUNDARY =
"paperclip.workspace_relative_display.v1";
function record(value: unknown): Record<string, unknown> {
return value !== null && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: {};
}
/**
* Converts provider paths to workspace-relative display targets using the
* sidecar host's path semantics. A URI is not a path. Windows separators are
* canonicalized for PRP; POSIX backslashes and colons remain literal filename
* characters. Consumers must treat the result as display data, never as an
* authorization to access a file.
*/
export function safeAcpxLocations(
locations: readonly unknown[] | null | undefined,
workingDirectory: string | null | undefined,
): Array<Record<string, unknown>> {
if (!workingDirectory) return [];
const cwd = resolve(workingDirectory);
return (locations ?? []).slice(0, 2_000).flatMap((location) => {
const candidate = record(location);
const rawPath = typeof candidate.path === "string" ? candidate.path : "";
if (!rawPath || rawPath.includes("\0")) return [];
const absolute = isAbsolute(rawPath)
? resolve(rawPath)
: resolve(cwd, rawPath);
const local = relative(cwd, absolute);
if (!local || isAbsolute(local)) return [];
const portable = sep === "\\" ? local.replaceAll("\\", "/") : local;
if (
portable.startsWith("/") ||
portable.split("/").some((segment) => segment === "..")
) {
return [];
}
return [
{
path: [...portable].slice(0, 4_000).join(""),
line: candidate.line ?? null,
pathBoundary: ACPX_WORKSPACE_RELATIVE_DISPLAY_BOUNDARY,
},
];
});
}

View File

@ -511,9 +511,9 @@ describe("Codex ACPX harness driver", () => {
fixture.host.close.mockImplementation(({ reason }) =>
reason.includes("scheduled quarantined cleanup recovery")
? // Exercise the complete production host bound: two seconds for
// active-turn cancellation plus six seconds for protocol/TERM/KILL.
new Promise<void>((resolve) => setTimeout(resolve, 8_500))
? // Exercise the complete production host bound: active-turn
// cancellation plus bounded protocol, TERM, and guardian-group KILL.
new Promise<void>((resolve) => setTimeout(resolve, 9_500))
: Promise.resolve(),
);
await vi.advanceTimersToNextTimerAsync();
@ -536,7 +536,7 @@ describe("Codex ACPX harness driver", () => {
reason:
"runtime close persistently failed (scheduled quarantined cleanup recovery)",
});
await vi.advanceTimersByTimeAsync(8_499);
await vi.advanceTimersByTimeAsync(9_499);
expect(admissionSettled).toBe(false);
await vi.advanceTimersByTimeAsync(1);
await expect(admission).resolves.toBeDefined();
@ -638,7 +638,7 @@ describe("Codex ACPX harness driver", () => {
setTimeout(() => {
if (attempt < 3) reject(new Error("transient quarantine failure"));
else resolve();
}, 8_000);
}, 9_000);
});
});
const session = await fixture.driver.openSession({
@ -669,7 +669,7 @@ describe("Codex ACPX harness driver", () => {
admissionSettled = true;
},
);
await vi.advanceTimersByTimeAsync(23_000);
await vi.advanceTimersByTimeAsync(26_000);
expect(admissionSettled).toBe(false);
expect(fixture.host.close).toHaveBeenCalledTimes(7);
await vi.advanceTimersByTimeAsync(2_000);
@ -976,7 +976,7 @@ describe("Codex ACPX harness driver", () => {
admissionSettled = true;
})
.catch(() => undefined);
await vi.advanceTimersByTimeAsync(34_999);
await vi.advanceTimersByTimeAsync(38_999);
expect(admissionSettled).toBe(false);
await vi.advanceTimersByTimeAsync(2);
await expect(admission).rejects.toThrow("exceeded the admission grace");
@ -1702,9 +1702,11 @@ describe("Codex ACPX harness driver", () => {
const cancellation = new Error("recovery cancelled before start");
controller.abort(cancellation);
await expect(fixture.driver.recoverSession!(snapshot, {
signal: controller.signal,
})).resolves.toEqual({
await expect(
fixture.driver.recoverSession!(snapshot, {
signal: controller.signal,
}),
).resolves.toEqual({
recovered: false,
reason: cancellation.message,
});
@ -1850,7 +1852,9 @@ describe("Codex ACPX harness driver", () => {
semanticResult: { turnId: semanticTurn.turnId },
});
expect(snapshot.terminalTurns?.at(-1)?.turnId).toBe(laterTurn.turnId);
await session.close({ reason: "simulate unsuccessful follow-up recovery" });
await session.close({
reason: "simulate unsuccessful follow-up recovery",
});
await expect(fixture.driver.recoverSession!(snapshot)).resolves.toEqual({
recovered: false,
@ -1882,7 +1886,11 @@ describe("Codex ACPX harness driver", () => {
});
fixture.finishTurn({
status: "failed",
error: { code: "provider_retry", message: "Retry the turn", retryable: true },
error: {
code: "provider_retry",
message: "Retry the turn",
retryable: true,
},
});
await firstTerminal;
@ -1906,10 +1914,12 @@ describe("Codex ACPX harness driver", () => {
});
expect(snapshot.semanticResult?.turnId).not.toBe(first.turnId);
expect(snapshot.terminalTurns?.at(-1)?.turnId).toBe(second.turnId);
expect(snapshot.terminalTurns).toEqual(expect.arrayContaining([
expect.objectContaining({ turnId: first.turnId }),
expect.objectContaining({ turnId: second.turnId }),
]));
expect(snapshot.terminalTurns).toEqual(
expect.arrayContaining([
expect.objectContaining({ turnId: first.turnId }),
expect.objectContaining({ turnId: second.turnId }),
]),
);
const successfulTerminal = snapshot.terminalTurns?.find(
(terminal) => terminal.turnId === second.turnId,
);
@ -1919,15 +1929,19 @@ describe("Codex ACPX harness driver", () => {
});
await session.close({ reason: "simulate successful retry recovery" });
await expect(fixture.driver.recoverSession!({
...snapshot,
activeTurnId: first.turnId,
})).resolves.toEqual({
await expect(
fixture.driver.recoverSession!({
...snapshot,
activeTurnId: first.turnId,
}),
).resolves.toEqual({
recovered: false,
reason:
"persisted Codex ACPX active turn is not the completed semantic settlement",
});
await expect(fixture.driver.recoverSession!(snapshot)).resolves.toMatchObject({
await expect(
fixture.driver.recoverSession!(snapshot),
).resolves.toMatchObject({
recovered: true,
});
});
@ -1984,7 +1998,9 @@ describe("Codex ACPX harness driver", () => {
});
await session.close({ reason: "simulate reaffirmed result recovery" });
await expect(fixture.driver.recoverSession!(snapshot)).resolves.toMatchObject({
await expect(
fixture.driver.recoverSession!(snapshot),
).resolves.toMatchObject({
recovered: true,
});
});
@ -2031,9 +2047,7 @@ describe("Codex ACPX harness driver", () => {
});
const failedEvents = await failedTerminal;
expect(
failedEvents.filter(
(event) => event.eventType === "run.result.proposed",
),
failedEvents.filter((event) => event.eventType === "run.result.proposed"),
).toHaveLength(1);
const snapshot = await session.snapshot();
@ -2296,10 +2310,12 @@ describe("Codex ACPX harness driver", () => {
}));
for (const activeTurnId of [turnId, null]) {
await expect(fixture.driver.recoverSession!({
...snapshot,
activeTurnId,
})).resolves.toEqual({
await expect(
fixture.driver.recoverSession!({
...snapshot,
activeTurnId,
}),
).resolves.toEqual({
recovered: false,
reason:
"persisted Codex ACPX resultless recovery requires a completed terminal turn",
@ -2403,7 +2419,9 @@ function driverFixture(
finishTurn(result: Awaited<AcpxRuntimeTurn["result"]>): void;
} {
let turnCount = 0;
let activeResult: ReturnType<typeof deferred<Awaited<AcpxRuntimeTurn["result"]>>> | null = null;
let activeResult: ReturnType<
typeof deferred<Awaited<AcpxRuntimeTurn["result"]>>
> | null = null;
const createTurn = (): AcpxRuntimeTurn => {
activeResult = deferred<Awaited<AcpxRuntimeTurn["result"]>>();
return {
@ -2442,7 +2460,10 @@ function driverFixture(
};
};
const host = fakeHost(createTurn, () =>
activeResult?.resolve({ status: "cancelled", stopReason: "session_closed" }),
activeResult?.resolve({
status: "cancelled",
stopReason: "session_closed",
}),
);
let hostOptions: OpenAcpxRuntimeHostOptions | null = null;
const openHost = vi.fn(

View File

@ -62,6 +62,7 @@ import {
import {
ACPX_TURN_CANCELLATION_SHUTDOWN_BOUND_MS,
AcpxRuntimeHost,
type AcpxRetainedCleanupFailure,
type AcpxRuntimeTurn,
type OpenAcpxRuntimeHostOptions,
} from "./runtime-host.js";
@ -197,8 +198,6 @@ export class CodexAcpxDriver implements HarnessDriver {
AcpxRuntimeHost.open(hostOptions, {
openRuntime: openCodexAcpxRuntime,
reportRetainedCleanupFailure: reportRetainedAcpxCleanupFailure,
} as Parameters<typeof AcpxRuntimeHost.open>[1] & {
reportRetainedCleanupFailure: typeof reportRetainedAcpxCleanupFailure;
}));
this.#closeSettlementTimeoutMs =
dependencies.closeSettlementTimeoutMs ?? CLOSE_TURN_SETTLEMENT_TIMEOUT_MS;
@ -1999,11 +1998,9 @@ function safeMessage(error: unknown): string {
.slice(0, 4_000);
}
function reportRetainedAcpxCleanupFailure(input: {
resource: "credential" | "command" | "runtime";
attempt: number;
error: unknown;
}): void {
function reportRetainedAcpxCleanupFailure(
input: AcpxRetainedCleanupFailure,
): void {
const errorName = input.error instanceof Error ? input.error.name : "Error";
process.emitWarning(
JSON.stringify({

View File

@ -70,6 +70,15 @@ describe("managed Codex credentials", () => {
});
expect(lease.mode).toBe("inline_json");
expect(lease.lifetimeFenceFds).toHaveLength(2);
expect(lease.lifetimeFenceFds.every(Number.isSafeInteger)).toBe(true);
expect(lease.lifetimeFenceFds[0]).not.toBe(lease.lifetimeFenceFds[1]);
await expect(
lease.activateLifetimeOwner(process.pid),
).resolves.toBeUndefined();
await expect(lease.activateLifetimeOwner(0)).rejects.toThrow(
"lifetime owner is invalid",
);
const cleanupIntent = join(
fixture.home,
".paperclip-auth-cleanup-required",

View File

@ -46,6 +46,8 @@ try {
interface CredentialHomeLock {
assertHeld(): void;
inheritanceFds(): readonly [number, number];
activateLifetimeOwner(pid: number): Promise<void>;
release(): Promise<void>;
}
@ -112,6 +114,10 @@ export type ManagedCodexCredentialMode =
export interface ManagedCodexCredentialLease {
readonly path: string;
readonly mode: ManagedCodexCredentialMode;
/** Duplicate both quorum listeners into the provider lifetime sentinel. */
readonly lifetimeFenceFds: readonly [number, number];
/** Validate the guardian while the credential quorum is still held. */
activateLifetimeOwner(pid: number): Promise<void>;
close(): Promise<void>;
}
@ -340,6 +346,24 @@ async function acquireCredentialHomeLock(
await Promise.allSettled(servers.map(closeCredentialLeaseServer));
throw error;
}
let inheritanceFds: readonly [number, number];
try {
const first = credentialLeaseServerFd(servers[0]!);
const second = credentialLeaseServerFd(servers[1]!);
if (first === second) {
throw new Error(
"Managed Codex credential ownership listeners are not distinct",
);
}
inheritanceFds = Object.freeze([first, second]) as readonly [
number,
number,
];
} catch (error) {
released = true;
await Promise.allSettled(servers.map(closeCredentialLeaseServer));
throw error;
}
return Object.freeze({
assertHeld(): void {
@ -352,6 +376,16 @@ async function acquireCredentialHomeLock(
throw new Error("Managed Codex credential ownership was lost");
}
},
inheritanceFds(): readonly [number, number] {
this.assertHeld();
return inheritanceFds;
},
async activateLifetimeOwner(pid: number): Promise<void> {
this.assertHeld();
if (!Number.isSafeInteger(pid) || pid < 1) {
throw new Error("Managed Codex credential lifetime owner is invalid");
}
},
async release(): Promise<void> {
if (released) return;
const outcomes = await Promise.allSettled(
@ -396,6 +430,15 @@ function credentialLeasePorts(home: string): readonly number[] {
);
}
function credentialLeaseServerFd(server: Server): number {
const fd = (server as Server & { _handle?: { fd?: unknown } })._handle?.fd;
if (!Number.isSafeInteger(fd) || (fd as number) < 0) {
throw new Error(
"Managed Codex credential ownership listener cannot be inherited",
);
}
return fd as number;
}
async function listenForCredentialLease(
server: Server,
port: number,
@ -551,13 +594,29 @@ function credentialLease(
lock.assertHeld();
let closed = false;
let closeAttempt: Promise<void> | null = null;
let lifetimeOwnerAttempt: Promise<void> | null = null;
return Object.freeze({
path,
mode,
lifetimeFenceFds: lock.inheritanceFds(),
async activateLifetimeOwner(pid: number): Promise<void> {
if (closed || closeAttempt !== null) {
throw new Error("Managed Codex credential lease is closing");
}
if (lifetimeOwnerAttempt !== null) return await lifetimeOwnerAttempt;
const attempt = lock.activateLifetimeOwner(pid);
lifetimeOwnerAttempt = attempt;
try {
await attempt;
} finally {
if (lifetimeOwnerAttempt === attempt) lifetimeOwnerAttempt = null;
}
},
async close(): Promise<void> {
if (closed) return;
if (closeAttempt !== null) return await closeAttempt;
const attempt = (async () => {
await lifetimeOwnerAttempt?.catch(() => undefined);
const activeGeneration = activeCredentialLeaseGenerations.get(home);
if (activeGeneration !== ownerGeneration) {
// A failed close releases its generation only after publishing a

View File

@ -18,29 +18,39 @@ import type {
AcpxRuntimePortIdentity,
AcpxRuntimePortOpenOptions,
} from "./runtime-host.js";
import {
assertVerifiedAcpxProviderPlatform,
awaitVerifiedAcpxProviderExit,
awaitVerifiedAcpxProviderOwnership,
} from "./installation-integrity.js";
import { decideAcpxPermission } from "./permission-policy.js";
const VERIFIED_COMMAND_SENTINEL = "paperclip-verified-acpx-command";
const DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS = 2_000;
const MAX_ADMISSION_CLEANUP_RETRY_ATTEMPTS = 3;
const MAX_ADMISSION_CLEANUP_ATTEMPTS = 1 + MAX_ADMISSION_CLEANUP_RETRY_ATTEMPTS;
const RETAINED_ADMISSION_CLEANUP_RETRY_MIN_MS = 10;
const RETAINED_ADMISSION_CLEANUP_RETRY_MAX_MS = 30_000;
const RETAINED_ADMISSION_CLEANUP_RETRY_MAX_MS = 100;
const PROVIDER_TERM_EXIT_TIMEOUT_MS = 2_000;
const PROVIDER_KILL_EXIT_TIMEOUT_MS = 2_000;
const PROVIDER_SHUTDOWN_SCHEDULING_MARGIN_MS = 1_000;
const MAX_LATE_RUNTIME_CLEANUP_RECONCILIATION_ATTEMPTS = 3;
// Production shutdown waits for the protocol close bound before beginning the
// sequential TERM/KILL verification windows. Keep this exported package-local
// bound aligned with the implementation so admission can include the complete
// provider cleanup path instead of accounting for only part of it.
// sequential TERM/KILL verification windows plus a finite scheduling margin.
// Keep this exported package-local bound aligned with the complete
// implementation.
export const DEFAULT_CODEX_ACPX_RUNTIME_SHUTDOWN_BOUND_MS =
DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS +
PROVIDER_TERM_EXIT_TIMEOUT_MS +
PROVIDER_KILL_EXIT_TIMEOUT_MS;
PROVIDER_KILL_EXIT_TIMEOUT_MS +
PROVIDER_SHUTDOWN_SCHEDULING_MARGIN_MS;
// A close may outlive its caller-facing wait bound. Keep every exact attempt
// owned until it settles. A handle never starts a second protocol close while
// the first remains unresolved; late failure can start bounded reconciliation
// only after the exact attempt reaches a terminal outcome.
const activeRuntimeCleanupOwners = new Set<Promise<unknown>>();
const activeCodexRuntimeCleanupOwners = new Set<Promise<unknown>>();
const SESSION_HANDSHAKE_TIMEOUT_MS = 8_000;
class AcpxRuntimeCloseTimeoutError extends Error {
constructor() {
@ -49,6 +59,20 @@ class AcpxRuntimeCloseTimeoutError extends Error {
}
}
class AcpxRuntimeCloseFinalTimeoutError extends Error {
constructor() {
super("ACPX runtime close remained pending after its final cleanup watch");
this.name = "AcpxRuntimeCloseFinalTimeoutError";
}
}
class AcpxSessionHandshakeTimeoutError extends Error {
constructor() {
super("ACPX session handshake exceeded its admission deadline");
this.name = "AcpxSessionHandshakeTimeoutError";
}
}
export interface CodexAcpxRuntimeDependencies {
createRuntime?: (options: AcpRuntimeOptions) => AcpRuntime;
createRegistry?: (input: {
@ -56,8 +80,16 @@ export interface CodexAcpxRuntimeDependencies {
}) => AcpAgentRegistry;
createStore?: (input: { stateDir: string }) => AcpSessionStore;
runtimeCloseTimeoutMs?: number;
/** Internal test seam for autonomous failed-admission cleanup ownership. */
/** Internal test seam for the provider-session admission deadline. */
sessionHandshakeTimeoutMs?: number;
/** Internal test seam for verified guardian ownership transfer. */
awaitProviderOwnership?: (child: ChildProcess) => Promise<void>;
/** Internal test seam for independent provider-exit proof. */
awaitProviderExit?: (child: ChildProcess) => Promise<void>;
/** Retains autonomous cleanup ownership across the sidecar lifecycle. */
retainCleanup?: (cleanup: Promise<void>) => void;
/** Internal test seam for the fail-closed platform admission boundary. */
platform?: NodeJS.Platform;
}
/**
@ -77,21 +109,31 @@ export async function openCodexAcpxRuntime(
options.retainFailedAdmissionCleanup(Promise.resolve());
throw options.signal.reason;
}
// Verified ACPX command admission already fails closed on Windows because
// Node cannot atomically open the provider executable with O_NOFOLLOW there.
// Reject at the adapter boundary too: allowing a fabricated command lease to
// start a provider would create a cleanup state that cannot guarantee both a
// bounded sidecar exit and retained ownership of an unresponsive process
// tree when Node cannot safely signal a verified provider process group.
assertVerifiedAcpxProviderPlatform(dependencies.platform ?? process.platform);
if (options.profile.agent !== "codex") {
throw new Error(
"The production ACPX runtime currently supports Codex only",
);
}
// The verified-command boundary already refuses to mint a Windows command
// lease, because Node cannot pin its executable there. Repeat the platform
// gate at this lower boundary so alternate host wiring cannot launch a
// credential-bearing provider without a killable tree. `child.kill()` only
// terminates the direct Windows process, and taskkill cannot reliably find
// descendants after their original parent has exited; Windows support must
// therefore wait for an owned Job Object or equivalent containment.
if (process.platform === "win32") {
options.signal?.throwIfAborted();
const credentialFenceFds = options.credentialFenceFds;
if (
!Array.isArray(credentialFenceFds) ||
credentialFenceFds.length !== 2 ||
credentialFenceFds.some(
(fd) => !Number.isSafeInteger(fd) || (fd as number) < 0,
) ||
credentialFenceFds[0] === credentialFenceFds[1] ||
typeof options.activateCredentialFenceOwner !== "function"
) {
throw new Error(
"The production ACPX runtime requires provider process-tree containment unavailable on Windows",
"The production ACPX runtime requires an inherited credential-home fence",
);
}
@ -100,11 +142,6 @@ export async function openCodexAcpxRuntime(
const createRuntime = dependencies.createRuntime ?? createAcpRuntime;
const runtimeCloseTimeoutMs =
dependencies.runtimeCloseTimeoutMs ?? DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS;
const children = new SpawnedChildSet();
const baseStore = createStore({ stateDir: options.stateDirectory });
let failedHandshakeHandle: AcpRuntimeHandle | null = null;
let admissionCleanup: RuntimeAdmissionCleanup | null = null;
let abortedHandshakeCleanup: Promise<void> | null = null;
const retainedCleanupOwners = new WeakSet<Promise<void>>();
const retainCleanup = (cleanup: Promise<void>): void => {
if (retainedCleanupOwners.has(cleanup)) {
@ -114,6 +151,14 @@ export async function openCodexAcpxRuntime(
dependencies.retainCleanup?.(cleanup);
retainCodexRuntimeCleanup(cleanup);
};
const children = new SpawnedChildSet(
retainCleanup,
dependencies.awaitProviderOwnership,
dependencies.awaitProviderExit,
);
const baseStore = createStore({ stateDir: options.stateDirectory });
let failedHandshakeHandle: AcpRuntimeHandle | null = null;
let admissionCleanup: RuntimeAdmissionCleanup | null = null;
const rememberHandshakeHandle = (record: AcpSessionRecord): void => {
const runtimeSessionName = record.name?.trim();
if (
@ -219,16 +264,12 @@ export async function openCodexAcpxRuntime(
// been cancelled. Check at the last host-owned boundary so a late
// handshake cannot create a provider process after authority is gone.
options.signal?.throwIfAborted();
// A verified provider can create descendants that inherit its launch
// credential. Give the provider a dedicated POSIX process group so
// cleanup authority covers that complete credential-bearing tree.
options.assertWorkspaceHeld?.();
return children.add(
options.command.spawn(input.args, {
...input.options,
detached: true,
options.command.spawn(input.args, input.options, {
credentialFenceFds,
activateCredentialFenceOwner: options.activateCredentialFenceOwner!,
}) as ChildProcess,
true,
);
},
});
@ -236,75 +277,66 @@ export async function openCodexAcpxRuntime(
runtime,
children,
runtimeCloseTimeoutMs,
retainCleanup,
);
const handshake = Promise.resolve().then(() =>
runtime.ensureSession({
sessionKey: options.providerSessionKey,
agent: "codex",
mode: "persistent",
cwd: options.cwd,
sessionOptions: {
model: options.profile.qualificationModel,
...(options.systemInstructions
? { systemPrompt: { append: options.systemInstructions } }
: {}),
},
}),
);
let handle: AcpRuntimeHandle | null = null;
let lateCleanup: Promise<void> | null = null;
try {
const handshake = Promise.resolve().then(() =>
runtime.ensureSession({
sessionKey: options.providerSessionKey,
agent: "codex",
mode: "persistent",
cwd: options.cwd,
sessionOptions: {
model: options.profile.qualificationModel,
...(options.systemInstructions
? { systemPrompt: { append: options.systemInstructions } }
: {}),
},
}),
const boundedHandshake = boundedSessionHandshake(
handshake,
dependencies.sessionHandshakeTimeoutMs ?? SESSION_HANDSHAKE_TIMEOUT_MS,
);
if (options.signal === undefined) {
handle = await handshake;
} else {
try {
handle = await raceRuntimeHandshakeWithAbort(handshake, options.signal);
} catch (error) {
if (options.signal.aborted) {
abortedHandshakeCleanup = handshake.then(
(lateHandle) =>
admissionCleanup!.runRetained(
lateHandle,
"ACPX runtime admission aborted",
),
() =>
admissionCleanup!.runRetained(
failedHandshakeHandle,
"ACPX runtime admission aborted",
),
);
retainCleanup(abortedHandshakeCleanup);
}
throw error;
}
// The promise and abort notification can settle in the same turn. Do
// not admit a handle if cancellation won immediately afterward.
options.signal.throwIfAborted();
}
handle = options.signal
? await raceRuntimeHandshakeWithAbort(boundedHandshake, options.signal)
: await boundedHandshake;
// A provider can answer only after the verified sentinel is armed, but do
// not admit the session until the owner has observed that exact handoff.
await children.verifyLifetimeOwnership();
// The handshake or lifetime-ownership observation can settle in the same
// turn as cancellation. Never admit that newly acquired authority.
options.signal?.throwIfAborted();
} catch (error) {
const aborted = options.signal?.aborted === true;
if (aborted || error instanceof AcpxSessionHandshakeTimeoutError) {
lateCleanup = lateHandshakeCleanup(
handshake,
admissionCleanup,
aborted
? "ACPX runtime admission aborted"
: "ACPX session handshake completed after its admission deadline",
);
retainCleanup(lateCleanup);
}
const cleanupHandle = handle ?? failedHandshakeHandle;
const cleanupReason = aborted
? "ACPX runtime admission aborted"
: "ACPX session handshake failed";
const cleanupErrors = await admissionCleanup.run(
cleanupHandle,
options.signal?.aborted
? "ACPX runtime admission aborted"
: "ACPX session handshake failed",
cleanupReason,
);
const retainedCleanup =
cleanupErrors.length === 0
? Promise.resolve()
: admissionCleanup.runRetained(
cleanupHandle,
options.signal?.aborted
? "ACPX runtime admission aborted"
: "ACPX session handshake failed",
);
: admissionCleanup.runRetained(cleanupHandle, cleanupReason);
const cleanupProof =
abortedHandshakeCleanup === null
lateCleanup === null
? retainedCleanup
: Promise.all([retainedCleanup, abortedHandshakeCleanup]).then(
() => undefined,
);
: Promise.all([retainedCleanup, lateCleanup]).then(() => undefined);
options.retainFailedAdmissionCleanup(cleanupProof);
retainCleanup(cleanupProof);
if (cleanupErrors.length > 0) {
@ -330,17 +362,12 @@ export async function openCodexAcpxRuntime(
runtimeCloseTimeoutMs,
);
} catch (error) {
const cleanupErrors = await admissionCleanup.run(
handle,
"ACPX runtime identity validation failed",
);
const cleanupReason = "ACPX runtime identity validation failed";
const cleanupErrors = await admissionCleanup.run(handle, cleanupReason);
const cleanupProof =
cleanupErrors.length === 0
? Promise.resolve()
: admissionCleanup.runRetained(
handle,
"ACPX runtime identity validation failed",
);
: admissionCleanup.runRetained(handle, cleanupReason);
options.retainFailedAdmissionCleanup(cleanupProof);
retainCleanup(cleanupProof);
if (cleanupErrors.length > 0) {
@ -398,6 +425,7 @@ class RuntimeAdmissionCleanup {
string,
Promise<unknown | undefined>
>();
readonly #handleAttemptCounts = new Map<string, number>();
readonly #registeredTargets = new Map<
string,
RuntimeAdmissionCleanupTarget
@ -409,7 +437,6 @@ class RuntimeAdmissionCleanup {
private readonly runtime: AcpRuntime,
private readonly children: SpawnedChildSet,
private readonly runtimeCloseTimeoutMs: number,
private readonly retainCleanup: (cleanup: Promise<void>) => void,
) {}
run(handle: AcpRuntimeHandle | null, reason: string): Promise<unknown[]> {
@ -417,12 +444,9 @@ class RuntimeAdmissionCleanup {
runtimeAdmissionCleanupTargetKey(handle),
handle,
);
return this.#runAttempt(targetKey, handle, reason).then(({ errors }) => {
if (errors.length > 0) {
this.retainCleanup(this.runRetained(handle, reason));
}
return errors;
});
return this.#runAttempt(targetKey, handle, reason).then(
({ errors }) => errors,
);
}
runRetained(handle: AcpRuntimeHandle | null, reason: string): Promise<void> {
@ -430,12 +454,13 @@ class RuntimeAdmissionCleanup {
const targetKey = this.#resolveTargetKey(rawTargetKey, handle);
const existing = this.#registeredTargets.get(targetKey);
if (existing !== undefined) {
existing.handle =
existing.handle === null
? handle
: handle === null
? existing.handle
: preferRuntimeAdmissionCleanupHandle(existing.handle, handle);
if (existing.handle === null) existing.handle = handle;
else if (handle !== null) {
existing.handle = preferRuntimeAdmissionCleanupHandle(
existing.handle,
handle,
);
}
this.#targetAliases.set(rawTargetKey, targetKey);
return existing.cleanup!;
}
@ -492,24 +517,58 @@ class RuntimeAdmissionCleanup {
targetKey: string,
target: RuntimeAdmissionCleanupTarget,
): Promise<void> {
let runtimeTerminalError: unknown | null = null;
let processErrors: unknown[] = [];
let retryDelayMs = RETAINED_ADMISSION_CLEANUP_RETRY_MIN_MS;
// Retained cleanup is the continuing owner. Keep one runtime-close attempt
// in flight at a time and retry process-tree termination until both are
// confirmed complete; a finite budget would recreate an orphan boundary.
for (;;) {
const attempt = await this.#runAttempt(
targetKey,
target.handle,
runtimeTerminalError === null ? target.handle : null,
target.reason,
);
const runtimeNeedsRetry =
target.handle !== null &&
attempt.runtimeError !== undefined &&
!this.#closedHandles.has(targetKey);
const processNeedsRetry = attempt.processErrors.length > 0;
if (!runtimeNeedsRetry && !processNeedsRetry) {
return;
let runtimeError = attempt.runtimeError;
processErrors = attempt.processErrors;
if (attempt.pendingRuntimeClose !== undefined) {
// The configured timeout bounds the caller-facing pass, not the exact
// close promise. Give that exact promise one final bounded watch. A
// late rejection can then admit the next actual close attempt, while
// a second timeout terminalizes protocol cleanup without overlap.
const lateOutcome = await closeOutcomeWithin(
attempt.pendingRuntimeClose,
this.runtimeCloseTimeoutMs,
);
if (lateOutcome instanceof AcpxRuntimeCloseTimeoutError) {
runtimeTerminalError = new AcpxRuntimeCloseFinalTimeoutError();
runtimeError = runtimeTerminalError;
} else {
runtimeError = lateOutcome;
}
}
if (
runtimeTerminalError === null &&
runtimeError !== undefined &&
(this.#handleAttemptCounts.get(targetKey) ?? 0) >=
MAX_ADMISSION_CLEANUP_ATTEMPTS
) {
runtimeTerminalError = new AggregateError(
[runtimeError],
`ACPX failed-admission cleanup exhausted ${MAX_ADMISSION_CLEANUP_RETRY_ATTEMPTS} retry attempts`,
);
}
const runtimeNeedsRetry =
runtimeTerminalError === null &&
runtimeError !== undefined &&
!this.#closedHandles.has(targetKey);
const processNeedsRetry = processErrors.length > 0;
if (!runtimeNeedsRetry && !processNeedsRetry) {
if (runtimeTerminalError === null) return;
throw runtimeTerminalError;
}
// Runtime close retries are bounded above, but a live provider cannot
// be abandoned merely because its first termination passes failed.
// Keep this retained owner active with bounded backoff until process
// exit is observed. Once the process is gone, any terminal protocol
// cleanup error is still reported to the owner below.
await delay(retryDelayMs);
retryDelayMs = Math.min(
retryDelayMs * 2,
@ -526,13 +585,21 @@ class RuntimeAdmissionCleanup {
errors: unknown[];
runtimeError: unknown | undefined;
processErrors: unknown[];
pendingRuntimeClose?: Promise<unknown | undefined>;
}> {
const cleanup = this.#tail.then(async () => {
const errors: unknown[] = [];
let runtimeError: unknown | undefined;
let pendingRuntimeClose: Promise<unknown | undefined> | undefined;
if (handle !== null && !this.#closedHandles.has(targetKey)) {
runtimeError = await this.#closeHandleWithin(targetKey, handle, reason);
const runtimeOutcome = await this.#closeHandleWithin(
targetKey,
handle,
reason,
);
runtimeError = runtimeOutcome.error;
if (runtimeError !== undefined) errors.push(runtimeError);
pendingRuntimeClose = runtimeOutcome.pendingAttempt;
}
const processErrors = await this.children.terminate();
errors.push(...processErrors);
@ -540,6 +607,7 @@ class RuntimeAdmissionCleanup {
errors,
runtimeError,
processErrors,
...(pendingRuntimeClose === undefined ? {} : { pendingRuntimeClose }),
};
});
this.#tail = cleanup.then(
@ -553,9 +621,21 @@ class RuntimeAdmissionCleanup {
targetKey: string,
handle: AcpRuntimeHandle,
reason: string,
): Promise<unknown | undefined> {
): Promise<{
error: unknown | undefined;
pendingAttempt?: Promise<unknown | undefined>;
}> {
let attempt = this.#activeHandleAttempts.get(targetKey);
if (attempt === undefined) {
const attemptCount = this.#handleAttemptCounts.get(targetKey) ?? 0;
if (attemptCount >= MAX_ADMISSION_CLEANUP_ATTEMPTS) {
return {
error: new Error(
`ACPX failed-admission cleanup exhausted ${MAX_ADMISSION_CLEANUP_RETRY_ATTEMPTS} retry attempts`,
),
};
}
this.#handleAttemptCounts.set(targetKey, attemptCount + 1);
attempt = runtimeCloseOutcome(this.runtime, {
handle,
reason,
@ -569,7 +649,10 @@ class RuntimeAdmissionCleanup {
if (error === undefined) this.#closedHandles.add(targetKey);
});
}
return await closeOutcomeWithin(attempt, this.runtimeCloseTimeoutMs);
const error = await closeOutcomeWithin(attempt, this.runtimeCloseTimeoutMs);
return error instanceof AcpxRuntimeCloseTimeoutError
? { error, pendingAttempt: attempt }
: { error };
}
}
@ -644,6 +727,38 @@ function nonEmptyRuntimeIdentity(
return typeof value === "string" && value.length > 0 ? value : undefined;
}
async function boundedSessionHandshake(
handshake: Promise<AcpRuntimeHandle>,
timeoutMs: number,
): Promise<AcpRuntimeHandle> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
handshake,
new Promise<never>((_resolve, reject) => {
timer = setTimeout(
() => reject(new AcpxSessionHandshakeTimeoutError()),
timeoutMs,
);
timer.unref();
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
function lateHandshakeCleanup(
handshake: Promise<AcpRuntimeHandle>,
cleanup: RuntimeAdmissionCleanup,
reason: string,
): Promise<void> {
return handshake.then(
(lateHandle) => cleanup.runRetained(lateHandle, reason),
() => undefined,
);
}
function runtimePort(
runtime: AcpRuntime,
handle: AcpRuntimeHandle,
@ -901,9 +1016,7 @@ function runtimePort(
mode: "prompt",
requestId: input.requestId,
...(input.signal ? { signal: input.signal } : {}),
...(input.onElicitation
? { onElicitation: input.onElicitation }
: {}),
...(input.onElicitation ? { onElicitation: input.onElicitation } : {}),
});
},
close: closeRuntime,
@ -1005,177 +1118,319 @@ async function boundedCloseOutcome(
}
function delay(timeoutMs: number): Promise<void> {
return new Promise((resolve) => {
const timer = setTimeout(resolve, timeoutMs);
timer.unref?.();
});
return new Promise((resolve) => setTimeout(resolve, timeoutMs));
}
type ProviderExitOutcome =
| { exited: true }
| { exited: false; error: unknown };
class ProviderExitObservation {
#outcome: ProviderExitOutcome | null = null;
readonly #observers = new Set<(outcome: ProviderExitOutcome) => void>();
constructor(providerExit: Promise<void>) {
void providerExit.then(
() => this.#settle({ exited: true }),
(error: unknown) => this.#settle({ exited: false, error }),
);
}
observe(observer: (outcome: ProviderExitOutcome) => void): void {
if (this.#outcome) observer(this.#outcome);
else this.#observers.add(observer);
}
async waitWithin(
timeoutMs: number,
): Promise<{ exited: boolean; error?: unknown }> {
if (this.#outcome) return this.#outcome;
return await new Promise((resolve) => {
const finish = (outcome: ProviderExitOutcome | { exited: false }) => {
clearTimeout(timer);
this.#observers.delete(finish);
resolve(outcome);
};
const timer = setTimeout(() => finish({ exited: false }), timeoutMs);
timer.unref();
this.#observers.add(finish);
if (this.#outcome) finish(this.#outcome);
});
}
#settle(outcome: ProviderExitOutcome): void {
if (this.#outcome) return;
this.#outcome = outcome;
for (const observer of this.#observers) observer(outcome);
this.#observers.clear();
}
}
class SpawnedChildSet {
readonly #children = new Map<ChildProcess, SpawnedProviderProcess>();
readonly #children = new Set<ChildProcess>();
readonly #errors = new Set<unknown>();
readonly #providerExits = new Map<ChildProcess, ProviderExitObservation>();
readonly #terminations = new Map<ChildProcess, Promise<unknown[]>>();
readonly #lifetimeOwnership: Promise<void>[] = [];
#lifetimeOwnershipSealed = false;
#sealed = false;
add(child: ChildProcess, processGroup: boolean): ChildProcess {
constructor(
private readonly retainCleanup?: (cleanup: Promise<void>) => void,
private readonly awaitProviderOwnership: (
child: ChildProcess,
) => Promise<void> = awaitVerifiedAcpxProviderOwnership,
private readonly awaitProviderExit: (
child: ChildProcess,
) => Promise<void> = awaitVerifiedAcpxProviderExit,
) {}
add(child: ChildProcess): ChildProcess {
let exitProof: Promise<void>;
try {
exitProof = this.awaitProviderExit(child);
} catch (error) {
exitProof = Promise.reject(error);
}
const providerExit = new ProviderExitObservation(exitProof);
this.#providerExits.set(child, providerExit);
this.#track(child, providerExit);
const ownership = this.awaitProviderOwnership(child);
void ownership.catch(() => undefined);
this.#lifetimeOwnership.push(ownership);
if (this.#sealed || this.#lifetimeOwnershipSealed) {
// Once the stable-empty cleanup point is sealed, ACPX no longer has
// authority to create provider work. Retain an immediate-kill attempt
// through exit verification before rejecting the spawn itself.
const termination = this.#startTermination(child, true);
const cleanup = termination.then((errors) => {
if (errors.length > 0) {
throw new AggregateError(
errors,
"ACPX post-seal provider cleanup failed",
);
}
});
this.retainCleanup?.(cleanup);
void cleanup.catch(() => undefined);
throw new Error(
this.#sealed
? "ACPX provider spawned after cleanup was sealed"
: "ACPX provider spawned after ownership admission was sealed",
);
}
return child;
}
async verifyLifetimeOwnership(): Promise<void> {
for (;;) {
const ownership = this.#lifetimeOwnership.splice(0);
if (ownership.length === 0) {
// This check and seal are synchronous. Any spawn added while an
// earlier batch was pending is observed by the next loop iteration;
// no later provider can race admission after the stable-empty point.
this.#lifetimeOwnershipSealed = true;
return;
}
await Promise.all(ownership);
}
}
#track(child: ChildProcess, providerExit: ProviderExitObservation): void {
this.#children.add(child);
const onError = (error: unknown) => this.#errors.add(error);
const tracked: SpawnedProviderProcess = {
child,
processGroupId: processGroup ? (child.pid ?? null) : null,
onError,
let guardianExited = !running(child);
let providerExited = false;
const forgetIfReleased = () => {
if (!guardianExited || !providerExited) return;
this.#children.delete(child);
this.#providerExits.delete(child);
child.off("error", onError);
child.off("exit", onGuardianExit);
child.off("close", onGuardianExit);
};
this.#children.set(child, tracked);
const forgetExitedTree = () => {
if (!providerTreeRunning(tracked)) this.#forget(tracked);
const onGuardianExit = () => {
guardianExited = true;
forgetIfReleased();
};
// ChildProcess reports some spawn and signal-delivery failures through an
// asynchronous `error` event. Observe those for the child's whole tracked
// lifetime so cleanup can report them instead of crashing runnerd.
child.on("error", onError);
child.once("exit", forgetExitedTree);
child.once("close", forgetExitedTree);
return child;
child.once("exit", onGuardianExit);
child.once("close", onGuardianExit);
providerExit.observe((outcome) => {
if (outcome.exited) {
providerExited = true;
forgetIfReleased();
} else {
this.#errors.add(outcome.error);
}
});
}
async terminate(): Promise<unknown[]> {
const errors: unknown[] = [];
const children = [...this.#children.values()];
await Promise.all(
children.map(async (tracked) => {
if (providerTreeRunning(tracked)) {
const terminateOutcome = await signalAndWaitForExit(
tracked,
"SIGTERM",
PROVIDER_TERM_EXIT_TIMEOUT_MS,
);
if (terminateOutcome.error !== undefined) {
pushUnique(errors, terminateOutcome.error);
}
if (!terminateOutcome.exited && providerTreeRunning(tracked)) {
const killOutcome = await signalAndWaitForExit(
tracked,
"SIGKILL",
PROVIDER_KILL_EXIT_TIMEOUT_MS,
);
if (killOutcome.error !== undefined) {
pushUnique(errors, killOutcome.error);
}
if (!killOutcome.exited && providerTreeRunning(tracked)) {
errors.push(
new Error("ACPX provider did not exit after SIGKILL"),
);
}
}
}
if (!providerTreeRunning(tracked)) this.#forget(tracked);
}),
);
// Revoke spawn authority synchronously before the first await. Children
// already owned here receive the normal TERM/KILL sequence; every later
// spawn is rejected and its independently retained post-seal cleanup
// cannot extend this caller-facing shutdown without bound.
this.#sealed = true;
for (const child of this.#children) this.#startTermination(child);
const ownedTerminations = [...this.#terminations.values()];
const errors = (await Promise.all(ownedTerminations)).flat();
// A failed spawn or signal can emit `error` and then `close` before this
// method snapshots the live children. Keep those errors independently of
// child membership, report each object once, and drain them only after all
// in-flight termination attempts have had a chance to emit.
// child membership and report each object once after all owned attempts.
for (const error of this.#errors) pushUnique(errors, error);
this.#errors.clear();
return errors;
}
#forget(tracked: SpawnedProviderProcess): void {
if (this.#children.get(tracked.child) !== tracked) return;
this.#children.delete(tracked.child);
tracked.child.off("error", tracked.onError);
#startTermination(
child: ChildProcess,
immediateKill = false,
): Promise<unknown[]> {
const existing = this.#terminations.get(child);
if (existing) return existing;
const providerExit =
this.#providerExits.get(child) ??
new ProviderExitObservation(
Promise.reject(new Error("ACPX provider exit proof is unavailable")),
);
const termination = (
immediateKill
? terminatePostSealChild(child, providerExit)
: terminateChild(child, providerExit)
).catch((error: unknown) => [error]);
this.#terminations.set(child, termination);
termination.then(() => {
if (this.#terminations.get(child) === termination) {
this.#terminations.delete(child);
}
});
return termination;
}
}
interface SpawnedProviderProcess {
child: ChildProcess;
processGroupId: number | null;
onError: (error: unknown) => void;
async function terminatePostSealChild(
child: ChildProcess,
providerExit: ProviderExitObservation,
): Promise<unknown[]> {
const errors: unknown[] = [];
// Verified production children override ChildProcess.kill so this SIGKILL
// request revokes the owner pipe and wakes the live guardian, which retains
// authority to reap the whole group. Never copy the numeric PGID into a
// later signal owner.
const killOutcome = await signalAndWaitForVerifiedProviderExit(
child,
"SIGKILL",
PROVIDER_KILL_EXIT_TIMEOUT_MS,
providerExit,
);
for (const error of killOutcome.errors) pushUnique(errors, error);
if (!killOutcome.exited) {
errors.push(
new Error("ACPX post-seal provider did not exit after SIGKILL"),
);
}
return errors;
}
async function terminateChild(
child: ChildProcess,
providerExit: ProviderExitObservation,
): Promise<unknown[]> {
const errors: unknown[] = [];
const terminateOutcome = await signalAndWaitForVerifiedProviderExit(
child,
"SIGTERM",
PROVIDER_TERM_EXIT_TIMEOUT_MS,
providerExit,
);
for (const error of terminateOutcome.errors) pushUnique(errors, error);
if (!terminateOutcome.exited) {
errors.push(new Error("ACPX provider did not exit after SIGTERM"));
// A live verified guardian still pins the PGID. Its protected `kill`
// override revokes the owner pipe and wakes it to reap the group. If the
// guardian already exited, do not signal a saved identifier; retain local
// cleanup while waiting for the provider-only descriptor to reach EOF.
const killOutcome = await signalAndWaitForVerifiedProviderExit(
child,
"SIGKILL",
PROVIDER_KILL_EXIT_TIMEOUT_MS,
providerExit,
);
for (const error of killOutcome.errors) pushUnique(errors, error);
if (!killOutcome.exited) {
errors.push(new Error("ACPX provider did not exit after SIGKILL"));
}
}
// Never unref a child whose guardian exit and provider-only EOF were not
// both observed. Local cleanup retains it instead of transferring a reusable
// PGID or releasing credential ownership early.
return errors;
}
function running(child: ChildProcess): boolean {
return child.exitCode === null && child.signalCode === null;
}
function providerTreeRunning(tracked: SpawnedProviderProcess): boolean {
if (tracked.processGroupId === null) return running(tracked.child);
try {
process.kill(-tracked.processGroupId, 0);
return true;
} catch (error) {
return errorCode(error) !== "ESRCH";
}
async function signalAndWaitForVerifiedProviderExit(
child: ChildProcess,
signal: NodeJS.Signals,
timeoutMs: number,
providerExit: ProviderExitObservation,
): Promise<{ exited: boolean; errors: unknown[] }> {
const [guardian, provider] = await Promise.all([
signalAndWaitForExit(child, signal, timeoutMs),
providerExit.waitWithin(timeoutMs),
]);
const errors: unknown[] = [];
if (guardian.error !== undefined) pushUnique(errors, guardian.error);
if (provider.error !== undefined) pushUnique(errors, provider.error);
return {
exited: guardian.exited && provider.exited,
errors,
};
}
async function signalAndWaitForExit(
tracked: SpawnedProviderProcess,
child: ChildProcess,
signal: NodeJS.Signals,
timeoutMs: number,
): Promise<{ exited: boolean; error?: unknown }> {
if (!providerTreeRunning(tracked)) return { exited: true };
const { child } = tracked;
if (!running(child)) return { exited: true };
return await new Promise<{ exited: boolean; error?: unknown }>((resolve) => {
let settled = false;
const finish = (outcome: { exited: boolean; error?: unknown }) => {
if (settled) return;
settled = true;
clearTimeout(timer);
if (poll !== undefined) clearInterval(poll);
child.off("exit", onExit);
child.off("close", onExit);
child.off("error", onError);
resolve(outcome);
};
const onExit = () => {
if (!providerTreeRunning(tracked)) finish({ exited: true });
};
const onExit = () => finish({ exited: true });
const onError = (error: unknown) => finish({ exited: false, error });
const timer = setTimeout(
() => finish({ exited: !providerTreeRunning(tracked) }),
timeoutMs,
);
const timer = setTimeout(() => finish({ exited: false }), timeoutMs);
timer.unref();
const poll =
tracked.processGroupId === null
? undefined
: setInterval(() => {
if (!providerTreeRunning(tracked)) finish({ exited: true });
}, 25);
poll?.unref();
child.once("exit", onExit);
child.once("close", onExit);
child.once("error", onError);
if (!providerTreeRunning(tracked)) {
if (!running(child)) {
finish({ exited: true });
return;
}
try {
if (tracked.processGroupId === null) {
if (!child.kill(signal) && providerTreeRunning(tracked)) {
finish({
exited: false,
error: new Error(`ACPX provider rejected ${signal}`),
});
return;
}
} else {
process.kill(-tracked.processGroupId, signal);
}
if (!providerTreeRunning(tracked)) finish({ exited: true });
child.kill(signal);
if (!running(child)) finish({ exited: true });
} catch (error) {
if (errorCode(error) === "ESRCH" && !providerTreeRunning(tracked)) {
finish({ exited: true });
} else {
finish({ exited: false, error });
}
finish({ exited: false, error });
}
});
}
function errorCode(error: unknown): string | undefined {
if (typeof error !== "object" || error === null || !("code" in error)) {
return undefined;
}
return typeof error.code === "string" ? error.code : undefined;
}
function pushUnique(errors: unknown[], error: unknown): void {
if (!errors.includes(error)) errors.push(error);
}

View File

@ -1,11 +1,12 @@
import { createHash } from "node:crypto";
import type { ChildProcess } from "node:child_process";
import { fork, type ChildProcess } from "node:child_process";
import { once } from "node:events";
import {
chmod,
link,
mkdir,
mkdtemp,
readFile,
realpath,
rename,
rm,
@ -13,6 +14,7 @@ import {
stat,
writeFile,
} from "node:fs/promises";
import { createServer, type Server } from "node:net";
import { tmpdir } from "node:os";
import { dirname, join } from "node:path";
@ -20,14 +22,19 @@ import { afterEach, describe, expect, it, vi } from "vitest";
import { resolveQualifiedAcpxProfile } from "./qualified-profiles.js";
import {
awaitVerifiedAcpxProviderExit,
awaitVerifiedAcpxProviderOwnership,
guardSnapshotModuleLookup,
guardSnapshotModuleResolution,
reapCurrentProviderProcessGroup,
sanitizedNodeEnvironment,
snapshotDescriptorAncestorIndex,
snapshotDescriptorResolution,
verifiedExecutableOpenFlags,
verifyQualifiedAcpxInstallation,
type VerifiedAcpxProviderLifetime,
} from "./installation-integrity.js";
import { stageManagedCodexCredential } from "./codex-credentials.js";
const temporaryDirectories: string[] = [];
const descriptorCommandPath = "/proc/self/fd/4/server.js";
@ -41,6 +48,57 @@ afterEach(async () => {
});
describe("ACPX installation integrity", () => {
it("rejects an unregistered provider exit proof", async () => {
await expect(
awaitVerifiedAcpxProviderExit({} as ChildProcess),
).rejects.toThrow("provider exit proof is unavailable");
});
it("never signals a dead guardian's saved process-group identity", () => {
const signalCurrentGroup = vi.fn(
(_pid: number, _signal: NodeJS.Signals) => true,
);
reapCurrentProviderProcessGroup(
signalCurrentGroup,
4_321,
vi.fn((_code: number) => undefined),
);
expect(signalCurrentGroup).toHaveBeenCalledOnce();
expect(signalCurrentGroup).toHaveBeenCalledWith(0, "SIGKILL");
const signalSelfAfterGroupFailure = vi.fn(
(pid: number, _signal: NodeJS.Signals) => {
if (pid === 0) throw new Error("group signal unavailable");
},
);
const exit = vi.fn((_code: number) => undefined);
reapCurrentProviderProcessGroup(signalSelfAfterGroupFailure, 4_321, exit);
expect(signalSelfAfterGroupFailure.mock.calls).toEqual([
[0, "SIGKILL"],
[4_321, "SIGKILL"],
]);
expect(exit).not.toHaveBeenCalled();
const failedSignals = vi.fn((_pid: number, _signal: NodeJS.Signals) => {
throw new Error("signal unavailable");
});
reapCurrentProviderProcessGroup(failedSignals, 4_321, exit);
expect(failedSignals.mock.calls).toEqual([
[0, "SIGKILL"],
[4_321, "SIGKILL"],
]);
expect(exit).toHaveBeenCalledWith(1);
expect(
[
...signalCurrentGroup.mock.calls,
...signalSelfAfterGroupFailure.mock.calls,
...failedSignals.mock.calls,
]
.map(([pid]) => pid)
.filter((pid) => pid < 0),
).toEqual([]);
});
it("does not delegate non-Linux snapshot filesystem lookups", () => {
for (const platform of ["darwin", "freebsd", "win32"] as const) {
const nextResolve = vi.fn(() => ({ url: "file:///attacker.js" }));
@ -1115,6 +1173,370 @@ describe("ACPX installation integrity", () => {
await expectFailure(child, "requires Linux descriptor-pinned paths");
}
});
it.runIf(process.platform !== "win32")(
"rejects incomplete or duplicate provider credential quorum descriptors",
async () => {
const fixture = await persistentInstallationFixture();
const installation = await verifyQualifiedAcpxInstallation(
fixture.profile,
fixture.resolve,
);
const invalidLifetimes = [
{
credentialFenceFds: [42],
activateCredentialFenceOwner: async () => undefined,
},
{
credentialFenceFds: [42, 42],
activateCredentialFenceOwner: async () => undefined,
},
{
credentialFenceFds: [42, -1],
activateCredentialFenceOwner: async () => undefined,
},
{
credentialFenceFds: [42, 43],
},
] as unknown as readonly VerifiedAcpxProviderLifetime[];
for (const lifetime of invalidLifetimes) {
const command = await installation.openCommand();
expect(() => command.spawn([], {}, lifetime)).toThrow(
"ACPX provider credential fence is invalid",
);
}
},
);
it.runIf(process.platform === "linux")(
"keeps the staged credential fenced through owner SIGKILL and reaps the provider group",
async () => {
const fixture = await persistentInstallationFixture();
const ownerScript = join(fixture.root, "provider-owner.mjs");
const pidFile = join(fixture.root, "provider.pid");
const credentialHome = join(fixture.root, "codex-home");
await mkdir(credentialHome, { mode: 0o700 });
const moduleUrl = new URL("./installation-integrity.ts", import.meta.url)
.href;
const credentialModuleUrl = new URL(
"./codex-credentials.ts",
import.meta.url,
).href;
await writeFile(
ownerScript,
[
`const module = await import(${JSON.stringify(moduleUrl)});`,
`const credentials = await import(${JSON.stringify(credentialModuleUrl)});`,
`const profile = ${JSON.stringify(fixture.profile)};`,
`const credential = await credentials.stageManagedCodexCredential({ agentHomeDirectory: ${JSON.stringify(credentialHome)}, environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"original"}' } });`,
`const paths = new Map(${JSON.stringify([...fixture.paths])});`,
"const installation = await module.verifyQualifiedAcpxInstallation(profile, (name) => paths.get(name));",
"const lease = await installation.openCommand();",
`const provider = lease.spawn([], { env: { ...process.env, PAPERCLIP_PROVIDER_PID_FILE: ${JSON.stringify(pidFile)} } }, { credentialFenceFds: credential.lifetimeFenceFds, activateCredentialFenceOwner: (pid) => credential.activateLifetimeOwner(pid) });`,
"await module.awaitVerifiedAcpxProviderOwnership(provider);",
'process.send?.({ type: "ready", guardianPid: provider.pid });',
"process.stdin.resume();",
].join("\n"),
);
const owner = fork(ownerScript, [], {
execArgv: ["--import", "tsx"],
stdio: ["pipe", "ignore", "pipe", "ipc"],
});
let guardianPid = 0;
let providerPid = 0;
try {
const ready = (await childMessage(owner, "ready")) as {
guardianPid: number;
};
guardianPid = ready.guardianPid;
providerPid = Number.parseInt(await waitForFile(pidFile), 10);
expect(processAlive(providerPid)).toBe(true);
process.kill(guardianPid, "SIGSTOP");
try {
owner.kill("SIGKILL");
await once(owner, "exit");
// The stopped sentinel cannot answer any application protocol. Its
// two inherited quorum listeners nevertheless prevent a second owner.
await expect(
stageManagedCodexCredential({
agentHomeDirectory: credentialHome,
environment: {
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
},
}),
).rejects.toThrow("already has an active lease");
expect(processAlive(providerPid)).toBe(true);
} finally {
if (owner.exitCode === null && owner.signalCode === null) {
owner.kill("SIGKILL");
await once(owner, "exit").catch(() => undefined);
}
// SIGSTOP pins this exact live guardian PID against reuse until the
// matching resume. Owner-pipe EOF then makes it self-reap its group.
process.kill(guardianPid, "SIGCONT");
await waitUntil(() => !processAlive(providerPid));
}
let contender: Awaited<
ReturnType<typeof stageManagedCodexCredential>
> | null = null;
await waitUntilAsync(async () => {
try {
contender = await stageManagedCodexCredential({
agentHomeDirectory: credentialHome,
environment: {
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
},
});
return true;
} catch {
return false;
}
});
await contender!.close();
} finally {
if (owner.exitCode === null && owner.signalCode === null) {
// This direct child handle owns the guardian pipe. Closing it lets
// the live guardian reap only its own still-pinned group; never
// signal a saved guardian PGID from cleanup.
owner.kill("SIGKILL");
await once(owner, "exit").catch(() => undefined);
}
}
},
);
it.runIf(process.platform === "linux")(
"reaps a fenced provider when its lifetime guardian is SIGKILLed",
async () => {
const fixture = await persistentInstallationFixture();
const ownerScript = join(fixture.root, "guardian-owner.mjs");
const pidFile = join(fixture.root, "guardian-provider.pid");
const credentialHome = join(fixture.root, "guardian-codex-home");
await mkdir(credentialHome, { mode: 0o700 });
const moduleUrl = new URL("./installation-integrity.ts", import.meta.url)
.href;
const credentialModuleUrl = new URL(
"./codex-credentials.ts",
import.meta.url,
).href;
await writeFile(
ownerScript,
[
`const module = await import(${JSON.stringify(moduleUrl)});`,
`const credentials = await import(${JSON.stringify(credentialModuleUrl)});`,
`const profile = ${JSON.stringify(fixture.profile)};`,
`const credential = await credentials.stageManagedCodexCredential({ agentHomeDirectory: ${JSON.stringify(credentialHome)}, environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"original"}' } });`,
`const paths = new Map(${JSON.stringify([...fixture.paths])});`,
"const installation = await module.verifyQualifiedAcpxInstallation(profile, (name) => paths.get(name));",
"const lease = await installation.openCommand();",
`const provider = lease.spawn([], { env: { ...process.env, PAPERCLIP_PROVIDER_PID_FILE: ${JSON.stringify(pidFile)} } }, { credentialFenceFds: credential.lifetimeFenceFds, activateCredentialFenceOwner: (pid) => credential.activateLifetimeOwner(pid) });`,
"await module.awaitVerifiedAcpxProviderOwnership(provider);",
'process.send?.({ type: "ready", guardianPid: provider.pid });',
"process.stdin.resume();",
].join("\n"),
);
const owner = fork(ownerScript, [], {
execArgv: ["--import", "tsx"],
stdio: ["pipe", "ignore", "pipe", "ipc"],
});
let guardianPid = 0;
let providerPid = 0;
try {
const ready = (await childMessage(owner, "ready")) as {
guardianPid: number;
};
guardianPid = ready.guardianPid;
providerPid = Number.parseInt(await waitForFile(pidFile), 10);
expect(processAlive(providerPid)).toBe(true);
// Freeze the provider so it cannot process guardian-pipe EOF itself.
// The armed credential-free peer must still reap the current group.
process.kill(providerPid, "SIGSTOP");
await waitUntilAsync(() => processStopped(providerPid));
process.kill(guardianPid, "SIGKILL");
owner.kill("SIGKILL");
await once(owner, "exit");
await waitUntil(() => !processAlive(providerPid));
const contender = await stageManagedCodexCredential({
agentHomeDirectory: credentialHome,
environment: {
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
},
});
await contender.close();
} finally {
if (owner.exitCode === null && owner.signalCode === null) {
owner.kill("SIGKILL");
await once(owner, "exit").catch(() => undefined);
}
if (providerPid > 0 && processAlive(providerPid)) {
// Failure cleanup only: allow the provider's own guardian-loss
// callback to reap its still-pinned group if the watchdog regressed.
process.kill(providerPid, "SIGCONT");
await waitUntil(() => !processAlive(providerPid));
}
}
},
);
it.runIf(process.platform === "linux")(
"reaps a stopped provider after an external guardian kill",
async () => {
const fixture = await persistentInstallationFixture();
const pidFile = join(fixture.root, "provider-exit-proof.pid");
const fences = await Promise.all([
listenOnLoopback(),
listenOnLoopback(),
]);
const fenceFds = fences.map(
(fence) =>
(fence as Server & { _handle?: { fd?: number } })._handle?.fd,
);
expect(fenceFds.every(Number.isSafeInteger)).toBe(true);
const installation = await verifyQualifiedAcpxInstallation(
fixture.profile,
fixture.resolve,
);
const guardian = (await installation.openCommand()).spawn(
[],
{ env: { ...process.env, PAPERCLIP_PROVIDER_PID_FILE: pidFile } },
{
credentialFenceFds: [fenceFds[0]!, fenceFds[1]!],
activateCredentialFenceOwner: async () => undefined,
},
);
await awaitVerifiedAcpxProviderOwnership(guardian);
const providerExit = awaitVerifiedAcpxProviderExit(guardian);
const providerPid = Number.parseInt(await waitForFile(pidFile), 10);
const guardianExit = once(guardian, "exit");
process.kill(providerPid, "SIGSTOP");
await waitUntilAsync(() => processStopped(providerPid));
try {
// Bypass the protected cleanup method to model SIGKILL/OOM of the
// guardian itself. Its credential-free peer must reap the stopped
// provider without waiting for provider JavaScript to run.
process.kill(guardian.pid!, "SIGKILL");
await guardianExit;
await providerExit;
await waitUntil(() => !processAlive(providerPid));
} finally {
if (guardian.exitCode === null && guardian.signalCode === null) {
guardian.kill("SIGKILL");
await guardianExit.catch(() => undefined);
}
if (processAlive(providerPid)) {
process.kill(providerPid, "SIGCONT");
await waitUntil(() => !processAlive(providerPid));
}
await Promise.all(fences.map(closeServer));
}
},
);
it.runIf(process.platform === "linux")(
"dismisses the lifetime sentinel only after normal provider-group cleanup",
async () => {
const fixture = await persistentInstallationFixture();
const pidFile = join(fixture.root, "normal-provider.pid");
const fences = await Promise.all([
listenOnLoopback(),
listenOnLoopback(),
]);
const fenceFds = fences.map(
(fence) =>
(fence as Server & { _handle?: { fd?: number } })._handle?.fd,
);
expect(fenceFds.every(Number.isSafeInteger)).toBe(true);
expect(fenceFds[0]).not.toBe(fenceFds[1]);
const installation = await verifyQualifiedAcpxInstallation(
fixture.profile,
fixture.resolve,
);
const provider = (await installation.openCommand()).spawn(
[],
{ env: { ...process.env, PAPERCLIP_PROVIDER_PID_FILE: pidFile } },
{
credentialFenceFds: [fenceFds[0]!, fenceFds[1]!],
activateCredentialFenceOwner: async () => undefined,
},
);
await awaitVerifiedAcpxProviderOwnership(provider);
const providerPid = Number.parseInt(await waitForFile(pidFile), 10);
const ports = fences.map(
(fence) => (fence.address() as { port: number }).port,
);
provider.kill("SIGTERM");
await Promise.all(fences.map(closeServer));
await once(provider, "exit");
await waitUntil(() => !processAlive(providerPid));
await Promise.all(
ports.map((port) =>
expect(canBindLoopbackPort(port)).resolves.toBe(true),
),
);
},
);
it.runIf(process.platform === "linux")(
"reaps a stopped provider across repeated guardian cleanup requests",
async () => {
const fixture = await persistentInstallationFixture();
const pidFile = join(fixture.root, "emergency-provider.pid");
const fences = await Promise.all([
listenOnLoopback(),
listenOnLoopback(),
]);
const fenceFds = fences.map(
(fence) =>
(fence as Server & { _handle?: { fd?: number } })._handle?.fd,
);
expect(fenceFds.every(Number.isSafeInteger)).toBe(true);
expect(fenceFds[0]).not.toBe(fenceFds[1]);
const installation = await verifyQualifiedAcpxInstallation(
fixture.profile,
fixture.resolve,
);
const guardian = (await installation.openCommand()).spawn(
[],
{ env: { ...process.env, PAPERCLIP_PROVIDER_PID_FILE: pidFile } },
{
credentialFenceFds: [fenceFds[0]!, fenceFds[1]!],
activateCredentialFenceOwner: async () => undefined,
},
);
await awaitVerifiedAcpxProviderOwnership(guardian);
const providerPid = Number.parseInt(await waitForFile(pidFile), 10);
const guardianExit = once(guardian, "exit");
process.kill(providerPid, "SIGSTOP");
process.kill(guardian.pid!, "SIGSTOP");
try {
expect(guardian.kill("SIGKILL")).toBe(true);
// Retry synchronously while the resumed guardian has not yet processed
// owner-pipe EOF. Each retry wakes the exact guardian; the guardian
// remains alive to reap the whole provider group itself.
expect(guardian.kill("SIGKILL")).toBe(true);
await guardianExit;
await waitUntil(() => !processAlive(providerPid));
} finally {
if (processAlive(providerPid)) {
// The stopped provider still pins this exact PID. Resume it only for
// failure cleanup so guardian-pipe EOF can make it self-reap.
process.kill(providerPid, "SIGCONT");
}
if (guardian.exitCode === null && guardian.signalCode === null) {
process.kill(guardian.pid!, "SIGCONT");
guardian.kill("SIGKILL");
await guardianExit.catch(() => undefined);
}
await Promise.all(fences.map(closeServer));
}
},
);
});
async function expectOutput(
@ -1161,6 +1583,129 @@ async function expectFailure(
expect(stderr).toContain(expected);
}
async function persistentInstallationFixture() {
const fixture = await installationFixture();
const command = [
"#!/usr/bin/env node",
'const fs = require("node:fs");',
"fs.writeFileSync(process.env.PAPERCLIP_PROVIDER_PID_FILE, String(process.pid));",
"setInterval(() => undefined, 1_000);",
].join("\n");
await writeFile(fixture.commandPath, command);
return {
...fixture,
command,
profile: {
...fixture.profile,
commandDigest: `sha256:${createHash("sha256").update(command).digest("hex")}`,
},
};
}
async function childMessage(
child: ChildProcess,
type: string,
): Promise<Record<string, unknown>> {
return await new Promise((resolve, reject) => {
const timer = setTimeout(
() => reject(new Error(`Timed out waiting for child message ${type}`)),
5_000,
);
const onExit = (code: number | null, signal: NodeJS.Signals | null) => {
clearTimeout(timer);
reject(new Error(`Child exited before ${type}: ${code ?? signal}`));
};
child.once("exit", onExit);
child.on("message", (message) => {
if (
typeof message !== "object" ||
message === null ||
(message as { type?: unknown }).type !== type
)
return;
clearTimeout(timer);
child.off("exit", onExit);
resolve(message as Record<string, unknown>);
});
});
}
async function waitForFile(path: string): Promise<string> {
let value = "";
await waitUntilAsync(async () => {
try {
value = await readFile(path, "utf8");
return value.length > 0;
} catch {
return false;
}
});
return value;
}
function processAlive(pid: number): boolean {
try {
process.kill(pid, 0);
return true;
} catch {
return false;
}
}
async function processStopped(pid: number): Promise<boolean> {
try {
const status = await readFile(`/proc/${pid}/status`, "utf8");
return /^State:\s+T/m.test(status);
} catch {
return false;
}
}
async function listenOnLoopback(port = 0): Promise<Server> {
const server = createServer();
await new Promise<void>((resolve, reject) => {
server.once("error", reject);
server.listen(
{ host: "127.0.0.1", port, exclusive: true, reusePort: false },
resolve,
);
});
return server;
}
async function closeServer(server: Server): Promise<void> {
if (!server.listening) return;
await new Promise<void>((resolve, reject) =>
server.close((error) => (error ? reject(error) : resolve())),
);
}
async function canBindLoopbackPort(port: number): Promise<boolean> {
try {
const server = await listenOnLoopback(port);
await closeServer(server);
return true;
} catch (error) {
if ((error as NodeJS.ErrnoException).code === "EADDRINUSE") return false;
throw error;
}
}
async function waitUntil(predicate: () => boolean): Promise<void> {
await waitUntilAsync(async () => predicate());
}
async function waitUntilAsync(
predicate: () => Promise<boolean>,
): Promise<void> {
const deadline = Date.now() + 5_000;
while (Date.now() < deadline) {
if (await predicate()) return;
await new Promise<void>((resolve) => setTimeout(resolve, 20));
}
throw new Error("Timed out waiting for subprocess state");
}
async function installationFixture() {
const root = await mkdtemp(join(tmpdir(), "paperclip-acpx-installation-"));
temporaryDirectories.push(root);
@ -1206,6 +1751,7 @@ async function installationFixture() {
runtimeDirectory,
serverPackageJsonPath,
runtimePackageJsonPath,
paths,
resolve(packageName: string): string {
const resolved = paths.get(packageName);
if (!resolved) throw new Error(`unexpected package ${packageName}`);

View File

@ -22,7 +22,7 @@ import {
relative,
resolve,
} from "node:path";
import type { Writable } from "node:stream";
import type { Readable, Writable } from "node:stream";
import type { QualifiedAcpxProfile } from "./qualified-profiles.js";
@ -32,6 +32,169 @@ const COMMAND_SOURCE_FD = 3;
const COMMAND_DIRECTORY_FD = 4;
const DEPENDENCY_ANCESTOR_FD_START = 5;
const MAX_DEPENDENCY_ANCESTORS = 64;
const PROVIDER_WATCHDOG_HANDSHAKE_TIMEOUT_MS = 2_000;
const PROVIDER_GUARDIAN_HANDSHAKE_TIMEOUT_MS = 5_000;
const PROVIDER_LIFETIME_WATCHDOG_SOURCE = `
const fs = require("node:fs");
let reaped = false;
const reap = () => {
if (reaped) return;
reaped = true;
try {
// Resolve the watchdog's current group at signal-delivery time. The live
// watchdog itself pins that identity until this atomic reap.
process.kill(0, "SIGKILL");
} catch {
try {
process.kill(process.pid, "SIGKILL");
} catch {
process.exit(1);
}
}
};
const owner = fs.createReadStream("", { fd: 3, autoClose: false });
owner.once("end", reap);
owner.once("error", reap);
owner.resume();
try {
fs.writeSync(4, "armed\\n");
} catch {
reap();
}
`;
export const PROVIDER_LIFETIME_GUARDIAN_SOURCE = `
const fs = require("node:fs");
const { spawn } = require("node:child_process");
const WATCHDOG_SOURCE = ${JSON.stringify(PROVIDER_LIFETIME_WATCHDOG_SOURCE)};
const dependencyAncestorCount = Number.parseInt(process.argv[4], 10);
if (!Number.isSafeInteger(dependencyAncestorCount) || dependencyAncestorCount < 0 || dependencyAncestorCount > ${MAX_DEPENDENCY_ANCESTORS}) throw new Error("ACPX provider dependency ancestry is invalid");
const OWNER_FD = ${DEPENDENCY_ANCESTOR_FD_START} + dependencyAncestorCount;
const OWNERSHIP_FD = OWNER_FD + 1;
const PROVIDER_EXIT_FD = OWNERSHIP_FD + 1;
const CREDENTIAL_FENCE_FD_START = PROVIDER_EXIT_FD + 1;
const dependencyAncestorFds = Array.from({ length: dependencyAncestorCount }, (_, index) => ${DEPENDENCY_ANCESTOR_FD_START} + index);
const PROVIDER_GUARDIAN_FD = ${DEPENDENCY_ANCESTOR_FD_START} + dependencyAncestorCount;
let provider;
let watchdog;
let reaped = false;
let shutdownStarted = false;
const reap = () => {
if (reaped) return;
reaped = true;
// This sentinel is the provider group's leader. It remains alive until this
// one atomic signal, pinning the numeric group identity against PID reuse.
process.kill(-process.pid, "SIGKILL");
};
const owner = fs.createReadStream("", { fd: OWNER_FD, autoClose: false });
owner.once("end", reap);
owner.once("error", reap);
owner.resume();
// Fail before provider code exists unless both inherited quorum fences are live.
fs.fstatSync(CREDENTIAL_FENCE_FD_START);
fs.fstatSync(CREDENTIAL_FENCE_FD_START + 1);
const shutdown = () => {
if (shutdownStarted || reaped) return;
shutdownStarted = true;
try {
provider?.kill("SIGTERM");
} catch {
reap();
return;
}
setTimeout(reap, 1_000);
};
process.on("SIGTERM", shutdown);
process.on("SIGINT", shutdown);
process.on("SIGHUP", shutdown);
const startProvider = () => {
if (provider || reaped || shutdownStarted) return;
try {
provider = spawn(
process.execPath,
["--eval", process.argv[1], ...process.argv.slice(2)],
{
cwd: process.cwd(),
detached: false,
env: process.env,
shell: false,
// The provider observes this guardian-owned pipe directly. Kernel EOF
// therefore revokes it even when SIGKILL/OOM prevents our JS reap path.
// It also inherits both quorum fences until that self-reap completes.
stdio: [0, 1, 2, ${COMMAND_SOURCE_FD}, ${COMMAND_DIRECTORY_FD}, ...dependencyAncestorFds, "pipe", PROVIDER_EXIT_FD, CREDENTIAL_FENCE_FD_START, CREDENTIAL_FENCE_FD_START + 1],
windowsHide: true,
},
);
provider.once("error", reap);
provider.once("exit", reap);
provider.once("spawn", () => {
try {
if (reaped || shutdownStarted) {
reap();
return;
}
// The provider now owns the only child-side copy. Parent-side EOF is an
// independent kernel observation of provider exit even if this guardian
// is killed before it can reap the group.
fs.closeSync(PROVIDER_EXIT_FD);
fs.writeSync(OWNERSHIP_FD, "owned\\n");
} catch {
reap();
}
});
} catch {
reap();
}
};
try {
// A credential-free peer in this same process group reaps the group through
// its live identity if this guardian is killed before it can run its reap.
// Its private owner pipe reaches kernel EOF on guardian death even while the
// provider is stopped and unable to process its own guardian-loss callback.
watchdog = spawn(process.execPath, ["--eval", WATCHDOG_SOURCE], {
cwd: process.cwd(),
detached: false,
env: {},
shell: false,
stdio: ["ignore", "ignore", "ignore", "pipe", "pipe"],
windowsHide: true,
});
const watchdogOwnerPipe = watchdog.stdio[3];
const watchdogReady = watchdog.stdio[4];
if (watchdogOwnerPipe == null) throw new Error("ACPX provider lifetime watchdog omitted its owner pipe");
if (watchdogReady == null) throw new Error("ACPX provider lifetime watchdog omitted its readiness pipe");
watchdogOwnerPipe.once("error", reap);
watchdog.once("error", reap);
watchdog.once("exit", reap);
let watchdogOutput = "";
let watchdogArmed = false;
const watchdogReadyTimeout = setTimeout(reap, ${PROVIDER_WATCHDOG_HANDSHAKE_TIMEOUT_MS});
watchdogReadyTimeout.unref();
const rejectUnarmedWatchdog = () => {
if (!watchdogArmed) reap();
};
watchdogReady.once("error", rejectUnarmedWatchdog);
watchdogReady.once("close", rejectUnarmedWatchdog);
watchdogReady.on("data", (chunk) => {
watchdogOutput += chunk.toString();
if (watchdogOutput.length > 64) {
reap();
return;
}
if (!watchdogOutput.includes("armed\\n")) return;
watchdogArmed = true;
clearTimeout(watchdogReadyTimeout);
watchdogReady.removeAllListeners("data");
startProvider();
});
} catch {
reap();
}
`;
const providerGuardianOwnership = new WeakMap<ChildProcess, Promise<void>>();
const providerExitProof = new WeakMap<ChildProcess, Promise<void>>();
export type AcpxPackageJsonResolver = (packageName: string) => string;
@ -51,10 +214,69 @@ export interface VerifiedAcpxCommandLease {
spawn(
args?: readonly string[],
options?: SpawnOptionsWithoutStdio,
lifetime?: VerifiedAcpxProviderLifetime,
): ChildProcess;
close(): Promise<void>;
}
export interface VerifiedAcpxProviderLifetime {
/** Two listening sockets that fence the canonical Codex credential home. */
credentialFenceFds: readonly [number, number];
/** Validate the guardian before provider admission can succeed. */
activateCredentialFenceOwner(pid: number): Promise<void>;
}
/** Fail closed where verified provider-group ownership cannot be guaranteed. */
export function assertVerifiedAcpxProviderPlatform(
platform: NodeJS.Platform,
): void {
if (platform === "win32") {
throw new Error(
"The production ACPX runtime is unavailable on Windows because verified provider launch requires atomic no-follow file opening",
);
}
}
/** Reap only the group the live provider belongs to at signal-delivery time. */
export function reapCurrentProviderProcessGroup(
kill: (pid: number, signal: NodeJS.Signals) => unknown,
currentPid: number,
exit: (code: number) => unknown,
): void {
try {
// POSIX pid zero names the caller's current process group. Unlike a saved
// guardian PGID, the kernel resolves this ownership at the instant of the
// signal, so a dead guardian's recycled identifier can never be targeted.
kill(0, "SIGKILL");
} catch {
try {
// The caller's own live PID cannot be recycled out from under it. This
// fallback still revokes the provider if whole-group signaling fails.
kill(currentPid, "SIGKILL");
} catch {
exit(1);
}
}
}
/** Wait until the verified wrapper has armed owner-death and credential fencing. */
export async function awaitVerifiedAcpxProviderOwnership(
child: ChildProcess,
): Promise<void> {
await (providerGuardianOwnership.get(child) ?? Promise.resolve());
}
/** Wait for kernel EOF on the descriptor held only by the provider process. */
export async function awaitVerifiedAcpxProviderExit(
child: ChildProcess,
): Promise<void> {
const exitProof = providerExitProof.get(child);
if (!exitProof) {
throw new Error("ACPX provider exit proof is unavailable");
}
await exitProof;
}
interface VerifiedAcpxCommandIdentity {
device: string;
inode: string;
@ -77,6 +299,8 @@ type AcpxCommandFormat = "commonjs" | "module";
const COMMONJS_SNAPSHOT_BOOTSTRAP = snapshotBootstrap("commonjs");
const MODULE_SNAPSHOT_BOOTSTRAP = snapshotBootstrap("module");
const GUARDED_COMMONJS_SNAPSHOT_BOOTSTRAP = snapshotBootstrap("commonjs", true);
const GUARDED_MODULE_SNAPSHOT_BOOTSTRAP = snapshotBootstrap("module", true);
/** Resolve and verify every installed artifact bound by a qualified profile. */
export async function verifyQualifiedAcpxInstallation(
@ -564,43 +788,124 @@ function commandLease(
spawn(
args: readonly string[] = [],
options: SpawnOptionsWithoutStdio = {},
lifetime?: VerifiedAcpxProviderLifetime,
): ChildProcess {
if (consumed) throw new Error("Verified ACPX command lease is closed");
consumed = true;
let child: ChildProcess;
try {
const guarded = lifetime !== undefined;
if (guarded) assertVerifiedAcpxProviderPlatform(process.platform);
const providerBootstrap = guarded
? format === "module"
? GUARDED_MODULE_SNAPSHOT_BOOTSTRAP
: GUARDED_COMMONJS_SNAPSHOT_BOOTSTRAP
: format === "module"
? MODULE_SNAPSHOT_BOOTSTRAP
: COMMONJS_SNAPSHOT_BOOTSTRAP;
const providerOwnershipFd =
DEPENDENCY_ANCESTOR_FD_START + dependencyAncestors.length + 1;
const providerExitFd = providerOwnershipFd + 1;
if (
guarded &&
(!Array.isArray(lifetime.credentialFenceFds) ||
lifetime.credentialFenceFds.length !== 2 ||
lifetime.credentialFenceFds.some(
(fd) => !Number.isSafeInteger(fd) || fd < 0,
) ||
lifetime.credentialFenceFds[0] === lifetime.credentialFenceFds[1] ||
typeof lifetime.activateCredentialFenceOwner !== "function")
) {
throw new Error("ACPX provider credential fence is invalid");
}
child = spawnChildProcess(
process.execPath,
[
// Keep resolved module URLs on the retained descriptor paths so
// the hook can distinguish them from ordinary host ancestry.
"--preserve-symlinks",
"--eval",
format === "module"
? MODULE_SNAPSHOT_BOOTSTRAP
: COMMONJS_SNAPSHOT_BOOTSTRAP,
commandDirectoryPath,
commandName,
String(dependencyAncestors.length),
String(serverDependencyAncestorCount),
serverPackageFormat,
JSON.stringify(dependencyAncestorFormats),
...args,
],
guarded
? [
// Keep resolved module URLs on the retained descriptor paths
// so the hook can distinguish them from host ancestry.
"--preserve-symlinks",
"--eval",
PROVIDER_LIFETIME_GUARDIAN_SOURCE,
providerBootstrap,
commandDirectoryPath,
commandName,
String(dependencyAncestors.length),
String(serverDependencyAncestorCount),
serverPackageFormat,
JSON.stringify(dependencyAncestorFormats),
...args,
]
: [
"--preserve-symlinks",
"--eval",
providerBootstrap,
commandDirectoryPath,
commandName,
String(dependencyAncestors.length),
String(serverDependencyAncestorCount),
serverPackageFormat,
JSON.stringify(dependencyAncestorFormats),
...args,
],
{
...options,
// In production this process is a persistent sentinel and group
// leader. It arms owner-death before spawning provider code, keeps
// both credential quorum listeners inherited, and pins the PGID
// until its single whole-group reap.
detached: process.platform !== "win32",
env: sanitizedNodeEnvironment(options.env),
shell: false,
stdio: [
"pipe",
"pipe",
"pipe",
"pipe",
commandDirectory.fd,
...dependencyAncestors.map((handle) => handle.fd),
],
stdio: guarded
? [
"pipe",
"pipe",
"pipe",
"pipe",
commandDirectory.fd,
...dependencyAncestors.map((handle) => handle.fd),
"pipe",
"pipe",
"pipe",
...lifetime.credentialFenceFds,
]
: [
"pipe",
"pipe",
"pipe",
"pipe",
commandDirectory.fd,
...dependencyAncestors.map((handle) => handle.fd),
],
},
);
if (guarded) {
const guardianOwnerPipe = child.stdio[
providerOwnershipFd - 1
] as Writable | null;
if (guardianOwnerPipe === null) {
throw new Error(
"ACPX provider lifetime guardian omitted its owner pipe",
);
}
protectProviderGroupKill(child, guardianOwnerPipe);
const exitProof = providerExitHandshake(child, providerExitFd);
void exitProof.catch(() => undefined);
providerExitProof.set(child, exitProof);
const guardianPid = child.pid!;
const ownership = Promise.all([
providerOwnershipHandshake(child, providerOwnershipFd),
Promise.resolve().then(() =>
lifetime.activateCredentialFenceOwner(guardianPid),
),
]).then(() => undefined);
// Session construction can reject before the adapter reaches its
// explicit ownership await. Observe that early rejection now while
// preserving it for the admission boundary.
void ownership.catch(() => undefined);
providerGuardianOwnership.set(child, ownership);
}
} catch (error) {
verifiedBytes.fill(0);
releaseDirectoriesBestEffort();
@ -624,6 +929,133 @@ function commandLease(
};
}
function protectProviderGroupKill(
child: ChildProcess,
guardianOwnerPipe: Writable,
): void {
const signalGuardian = child.kill.bind(child);
let groupReaped = false;
let revocationStarted = false;
child.once("exit", () => {
groupReaped = true;
});
child.kill = (signal?: NodeJS.Signals | number): boolean => {
if (signal !== "SIGKILL" && signal !== 9) {
return signalGuardian(signal);
}
if (groupReaped) return false;
if (!revocationStarted) {
revocationStarted = true;
// Revocation closes the retained parent-to-guardian owner pipe. Resume
// the exact direct child as well: SIGCONT is harmless for a running
// guardian and lets a stopped guardian observe EOF and reap its own
// still-pinned group. Do not mark the group reaped until exit is seen.
guardianOwnerPipe.destroy();
}
try {
// Every retry wakes the exact live guardian so it can observe the owner
// pipe EOF and reap the whole group itself. Never SIGKILL the guardian:
// a stopped real provider could otherwise survive after the adapter
// forgets the only process that still pins its group identity.
signalGuardian("SIGCONT");
} catch {
// The pipe close remains the primary revocation operation. Retained
// cleanup keeps waiting for observed guardian exit and may retry wakeup.
}
return true;
};
}
function providerOwnershipHandshake(
child: ChildProcess,
ownershipFd: number,
): Promise<void> {
const output = (child.stdio as Array<Readable | Writable | null | undefined>)[
ownershipFd
] as Readable | null | undefined;
if (output == null) {
return Promise.reject(
new Error("ACPX provider lifetime guardian omitted its ownership pipe"),
);
}
return new Promise<void>((resolve, reject) => {
let settled = false;
let buffered = "";
const finish = (error?: Error): void => {
if (settled) return;
settled = true;
clearTimeout(timer);
child.off("error", onError);
child.off("close", onClose);
output.off("data", onData);
if (error) reject(error);
else resolve();
};
const onError = (): void =>
finish(new Error("ACPX provider lifetime guardian failed to start"));
const onClose = (): void =>
finish(
new Error(
"ACPX provider lifetime guardian exited before ownership transfer",
),
);
const onData = (chunk: Buffer | string): void => {
buffered += chunk.toString();
if (buffered.includes("owned\n")) finish();
};
const timer = setTimeout(
() =>
finish(
new Error("ACPX provider lifetime guardian ownership timed out"),
),
PROVIDER_GUARDIAN_HANDSHAKE_TIMEOUT_MS,
);
timer.unref();
child.once("error", onError);
child.once("close", onClose);
output.on("data", onData);
});
}
function providerExitHandshake(
child: ChildProcess,
providerExitFd: number,
): Promise<void> {
const output = (child.stdio as Array<Readable | Writable | null | undefined>)[
providerExitFd
] as Readable | null | undefined;
if (output == null) {
return Promise.reject(
new Error("ACPX provider lifetime proof pipe was not created"),
);
}
return new Promise<void>((resolve, reject) => {
let settled = false;
const finish = (error?: Error): void => {
if (settled) return;
settled = true;
output.off("end", onEnd);
output.off("close", onClose);
output.off("error", onError);
if (error) reject(error);
else resolve();
};
const onEnd = (): void => finish();
const onClose = (): void =>
finish(
output.readableEnded
? undefined
: new Error("ACPX provider lifetime proof pipe closed before EOF"),
);
const onError = (): void =>
finish(new Error("ACPX provider lifetime proof pipe failed"));
output.once("end", onEnd);
output.once("close", onClose);
output.once("error", onError);
output.resume();
});
}
export function sanitizedNodeEnvironment(
environment: NodeJS.ProcessEnv | undefined,
): NodeJS.ProcessEnv {
@ -650,7 +1082,7 @@ export function sanitizedNodeEnvironment(
return sanitized;
}
function snapshotBootstrap(format: AcpxCommandFormat): string {
function snapshotBootstrap(format: AcpxCommandFormat, guarded = false): string {
return [
'const fs = require("node:fs");',
'const { isBuiltin, registerHooks } = require("node:module");',
@ -666,6 +1098,24 @@ function snapshotBootstrap(format: AcpxCommandFormat): string {
`if (!Number.isSafeInteger(dependencyAncestorCount) || dependencyAncestorCount < 0 || dependencyAncestorCount > ${MAX_DEPENDENCY_ANCESTORS}) throw new Error("ACPX provider dependency ancestry is invalid");`,
'if (!Number.isSafeInteger(serverDependencyAncestorCount) || serverDependencyAncestorCount < 0 || serverDependencyAncestorCount > dependencyAncestorCount) throw new Error("ACPX provider package ancestry is invalid");',
'if ((serverPackageFormat !== "module" && serverPackageFormat !== "commonjs") || !Array.isArray(dependencyAncestorFormats) || dependencyAncestorFormats.length !== dependencyAncestorCount || dependencyAncestorFormats.some((value) => value !== "module" && value !== "commonjs")) throw new Error("ACPX provider package formats are invalid");',
...(guarded
? [
`const guardianFd = ${DEPENDENCY_ANCESTOR_FD_START} + dependencyAncestorCount;`,
'const guardian = fs.createReadStream("", { fd: guardianFd, autoClose: false });',
`const reapCurrentProviderProcessGroup = ${reapCurrentProviderProcessGroup.toString()};`,
"const killProviderProcess = process.kill.bind(process);",
"const providerProcessId = process.pid;",
"const exitProviderProcess = process.exit.bind(process);",
"let guardianLost = false;",
"const reapOnGuardianLoss = () => { if (guardianLost) return; guardianLost = true; reapCurrentProviderProcessGroup(killProviderProcess, providerProcessId, exitProviderProcess); };",
'guardian.once("end", reapOnGuardianLoss);',
'guardian.once("error", reapOnGuardianLoss);',
"guardian.resume();",
"fs.fstatSync(guardianFd + 1);",
"fs.fstatSync(guardianFd + 2);",
"fs.fstatSync(guardianFd + 3);",
]
: []),
"const commandPath = resolve(commandDirectory, commandName);",
`const guardSnapshotModuleLookup = ${guardSnapshotModuleLookup.toString()};`,
`const directory = process.platform === "linux" ? "/proc/self/fd/${COMMAND_DIRECTORY_FD}" : commandDirectory;`,

View File

@ -413,7 +413,6 @@ describe("ACPX runtime host", () => {
.mockRejectedValueOnce(new Error("first admission cleanup failed"))
.mockImplementationOnce(() => retryClose),
});
await expect(
AcpxRuntimeHost.open(
{
@ -464,6 +463,79 @@ describe("ACPX runtime host", () => {
await contender.close();
});
it("bounds post-handshake model verification and cleans the runtime", async () => {
const fixture = await hostFixture();
const runtime = runtimePort({
getStatus: () => new Promise<never>(() => undefined),
});
const dependencies = fixture.dependencies({
openRuntime: async () => runtime,
});
dependencies.admissionVerificationTimeoutMs = 1;
await expect(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "approve-all",
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
},
dependencies,
),
).rejects.toThrow("admission verification exceeded its deadline");
expect(runtime.close).toHaveBeenCalledOnce();
expect(fixture.commandClose).toHaveBeenCalledOnce();
});
it("bounds post-handshake cleanup while retaining its exact owner", async () => {
const fixture = await hostFixture();
let finishRuntimeClose!: () => void;
const runtimeClose = new Promise<void>((resolve) => {
finishRuntimeClose = resolve;
});
const runtime = runtimePort({
getStatus: () => new Promise<never>(() => undefined),
onClose: () => runtimeClose,
});
const dependencies = fixture.dependencies({
openRuntime: async () => runtime,
});
let retainedAdmissionCleanup: Promise<void> | null = null;
dependencies.retainAdmissionCleanup = (cleanup) => {
retainedAdmissionCleanup = cleanup;
};
dependencies.admissionVerificationTimeoutMs = 1;
dependencies.admissionCleanupTimeoutMs = 1;
await expect(
AcpxRuntimeHost.open(
{
...fixture.options,
agent: "codex",
model: "gpt-5.6-sol",
permissionMode: "approve-all",
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
},
dependencies,
),
).rejects.toThrow("initialization and cleanup failed");
expect(runtime.close).toHaveBeenCalledOnce();
expect(fixture.commandClose).toHaveBeenCalledOnce();
expect(retainedAdmissionCleanup).not.toBeNull();
let cleanupSettled = false;
void retainedAdmissionCleanup!.finally(() => {
cleanupSettled = true;
});
await new Promise<void>((resolve) => setTimeout(resolve, 1));
expect(cleanupSettled).toBe(false);
finishRuntimeClose();
await retainedAdmissionCleanup;
expect(cleanupSettled).toBe(true);
});
it("retains credential ownership when runtime shutdown fails until retry succeeds", async () => {
const fixture = await hostFixture();
let failClose = true;
@ -871,6 +943,8 @@ describe("ACPX runtime host", () => {
const credentialAdmission = deferred<{
path: string;
mode: "inline_json";
lifetimeFenceFds: readonly [number, number];
activateLifetimeOwner(pid: number): Promise<void>;
close(): Promise<void>;
}>();
const cleanupFailure = new Error("transient credential cleanup failure");
@ -908,6 +982,8 @@ describe("ACPX runtime host", () => {
credentialAdmission.resolve({
path: lateCredentialPath,
mode: "inline_json",
lifetimeFenceFds: [42, 43],
activateLifetimeOwner: async () => undefined,
close: lateCredentialClose,
});
@ -1043,6 +1119,8 @@ describe("ACPX runtime host", () => {
stageCredential: async () => ({
path: join(fixture.root, "auth.json"),
mode: "inline_json",
lifetimeFenceFds: [42, 43],
activateLifetimeOwner: async () => undefined,
close: credentialClose,
}),
},

View File

@ -43,6 +43,15 @@ import {
import type { AcpxExpectedSessionIdentity } from "./sidecar-protocol.js";
export const ACPX_TURN_CANCELLATION_SHUTDOWN_BOUND_MS = 2_000;
const RUNTIME_ADMISSION_VERIFICATION_TIMEOUT_MS = 8_000;
const activeRuntimeHostCleanupOwners = new Set<Promise<unknown>>();
class AcpxRuntimeAdmissionTimeoutError extends Error {
constructor() {
super("ACPX runtime admission verification exceeded its deadline");
this.name = "AcpxRuntimeAdmissionTimeoutError";
}
}
const ACPX_ADMISSION_CLEANUP_BATCH_ATTEMPTS = 8;
const ACPX_ADMISSION_CLEANUP_RETRY_DELAY_MS = 10;
@ -88,6 +97,10 @@ export interface AcpxRuntimePortOpenOptions {
permissionMode: NativeAcpxPermissionMode;
permissionPolicy: ReturnType<typeof acpxRuntimePermissionPolicy>;
launchEnvironment: Readonly<NodeJS.ProcessEnv>;
/** Kernel credential-home quorum inherited by the provider sentinel. */
credentialFenceFds: readonly [number, number] | null;
/** Validate the guardian while the credential-home quorum is held. */
activateCredentialFenceOwner: ((pid: number) => Promise<void>) | null;
systemInstructions: string;
/** Revalidate a pinned recovery workspace at the provider spawn boundary. */
assertWorkspaceHeld?: () => void;
@ -101,12 +114,6 @@ export interface AcpxRuntimePortOpenOptions {
retainFailedAdmissionCleanup(cleanup: Promise<void>): void;
}
export interface AcpxRetainedCleanupFailure {
resource: "credential" | "command" | "runtime" | "tool_bridge";
attempt: number;
error: unknown;
}
export interface AcpxMcpServerBinding {
name: string;
url: string;
@ -116,6 +123,12 @@ export interface AcpxMcpServerBinding {
export type AcpxSemanticToolSession = Omit<RunnerToolBridgeOptions, "secret">;
export interface AcpxRetainedCleanupFailure {
resource: "credential" | "command" | "runtime" | "tool_bridge";
attempt: number;
error: unknown;
}
export interface AcpxRuntimeHostDependencies {
verifyInstallation?: (
profile: QualifiedAcpxProfile,
@ -123,6 +136,16 @@ export interface AcpxRuntimeHostDependencies {
/** Internal test seam for aborting credential acquisition. */
stageCredential?: typeof stageManagedCodexCredential;
openRuntime(options: AcpxRuntimePortOpenOptions): Promise<AcpxRuntimePort>;
/** Internal test seam for the post-handshake admission deadline. */
admissionVerificationTimeoutMs?: number;
/** Internal test seam for failed-admission cleanup. */
admissionCleanupTimeoutMs?: number;
/**
* Transfers failed-admission cleanup ownership to the embedding lifecycle.
* The callback receives the exact aggregate cleanup attempt before the
* bounded admission wait can return.
*/
retainAdmissionCleanup?: (cleanup: Promise<void>) => void;
/**
* Required observability channel for resources acquired after admission was
* aborted. Implementations must not throw from this callback.
@ -148,7 +171,6 @@ export interface OpenAcpxRuntimeHostOptions {
semanticTools?: AcpxSemanticToolSession;
}
const activeRuntimeHostCleanupOwners = new Set<Promise<unknown>>();
const RETAINED_CLEANUP_RETRY_INITIAL_DELAY_MS = 10;
const RETAINED_CLEANUP_RETRY_MAX_DELAY_MS = 1_000;
@ -254,6 +276,12 @@ export class AcpxRuntimeHost {
let toolBridge: RunnerToolBridge | null = null;
let runtime: AcpxRuntimePort | null = null;
let pendingRuntimeOwnsCredential = false;
const admissionVerificationTimeoutMs =
dependencies.admissionVerificationTimeoutMs ??
RUNTIME_ADMISSION_VERIFICATION_TIMEOUT_MS;
const admissionCleanupTimeoutMs =
dependencies.admissionCleanupTimeoutMs ??
RUNTIME_ADMISSION_VERIFICATION_TIMEOUT_MS;
let failedAdmissionCleanupTransferred = false;
let resolveFailedAdmissionCleanupTransfer!: () => void;
const failedAdmissionCleanupTransfer = new Promise<void>((resolve) => {
@ -340,6 +368,11 @@ export class AcpxRuntimeHost {
binding.permissionMode,
),
launchEnvironment: sandbox.launchEnvironment,
credentialFenceFds: credential?.lifetimeFenceFds ?? null,
activateCredentialFenceOwner:
typeof credential?.activateLifetimeOwner === "function"
? credential.activateLifetimeOwner.bind(credential)
: null,
systemInstructions: boundedInstructions(options.systemInstructions),
...(options.assertWorkspaceHeld === undefined
? {}
@ -379,12 +412,14 @@ export class AcpxRuntimeHost {
});
},
});
await runAbortableAdmissionStage(options.signal, () =>
requireVerifiedAcpxModel(runtime!, profile),
);
const runtimeIdentity = await runAbortableAdmissionStage(
options.signal,
() => runtime!.identity(),
() =>
boundedRuntimeAdmissionVerification(
runtime!,
profile,
admissionVerificationTimeoutMs,
),
);
const observedIdentity: AcpxExpectedSessionIdentity = {
kind: "acpx",
@ -411,14 +446,16 @@ export class AcpxRuntimeHost {
toolBridge,
});
} catch (error) {
const cleanupError = await cleanupRuntimeResources(
const cleanup = cleanupRuntimeResources(
runtime,
toolBridge,
pendingRuntimeOwnsCredential ? null : credential,
command,
"ACPX runtime initialization failed",
);
if (cleanupError) {
retainRuntimeHostCleanup(cleanup);
void cleanup.then((cleanupError) => {
if (!cleanupError) return;
retainFailedAcpxAdmissionCleanup({
runtime,
toolBridge,
@ -426,8 +463,30 @@ export class AcpxRuntimeHost {
command,
reason: "ACPX runtime initialization failed",
});
});
dependencies.retainAdmissionCleanup?.(
cleanup.then((cleanupError) => {
if (cleanupError) throw cleanupError;
}),
);
const cleanupOutcome = await awaitRuntimeHostCleanupWithin(
cleanup,
admissionCleanupTimeoutMs,
);
if (cleanupOutcome === "deferred") {
throw new AggregateError(
[error, ...cleanupError.errors],
[
error,
new Error(
"ACPX runtime initialization cleanup exceeded its shutdown timeout",
),
],
"ACPX runtime initialization and cleanup failed",
);
}
if (cleanupOutcome) {
throw new AggregateError(
[error, ...cleanupOutcome.errors],
"ACPX runtime initialization and cleanup failed",
);
}
@ -468,9 +527,7 @@ export class AcpxRuntimeHost {
text,
requestId,
...(input.signal ? { signal: input.signal } : {}),
...(input.onElicitation
? { onElicitation: input.onElicitation }
: {}),
...(input.onElicitation ? { onElicitation: input.onElicitation } : {}),
});
this.#activeTurn = turn;
void turn.result
@ -531,11 +588,15 @@ export class AcpxRuntimeHost {
this.#command,
reason,
);
if (cleanupError) errors.push(...cleanupError.errors);
if (!cleanupError) {
// Runtime, credential, and command ownership has been relinquished even
// when the provider never acknowledged turn cancellation. Preserve that
// cancellation error for this caller, but make later close calls
// idempotently observe the successfully closed host.
if (this.#activeTurn === activeTurn) this.#activeTurn = null;
this.#closed = true;
}
if (cleanupError) errors.push(...cleanupError.errors);
if (errors.length > 0) {
throw new AggregateError(errors, "ACPX runtime cleanup failed");
}
@ -557,8 +618,8 @@ async function acquireAbortableAdmissionResource<T>(input: {
acquire: () => Promise<T>;
resource: AcpxRetainedCleanupFailure["resource"];
releaseLate: (resource: T) => Promise<void>;
reportFailure: (failure: AcpxRetainedCleanupFailure) => void;
onAbortedPending?: (pending: Promise<T>) => void;
reportFailure: (failure: AcpxRetainedCleanupFailure) => void;
}): Promise<T> {
if (input.signal === undefined) return await input.acquire();
input.signal.throwIfAborted();
@ -611,53 +672,6 @@ function raceAdmissionWithAbort<T>(
});
}
async function releaseRetainedAdmissionResource<T>(input: {
resource: T;
resourceKind: AcpxRetainedCleanupFailure["resource"];
release: (resource: T) => Promise<void>;
reportFailure: (failure: AcpxRetainedCleanupFailure) => void;
}): Promise<void> {
let attempt = 0;
let retryDelayMs = RETAINED_CLEANUP_RETRY_INITIAL_DELAY_MS;
for (;;) {
attempt += 1;
try {
await input.release(input.resource);
return;
} catch (error) {
try {
input.reportFailure({
resource: input.resourceKind,
attempt,
error,
});
} catch {
// The required reporter is observational. A broken reporter must not
// relinquish ownership of the resource that still needs cleanup.
}
await waitForRetainedCleanupRetry(retryDelayMs);
retryDelayMs = Math.min(
retryDelayMs * 2,
RETAINED_CLEANUP_RETRY_MAX_DELAY_MS,
);
}
}
}
async function waitForRetainedCleanupRetry(delayMs: number): Promise<void> {
await new Promise<void>((resolve) => {
const timer = setTimeout(resolve, delayMs);
timer.unref?.();
});
}
function retainRuntimeHostCleanup(cleanup: Promise<unknown>): void {
activeRuntimeHostCleanupOwners.add(cleanup);
void cleanup
.finally(() => activeRuntimeHostCleanupOwners.delete(cleanup))
.catch(() => undefined);
}
function retainAbortedRuntimeAdmissionCleanup(input: {
pendingRuntime: Promise<AcpxRuntimePort>;
credential: ManagedCodexCredentialLease | null;
@ -775,6 +789,101 @@ async function waitForAdmissionCleanupRetry(delayMs: number): Promise<void> {
});
}
async function releaseRetainedAdmissionResource<T>(input: {
resource: T;
resourceKind: AcpxRetainedCleanupFailure["resource"];
release: (resource: T) => Promise<void>;
reportFailure: (failure: AcpxRetainedCleanupFailure) => void;
}): Promise<void> {
let attempt = 0;
let retryDelayMs = RETAINED_CLEANUP_RETRY_INITIAL_DELAY_MS;
for (;;) {
attempt += 1;
try {
await input.release(input.resource);
return;
} catch (error) {
try {
input.reportFailure({
resource: input.resourceKind,
attempt,
error,
});
} catch {
// The required reporter is observational. A broken reporter must not
// relinquish ownership of the resource that still needs cleanup.
}
await waitForRetainedCleanupRetry(retryDelayMs);
retryDelayMs = Math.min(
retryDelayMs * 2,
RETAINED_CLEANUP_RETRY_MAX_DELAY_MS,
);
}
}
}
async function waitForRetainedCleanupRetry(delayMs: number): Promise<void> {
await new Promise<void>((resolve) => {
const timer = setTimeout(resolve, delayMs);
timer.unref?.();
});
}
function retainRuntimeHostCleanup(cleanup: Promise<unknown>): void {
activeRuntimeHostCleanupOwners.add(cleanup);
void cleanup.then(
() => activeRuntimeHostCleanupOwners.delete(cleanup),
() => activeRuntimeHostCleanupOwners.delete(cleanup),
);
}
async function awaitRuntimeHostCleanupWithin(
cleanup: Promise<AggregateError | null>,
timeoutMs: number,
): Promise<AggregateError | null | "deferred"> {
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
cleanup,
new Promise<"deferred">((resolve) => {
timer = setTimeout(
() => resolve("deferred"),
Math.max(1, Math.floor(timeoutMs)),
);
timer.unref();
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
async function boundedRuntimeAdmissionVerification(
runtime: AcpxRuntimePort,
profile: QualifiedAcpxProfile,
timeoutMs: number,
): Promise<AcpxRuntimePortIdentity> {
const verification = Promise.resolve().then(async () => {
await requireVerifiedAcpxModel(runtime, profile);
return await runtime.identity();
});
let timer: ReturnType<typeof setTimeout> | undefined;
try {
return await Promise.race([
verification,
new Promise<never>((_resolve, reject) => {
timer = setTimeout(
() => reject(new AcpxRuntimeAdmissionTimeoutError()),
timeoutMs,
);
timer.unref();
}),
]);
} finally {
if (timer) clearTimeout(timer);
}
}
async function boundedCancellation(
cancellation: Promise<void>,
): Promise<unknown | null> {
@ -794,6 +903,7 @@ async function boundedCancellation(
}),
ACPX_TURN_CANCELLATION_SHUTDOWN_BOUND_MS,
);
timer.unref();
}),
]);
if (timer) clearTimeout(timer);

View File

@ -37,6 +37,12 @@ test("the runner pins only the Codex ACPX production dependencies", () => {
);
});
test("the package exposes only the reviewed Codex ACPX sidecar binary", () => {
assert.deepEqual(runnerPackage.bin, {
"paperclip-runner-acpx-sidecar": "./dist/cli/acpx-runtime-sidecar.js",
});
});
test("old and new pnpm configuration both apply the exact runtime patches", () => {
assert.equal(
rootPackage.pnpm.patchedDependencies["acpx@0.13.1"],