feat(runner): add Codex ACPX harness driver (#12405)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The runner package hides provider behavior behind the `HarnessDriver` contract. > - The admitted Codex ACPX runtime can open sessions, execute bounded turns, and use run-scoped semantic tools. > - The package still needs a driver that translates those turns into canonical PRP events and semantic results. > - The driver must preserve terminal facts under backpressure and retain cleanup ownership after caller-facing timeouts. > - Abort can win after the host transfers a credential-bearing admission but before the adapter body starts, so that boundary must publish a completed cleanup proof without changing the exact cancellation reason. > - An external close can join an autonomous reconciliation attempt; if it joins the exhausted final attempt, its batched intent must create one new bounded generation when that exact cleanup fails. > - A late cleanup failure needs a finite reconciliation budget that cannot renew itself without a distinct external intent. > - This pull request adds the Codex-only ACPX harness driver and the package-local lifecycle rules it needs. > - The benefit is a tested provider-neutral session boundary for later runnerd and server integration. ## Linked Issues or Issue Description Refs #12404 **Subsystem affected** This change affects `packages/paperclip-runner`, the Codex ACPX driver, and the provider-neutral native session runtime. **Problem or motivation** The package has an admitted Codex ACPX session, bounded turn control, and authenticated semantic tools. It does not have a `HarnessDriver` implementation that joins those parts and emits canonical PRP events. It also needs bounded ownership for provider cleanup that settles after a caller-facing timeout. Cancellation can win after the host schedules runtime admission and transfers the staged credential but before the adapter body starts; that rejected admission must still prove that provider cleanup is complete so the credential can be scrubbed and later admission can proceed. Separately, an external close that coalesces onto an exhausted autonomous reconciliation must not lose its cleanup intent if that exact protocol or provider-process cleanup fails. **Proposed solution** Add a Codex-only harness driver. It opens the admitted host, executes one active turn, normalizes ACPX events, dispatches run-scoped tools, and commits one schema-valid completion or blocked result. It provides bounded event storage, interruption, transcripts, usage, snapshots, diagnostics, and ordered close behavior. The native session runtime quarantines incomplete cleanup before another session can enter the same cleanup domain. The ACPX adapter records the immutable origin and attempt number of each exact close attempt. Autonomous reconciliation failures stay inside the three-attempt budget of the generation that created them. External callers that join an attempt are represented by one idempotent batched intent: success consumes it, failure on an earlier attempt uses the remaining same-generation retries, and failure on the exhausted final attempt creates exactly one new bounded generation. Both direct and late protocol/provider cleanup outcomes use the same rule. At the adapter entry boundary, an already-aborted admission transfers an already-complete cleanup proof before rethrowing the exact abort reason; the host retains credential cleanup until that proof settles. **Alternatives considered** The multi-provider integration driver was not copied because it mixes deferred providers and recovery behavior into the Codex path. Direct server registration was also deferred because this package slice must remain inactive and independently safe. Relabeling a coalesced autonomous attempt as external was rejected because it would let observers replenish retry budgets; starting another protocol close before the exact retained attempt settles was rejected because it would overlap cleanup ownership. **Roadmap alignment** This is package-local production hardening for the experimental runner. It does not enable a new adapter or change current agent execution selection. ## What Changed - Add a Codex-only ACPX `HarnessDriver` and session implementation. - Advertise only implemented capabilities. Keep resume, steering, runtime request resolution, runtime request handoff, goals, and thread lineage unavailable. - Emit canonical PRP turn, transcript, tool execution, final reply, result, failure, interruption, and usage facts. - Dispatch authorized dynamic tools through the authenticated semantic bridge. - Validate and commit one completion or blocked result with disposition and conflict checks. - Add stable bounded event identities, one-active-turn admission, terminal capacity reservation, and bounded interruption. - Redact authorization credentials from emitted events and retained transcripts. - Add read, reconcile, transcript, usage, snapshot, status, interruption, and ordered close surfaces. - Retain and quarantine host cleanup that outlives a caller-facing close bound. - Gate new native-session admission on prior cleanup in the same cleanup domain. - Preserve durable success and governed waits while provider cleanup continues under bounded ownership. - Transfer a completed cleanup proof when cancellation wins before the Codex adapter body, then preserve the caller's exact abort reason. - Add a host-level regression proving staged credentials are scrubbed, the credential lease can be reacquired, and a later runtime admission succeeds after that pre-entry abort. - Tag each exact ACPX close attempt with an immutable external or reconciliation origin and immutable reconciliation attempt number. - Keep timed-out autonomous reconciliation failures inside their originating three-attempt budget. - Batch concurrent external callers that join one reconciliation attempt so they cannot mint independent generations. - Consume a joined external intent on successful cleanup and on an earlier failed attempt that still has same-generation retries. - Renew exactly one bounded generation when a joined external intent reaches a failed, exhausted final reconciliation attempt. - Treat both protocol-close and provider-process cleanup failures as failed intent settlement, including non-timeout and timed-out late paths. - Prevent a coalescing external observer from relabeling an immutable autonomous attempt. - Reset the finite reconciliation budget only for a distinct external late-failure generation or one failed batched intent on an exhausted final attempt. - Isolate persistent-cleanup tests by cleanup domain and attach expected rejection handlers before fake timers release them. - Stabilize cleanup-settlement assertions exposed by GitHub Actions: observe retained proofs without relying on callback order, wait for the credential lease release rather than only the preceding credential-file deletion, and use a supported scalar size assertion instead of an unavailable Set matcher. - Add focused tests for driver behavior, event validation, bounded buffers, cleanup quarantine, admission gating, immutable attempt origins, final-attempt intent batching, direct and late cleanup failures, late success consumption, bounded reconciliation, exact pre-entry cancellation, credential recovery, and durable native-session outcomes. ## Verification - Exact head: `584cc420f6ca249cdc0a831779192ca827764d96`. - Stable patch ID for the combined exact delta: `5329d8123baf62c339a15bdff16717b3803ebb74`. - Stack position: #12404 is merged. This pull request targets `master`. #12406 is stacked on this pull request. - The exact pull request delta contains eight files: - `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-runtime-adapter.ts` - `packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-host.ts` - `packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts` - `packages/paperclip-runner/src/native-session-runtime.ts` - `packages/paperclip-runner/src/native-session-runtime.test.ts` - `git diff --check` passed for the exact eight-file delta. - This delta does not change dependencies, `pnpm-lock.yaml`, workflows, migrations, server selection, UI behavior, or production runner wiring. - GitHub previously exposed an unsupported Set matcher in the cleanup-settlement regression; this exact delta uses the repository-supported scalar `size` assertion without changing the tested behavior. - GitHub Actions: **PASS** for the exact head. The complete matrix is green after a failed-job-only rerun cleared one unrelated `plugin-worker-manager-duplex` flake; no patch or restack occurred. - Security checks: **PASS** for the exact head (Superagent, Snyk, Socket, and contributor trust). - Greptile: **PASS, 5/5** on the exact head with no open P1/P2 findings, recommendations, or follow-ups. - No local test result is claimed. GitHub Actions is the authoritative verification environment for this revision. ## Risks This change has medium package-local risk. It adds a new driver and changes native-session cleanup coordination. A lagging event consumer could otherwise lose terminal state. The driver reserves terminal capacity and rejects new work when bounded storage cannot safely accept it. A stalled, rejected, or late provider close could otherwise overlap a new session, retain credentials indefinitely, lose an external cleanup request, or consume unlimited retries. Cleanup-proof transfer keeps the staged credential owned across the pre-entry abort race, while the host scrubs it only after the adapter proves that no provider resource exists. Cleanup quarantine blocks conflicting admission and keeps exact attempts owned. Immutable attempt origins and attempt numbers prevent autonomous retries and coalesced observers from silently replenishing the cap. One batched external intent can renew one generation only after the exhausted final attempt fails; earlier failures remain within the original generation, and success consumes the intent. Each renewed generation remains capped at three autonomous attempts. The driver reports recovery and other unimplemented capabilities as unavailable. No server or runnerd factory selects this driver 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. 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/...`) 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 - [ ] 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 P1/P2 findings, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
3623a369aa
commit
74aabb7ea6
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
|
|
@ -32,14 +32,23 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
controller.abort(cancellation);
|
||||
const command = fakeCommand();
|
||||
const createRuntime = vi.fn();
|
||||
const retainFailedAdmissionCleanup = vi.fn();
|
||||
|
||||
await expect(
|
||||
openCodexAcpxRuntime(
|
||||
{ ...openOptions(command), signal: controller.signal },
|
||||
{
|
||||
...openOptions(command),
|
||||
signal: controller.signal,
|
||||
retainFailedAdmissionCleanup,
|
||||
},
|
||||
{ createRuntime },
|
||||
),
|
||||
).rejects.toBe(cancellation);
|
||||
|
||||
expect(retainFailedAdmissionCleanup).toHaveBeenCalledOnce();
|
||||
await expect(
|
||||
retainFailedAdmissionCleanup.mock.calls[0]?.[0],
|
||||
).resolves.toBeUndefined();
|
||||
expect(createRuntime).not.toHaveBeenCalled();
|
||||
expect(command.spawn).not.toHaveBeenCalled();
|
||||
});
|
||||
|
|
@ -248,6 +257,807 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("never overlaps a retained protocol close that has not settled", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
const runtimeClose = new Promise<void>(() => {});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(runtimeClose)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const child = fakeChild();
|
||||
const command = fakeCommand();
|
||||
vi.mocked(command.spawn).mockReturnValue(child);
|
||||
let runtimeOptions: AcpRuntimeOptions | undefined;
|
||||
const port = await openCodexAcpxRuntime(openOptions(command), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: (options) => {
|
||||
runtimeOptions = options;
|
||||
return runtime;
|
||||
},
|
||||
});
|
||||
runtimeOptions?.spawnAgent?.({
|
||||
command: "ignored",
|
||||
args: ["--stdio"],
|
||||
options: {},
|
||||
});
|
||||
|
||||
const firstClose = expect(
|
||||
port.close({ reason: "runtime close stalled" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
expect(child.kill).toHaveBeenCalledWith("SIGTERM");
|
||||
await firstClose;
|
||||
|
||||
// Repeated callers inherit the same bounded observation. A permanently
|
||||
// pending exact close remains the sole protocol attempt for this handle.
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
const secondClose = expect(
|
||||
port.close({ reason: "idempotent terminal close" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await secondClose;
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("allows a fresh close after a retained attempt rejects late", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectRuntimeClose!: (error: unknown) => void;
|
||||
const runtimeClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectRuntimeClose = reject;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(runtimeClose)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const child = fakeChild();
|
||||
const command = fakeCommand();
|
||||
vi.mocked(command.spawn).mockReturnValue(child);
|
||||
let runtimeOptions: AcpRuntimeOptions | undefined;
|
||||
const port = await openCodexAcpxRuntime(openOptions(command), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: (options) => {
|
||||
runtimeOptions = options;
|
||||
return runtime;
|
||||
},
|
||||
});
|
||||
runtimeOptions?.spawnAgent?.({
|
||||
command: "ignored",
|
||||
args: ["--stdio"],
|
||||
options: {},
|
||||
});
|
||||
|
||||
const firstClose = expect(
|
||||
port.close({ reason: "runtime close stalled" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await firstClose;
|
||||
expect(child.kill).toHaveBeenCalledWith("SIGTERM");
|
||||
|
||||
const protocolFailure = new Error("late protocol close failure");
|
||||
rejectRuntimeClose(protocolFailure);
|
||||
// The late-settlement observer schedules reconciliation asynchronously.
|
||||
// Wait for that fresh attempt instead of racing another caller against
|
||||
// the already-settled retained failure.
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2));
|
||||
await expect(
|
||||
port.close({ reason: "retry after retained failure" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX late protocol cleanup reconciliation 1",
|
||||
discardPersistentState: false,
|
||||
});
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("reconciles a retained close only after its late failure settles", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectFirstClose!: (error: unknown) => void;
|
||||
const firstRuntimeClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectFirstClose = reject;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(firstRuntimeClose)
|
||||
.mockResolvedValueOnce(undefined)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const child = fakeChild();
|
||||
const command = fakeCommand();
|
||||
vi.mocked(command.spawn).mockReturnValue(child);
|
||||
let runtimeOptions: AcpRuntimeOptions | undefined;
|
||||
const port = await openCodexAcpxRuntime(openOptions(command), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: (options) => {
|
||||
runtimeOptions = options;
|
||||
return runtime;
|
||||
},
|
||||
});
|
||||
runtimeOptions?.spawnAgent?.({
|
||||
command: "ignored",
|
||||
args: ["--stdio"],
|
||||
options: {},
|
||||
});
|
||||
|
||||
const firstClose = expect(
|
||||
port.close({ reason: "first protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await firstClose;
|
||||
|
||||
const inheritedClose = expect(
|
||||
port.close({ reason: "observe pending protocol close" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await inheritedClose;
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
|
||||
rejectFirstClose(new Error("older protocol close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2));
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX late protocol cleanup reconciliation 1",
|
||||
discardPersistentState: false,
|
||||
});
|
||||
await expect(
|
||||
port.close({ reason: "observe reconciled cleanup" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(runtime.close).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("shares a pending close across concurrent callers before reconciliation", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectFirstClose!: (error: unknown) => void;
|
||||
const firstRuntimeClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectFirstClose = reject;
|
||||
});
|
||||
let resolveFreshClose!: () => void;
|
||||
const freshRuntimeClose = new Promise<void>((resolve) => {
|
||||
resolveFreshClose = resolve;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(firstRuntimeClose)
|
||||
.mockReturnValueOnce(freshRuntimeClose)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const child = fakeChild();
|
||||
const command = fakeCommand();
|
||||
vi.mocked(command.spawn).mockReturnValue(child);
|
||||
let runtimeOptions: AcpRuntimeOptions | undefined;
|
||||
const port = await openCodexAcpxRuntime(openOptions(command), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: (options) => {
|
||||
runtimeOptions = options;
|
||||
return runtime;
|
||||
},
|
||||
});
|
||||
runtimeOptions?.spawnAgent?.({
|
||||
command: "ignored",
|
||||
args: ["--stdio"],
|
||||
options: {},
|
||||
});
|
||||
|
||||
const firstClose = expect(
|
||||
port.close({ reason: "first protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await firstClose;
|
||||
|
||||
const inheritedClose = expect(
|
||||
port.close({ reason: "observe pending protocol close" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
rejectFirstClose(new Error("retained protocol close failed"));
|
||||
await inheritedClose;
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2));
|
||||
|
||||
resolveFreshClose();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(2);
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX late protocol cleanup reconciliation 1",
|
||||
discardPersistentState: false,
|
||||
});
|
||||
await expect(
|
||||
port.close({ reason: "observe reconciled cleanup" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(runtime.close).toHaveBeenCalledTimes(2);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("does not let late settlements bypass the reconciliation retry bound", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 3 failed"));
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
});
|
||||
|
||||
const firstClose = expect(
|
||||
port.close({ reason: "first protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await firstClose;
|
||||
rejectInitialClose(new Error("initial protocol close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(4));
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("keeps timed-out reconciliation failures in their originating budget", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
const reconciliationRejectors: Array<(error: unknown) => void> = [];
|
||||
const reconciliationAttempts = Array.from(
|
||||
{ length: 3 },
|
||||
() =>
|
||||
new Promise<void>((_resolve, reject) => {
|
||||
reconciliationRejectors.push(reject);
|
||||
}),
|
||||
);
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockReturnValueOnce(reconciliationAttempts[0]!)
|
||||
.mockReturnValueOnce(reconciliationAttempts[1]!)
|
||||
.mockReturnValueOnce(reconciliationAttempts[2]!)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const firstClose = expect(
|
||||
port.close({ reason: "external protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await firstClose;
|
||||
rejectInitialClose(new Error("external close failed late"));
|
||||
|
||||
for (let index = 0; index < reconciliationRejectors.length; index += 1) {
|
||||
await vi.waitFor(() =>
|
||||
expect(runtime.close).toHaveBeenCalledTimes(index + 2),
|
||||
);
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: `ACPX late protocol cleanup reconciliation ${index + 1}`,
|
||||
discardPersistentState: false,
|
||||
});
|
||||
const overlappingExternalClose =
|
||||
index === 0
|
||||
? expect(
|
||||
port.close({
|
||||
reason: "external observer of reconciliation",
|
||||
}),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed")
|
||||
: null;
|
||||
// The external observer coalesces onto reconciliation attempt one. It
|
||||
// must not relabel that immutable attempt as an external generation.
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
if (overlappingExternalClose) await overlappingExternalClose;
|
||||
reconciliationRejectors[index]!(
|
||||
new Error(`reconciliation ${index + 1} failed late`),
|
||||
);
|
||||
}
|
||||
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
await Promise.resolve();
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("renews one budget when an external close joins the final reconciliation", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
let rejectFinalReconciliation!: (error: unknown) => void;
|
||||
const finalReconciliation = new Promise<void>((_resolve, reject) => {
|
||||
rejectFinalReconciliation = reject;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockReturnValueOnce(finalReconciliation)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const initialObserver = expect(
|
||||
port.close({ reason: "external protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await initialObserver;
|
||||
rejectInitialClose(new Error("external close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(4));
|
||||
|
||||
const finalObserver = expect(
|
||||
port.close({ reason: "external observer of final reconciliation" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await finalObserver;
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
|
||||
rejectFinalReconciliation(
|
||||
new Error("final reconciliation failed late"),
|
||||
);
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(5));
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX late protocol cleanup reconciliation 1",
|
||||
discardPersistentState: false,
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(5);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("batches coalesced external closes when the final reconciliation fails", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
let rejectFinalReconciliation!: (error: unknown) => void;
|
||||
const finalReconciliation = new Promise<void>((_resolve, reject) => {
|
||||
rejectFinalReconciliation = reject;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockReturnValueOnce(finalReconciliation)
|
||||
.mockRejectedValueOnce(new Error("renewed reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("renewed reconciliation 2 failed"))
|
||||
.mockRejectedValueOnce(new Error("renewed reconciliation 3 failed"));
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const initialObserver = expect(
|
||||
port.close({ reason: "external protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await initialObserver;
|
||||
rejectInitialClose(new Error("external close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(4));
|
||||
|
||||
const coalescedObservers = ["first", "second", "third"].map((label) =>
|
||||
expect(
|
||||
port.close({ reason: `${label} external observer` }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed"),
|
||||
);
|
||||
rejectFinalReconciliation(
|
||||
new Error("final reconciliation failed with joined observers"),
|
||||
);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
await Promise.all(coalescedObservers);
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(7));
|
||||
expect(
|
||||
vi
|
||||
.mocked(runtime.close)
|
||||
.mock.calls.slice(4)
|
||||
.map(([input]) => input.reason),
|
||||
).toEqual([
|
||||
"ACPX late protocol cleanup reconciliation 1",
|
||||
"ACPX late protocol cleanup reconciliation 2",
|
||||
"ACPX late protocol cleanup reconciliation 3",
|
||||
]);
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
await Promise.resolve();
|
||||
expect(runtime.close).toHaveBeenCalledTimes(7);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("consumes an external intent when a timed-out final reconciliation succeeds late", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
let resolveFinalReconciliation!: () => void;
|
||||
const finalReconciliation = new Promise<void>((resolve) => {
|
||||
resolveFinalReconciliation = resolve;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockReturnValueOnce(finalReconciliation);
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const initialObserver = expect(
|
||||
port.close({ reason: "external protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await initialObserver;
|
||||
rejectInitialClose(new Error("external close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(4));
|
||||
|
||||
const finalObserver = expect(
|
||||
port.close({
|
||||
reason: "external observer of successful reconciliation",
|
||||
}),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await finalObserver;
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
|
||||
resolveFinalReconciliation();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
await Promise.resolve();
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("renews an external intent when bounded protocol success cannot terminate the provider", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
let resolveFinalReconciliation!: () => void;
|
||||
const finalReconciliation = new Promise<void>((resolve) => {
|
||||
resolveFinalReconciliation = resolve;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockReturnValueOnce(finalReconciliation)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const child = fakeChild();
|
||||
child.kill = vi.fn(() => true);
|
||||
const command = fakeCommand();
|
||||
vi.mocked(command.spawn).mockReturnValue(child);
|
||||
let runtimeOptions: AcpRuntimeOptions | undefined;
|
||||
const port = await openCodexAcpxRuntime(openOptions(command), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: (options) => {
|
||||
runtimeOptions = options;
|
||||
return runtime;
|
||||
},
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const initialObserver = expect(
|
||||
port.close({ reason: "external protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await initialObserver;
|
||||
rejectInitialClose(new Error("external close failed late"));
|
||||
for (
|
||||
let turn = 0;
|
||||
turn < 50 && vi.mocked(runtime.close).mock.calls.length < 4;
|
||||
turn += 1
|
||||
) {
|
||||
await Promise.resolve();
|
||||
}
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
|
||||
runtimeOptions?.spawnAgent?.({
|
||||
command: "ignored",
|
||||
args: ["--stdio"],
|
||||
options: {},
|
||||
});
|
||||
const finalObserver = expect(
|
||||
port.close({ reason: "external observer of failed process cleanup" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
resolveFinalReconciliation();
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(4_000);
|
||||
await finalObserver;
|
||||
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(5));
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX late protocol cleanup reconciliation 1",
|
||||
discardPersistentState: false,
|
||||
});
|
||||
child.signalCode = "SIGKILL";
|
||||
child.emit("exit", null, "SIGKILL");
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(5);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("renews an external intent when late protocol success follows process cleanup failure", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
let resolveFinalReconciliation!: () => void;
|
||||
const finalReconciliation = new Promise<void>((resolve) => {
|
||||
resolveFinalReconciliation = resolve;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockReturnValueOnce(finalReconciliation)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const child = fakeChild();
|
||||
child.kill = vi.fn(() => true);
|
||||
const command = fakeCommand();
|
||||
vi.mocked(command.spawn).mockReturnValue(child);
|
||||
let runtimeOptions: AcpRuntimeOptions | undefined;
|
||||
const port = await openCodexAcpxRuntime(openOptions(command), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: (options) => {
|
||||
runtimeOptions = options;
|
||||
return runtime;
|
||||
},
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const initialObserver = expect(
|
||||
port.close({ reason: "external protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await initialObserver;
|
||||
rejectInitialClose(new Error("external close failed late"));
|
||||
for (
|
||||
let turn = 0;
|
||||
turn < 50 && vi.mocked(runtime.close).mock.calls.length < 4;
|
||||
turn += 1
|
||||
) {
|
||||
await Promise.resolve();
|
||||
}
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
|
||||
runtimeOptions?.spawnAgent?.({
|
||||
command: "ignored",
|
||||
args: ["--stdio"],
|
||||
options: {},
|
||||
});
|
||||
const finalObserver = expect(
|
||||
port.close({ reason: "external observer of failed process cleanup" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await vi.advanceTimersByTimeAsync(4_000);
|
||||
await finalObserver;
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
|
||||
child.signalCode = "SIGKILL";
|
||||
child.emit("exit", null, "SIGKILL");
|
||||
resolveFinalReconciliation();
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(5));
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX late protocol cleanup reconciliation 1",
|
||||
discardPersistentState: false,
|
||||
});
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(5);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("shares one reconciliation budget across repeated bounded observers", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectRetainedClose!: (error: unknown) => void;
|
||||
const retainedClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectRetainedClose = reject;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(retainedClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 3 failed"));
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
for (const reason of [
|
||||
"first observer",
|
||||
"second observer",
|
||||
"third observer",
|
||||
]) {
|
||||
const close = expect(port.close({ reason })).rejects.toThrow(
|
||||
"ACPX runtime and provider cleanup failed",
|
||||
);
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await close;
|
||||
}
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
rejectRetainedClose(new Error("retained close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(4));
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(4);
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("gives each late failure generation a bounded reconciliation budget", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let rejectInitialClose!: (error: unknown) => void;
|
||||
const initialClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectInitialClose = reject;
|
||||
});
|
||||
let rejectNewerClose!: (error: unknown) => void;
|
||||
const newerClose = new Promise<void>((_resolve, reject) => {
|
||||
rejectNewerClose = reject;
|
||||
});
|
||||
vi.mocked(runtime.close)
|
||||
.mockReturnValueOnce(initialClose)
|
||||
.mockRejectedValueOnce(new Error("reconciliation 1 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 2 failed"))
|
||||
.mockRejectedValueOnce(new Error("reconciliation 3 failed"))
|
||||
.mockReturnValueOnce(newerClose)
|
||||
.mockResolvedValueOnce(undefined);
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const firstClose = expect(
|
||||
port.close({ reason: "first protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await firstClose;
|
||||
rejectInitialClose(new Error("initial close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(4));
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
|
||||
const secondClose = expect(
|
||||
port.close({ reason: "newer protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await secondClose;
|
||||
expect(runtime.close).toHaveBeenCalledTimes(5);
|
||||
|
||||
rejectNewerClose(new Error("newer close failed late"));
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(6));
|
||||
expect(runtime.close).toHaveBeenLastCalledWith({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX late protocol cleanup reconciliation 1",
|
||||
discardPersistentState: false,
|
||||
});
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("marks a retained protocol close terminal when it succeeds late", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const runtime = fakeRuntime();
|
||||
let resolveRetainedClose!: () => void;
|
||||
const retainedClose = new Promise<void>((resolve) => {
|
||||
resolveRetainedClose = resolve;
|
||||
});
|
||||
vi.mocked(runtime.close).mockReturnValueOnce(retainedClose);
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
runtimeCloseTimeoutMs: 1,
|
||||
});
|
||||
|
||||
const close = expect(
|
||||
port.close({ reason: "protocol close stalls" }),
|
||||
).rejects.toThrow("ACPX runtime and provider cleanup failed");
|
||||
await Promise.resolve();
|
||||
await vi.advanceTimersByTimeAsync(1);
|
||||
await close;
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
|
||||
resolveRetainedClose();
|
||||
await vi.advanceTimersByTimeAsync(0);
|
||||
await expect(
|
||||
port.close({ reason: "observe late success" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("maps prompt turns to the admitted ACPX handle", async () => {
|
||||
const runtime = fakeRuntime();
|
||||
const turn = {
|
||||
|
|
@ -566,11 +1376,14 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
.mockResolvedValueOnce(undefined);
|
||||
const controller = new AbortController();
|
||||
const cancellation = new Error("runtime admission cancelled");
|
||||
const retainedAdmissionCleanups: Promise<void>[] = [];
|
||||
|
||||
const opening = openCodexAcpxRuntime(
|
||||
{
|
||||
...openOptions(fakeCommand()),
|
||||
signal: controller.signal,
|
||||
retainFailedAdmissionCleanup: (cleanup) =>
|
||||
retainedAdmissionCleanups.push(cleanup),
|
||||
},
|
||||
{
|
||||
createRegistry: () => registry(),
|
||||
|
|
@ -584,9 +1397,11 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
|
||||
controller.abort(cancellation);
|
||||
await expect(opening).rejects.toBe(cancellation);
|
||||
expect(retainedAdmissionCleanups).toHaveLength(1);
|
||||
resolveHandshake?.(HANDLE);
|
||||
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2));
|
||||
await expect(retainedAdmissionCleanups[0]).resolves.toBeUndefined();
|
||||
expect(vi.mocked(runtime.close).mock.calls[0]?.[0]).toEqual({
|
||||
handle: HANDLE,
|
||||
reason: "ACPX runtime admission aborted",
|
||||
|
|
@ -631,7 +1446,9 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
|
||||
controller.abort(cancellation);
|
||||
await expect(opening).rejects.toBe(cancellation);
|
||||
expect(retainedCleanups).toHaveLength(1);
|
||||
// The exact late-handshake owner and the host-facing aggregate proof are
|
||||
// distinct retained promises over the same cleanup obligation.
|
||||
expect(retainedCleanups).toHaveLength(2);
|
||||
resolveHandshake?.(HANDLE);
|
||||
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledOnce());
|
||||
|
|
@ -649,7 +1466,7 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
|
||||
resolveClose?.();
|
||||
await retainedCleanups[0];
|
||||
await expect(Promise.all(retainedCleanups)).resolves.toBeDefined();
|
||||
expect(cleanupSettled).toBe(true);
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
|
@ -892,13 +1709,22 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
} as never);
|
||||
|
||||
await expect(opening).rejects.toBeInstanceOf(Error);
|
||||
expect(retainedCleanups).toHaveLength(2);
|
||||
// Retain the pending handshake, the discovered runtime handle cleanup,
|
||||
// and their host-facing aggregate proof until the same obligation settles.
|
||||
expect(retainedCleanups).toHaveLength(3);
|
||||
const settledCleanups = new Set<Promise<void>>();
|
||||
for (const cleanup of retainedCleanups) {
|
||||
void cleanup
|
||||
.finally(() => settledCleanups.add(cleanup))
|
||||
.catch(() => undefined);
|
||||
}
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(5));
|
||||
await expect(retainedCleanups[0]).resolves.toBeUndefined();
|
||||
await vi.waitFor(() => expect(settledCleanups.size).toBe(1));
|
||||
expect(runtime.close).toHaveBeenCalledTimes(5);
|
||||
|
||||
rejectHandshake?.(new Error("test handshake stopped"));
|
||||
await expect(retainedCleanups[1]).rejects.toThrow("test handshake stopped");
|
||||
await vi.waitFor(() => expect(settledCleanups.size).toBe(3));
|
||||
await expect(Promise.all(retainedCleanups)).resolves.toBeDefined();
|
||||
});
|
||||
|
||||
it("aggregates asynchronous provider signal errors after a failed handshake", async () => {
|
||||
|
|
@ -1267,6 +2093,7 @@ function openOptions(
|
|||
},
|
||||
systemInstructions: "Use Paperclip tools.",
|
||||
mcpServers: [],
|
||||
retainFailedAdmissionCleanup: vi.fn(),
|
||||
};
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -24,6 +24,22 @@ const VERIFIED_COMMAND_SENTINEL = "paperclip-verified-acpx-command";
|
|||
const DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS = 2_000;
|
||||
const RETAINED_ADMISSION_CLEANUP_RETRY_MIN_MS = 10;
|
||||
const RETAINED_ADMISSION_CLEANUP_RETRY_MAX_MS = 30_000;
|
||||
const PROVIDER_TERM_EXIT_TIMEOUT_MS = 2_000;
|
||||
const PROVIDER_KILL_EXIT_TIMEOUT_MS = 2_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.
|
||||
export const DEFAULT_CODEX_ACPX_RUNTIME_SHUTDOWN_BOUND_MS =
|
||||
DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS +
|
||||
PROVIDER_TERM_EXIT_TIMEOUT_MS +
|
||||
PROVIDER_KILL_EXIT_TIMEOUT_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>>();
|
||||
|
||||
class AcpxRuntimeCloseTimeoutError extends Error {
|
||||
|
|
@ -53,12 +69,19 @@ export async function openCodexAcpxRuntime(
|
|||
options: AcpxRuntimePortOpenOptions,
|
||||
dependencies: CodexAcpxRuntimeDependencies = {},
|
||||
): Promise<AcpxRuntimePort> {
|
||||
if (options.signal?.aborted) {
|
||||
// The host may have already transferred its staged credential to this
|
||||
// pending admission before this microtask begins. No adapter resources
|
||||
// exist yet, so publish an already-complete cleanup proof before preserving
|
||||
// the caller's exact abort reason.
|
||||
options.retainFailedAdmissionCleanup(Promise.resolve());
|
||||
throw options.signal.reason;
|
||||
}
|
||||
if (options.profile.agent !== "codex") {
|
||||
throw new Error(
|
||||
"The production ACPX runtime currently supports Codex only",
|
||||
);
|
||||
}
|
||||
options.signal?.throwIfAborted();
|
||||
// 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
|
||||
|
|
@ -81,6 +104,7 @@ export async function openCodexAcpxRuntime(
|
|||
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)) {
|
||||
|
|
@ -236,14 +260,19 @@ export async function openCodexAcpxRuntime(
|
|||
handle = await raceRuntimeHandshakeWithAbort(handshake, options.signal);
|
||||
} catch (error) {
|
||||
if (options.signal.aborted) {
|
||||
retainCleanup(
|
||||
handshake.then((lateHandle) =>
|
||||
abortedHandshakeCleanup = handshake.then(
|
||||
(lateHandle) =>
|
||||
admissionCleanup!.runRetained(
|
||||
lateHandle,
|
||||
"ACPX runtime admission aborted",
|
||||
),
|
||||
),
|
||||
() =>
|
||||
admissionCleanup!.runRetained(
|
||||
failedHandshakeHandle,
|
||||
"ACPX runtime admission aborted",
|
||||
),
|
||||
);
|
||||
retainCleanup(abortedHandshakeCleanup);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
|
@ -252,12 +281,30 @@ export async function openCodexAcpxRuntime(
|
|||
options.signal.throwIfAborted();
|
||||
}
|
||||
} catch (error) {
|
||||
const cleanupHandle = handle ?? failedHandshakeHandle;
|
||||
const cleanupErrors = await admissionCleanup.run(
|
||||
handle ?? failedHandshakeHandle,
|
||||
cleanupHandle,
|
||||
options.signal?.aborted
|
||||
? "ACPX runtime admission aborted"
|
||||
: "ACPX session handshake failed",
|
||||
);
|
||||
const retainedCleanup =
|
||||
cleanupErrors.length === 0
|
||||
? Promise.resolve()
|
||||
: admissionCleanup.runRetained(
|
||||
cleanupHandle,
|
||||
options.signal?.aborted
|
||||
? "ACPX runtime admission aborted"
|
||||
: "ACPX session handshake failed",
|
||||
);
|
||||
const cleanupProof =
|
||||
abortedHandshakeCleanup === null
|
||||
? retainedCleanup
|
||||
: Promise.all([retainedCleanup, abortedHandshakeCleanup]).then(
|
||||
() => undefined,
|
||||
);
|
||||
options.retainFailedAdmissionCleanup(cleanupProof);
|
||||
retainCleanup(cleanupProof);
|
||||
if (cleanupErrors.length > 0) {
|
||||
throw new AggregateError(
|
||||
[error, ...cleanupErrors],
|
||||
|
|
@ -277,13 +324,23 @@ export async function openCodexAcpxRuntime(
|
|||
runtime,
|
||||
handle,
|
||||
requireIdentity(handle),
|
||||
admissionCleanup,
|
||||
children,
|
||||
runtimeCloseTimeoutMs,
|
||||
);
|
||||
} catch (error) {
|
||||
const cleanupErrors = await admissionCleanup.run(
|
||||
handle,
|
||||
"ACPX runtime identity validation failed",
|
||||
);
|
||||
const cleanupProof =
|
||||
cleanupErrors.length === 0
|
||||
? Promise.resolve()
|
||||
: admissionCleanup.runRetained(
|
||||
handle,
|
||||
"ACPX runtime identity validation failed",
|
||||
);
|
||||
options.retainFailedAdmissionCleanup(cleanupProof);
|
||||
retainCleanup(cleanupProof);
|
||||
if (cleanupErrors.length > 0) {
|
||||
throw new AggregateError(
|
||||
[error, ...cleanupErrors],
|
||||
|
|
@ -371,12 +428,12 @@ class RuntimeAdmissionCleanup {
|
|||
const targetKey = this.#resolveTargetKey(rawTargetKey, handle);
|
||||
const existing = this.#registeredTargets.get(targetKey);
|
||||
if (existing !== undefined) {
|
||||
if (handle !== null) {
|
||||
existing.handle =
|
||||
existing.handle === null
|
||||
? handle
|
||||
existing.handle =
|
||||
existing.handle === null
|
||||
? handle
|
||||
: handle === null
|
||||
? existing.handle
|
||||
: preferRuntimeAdmissionCleanupHandle(existing.handle, handle);
|
||||
}
|
||||
this.#targetAliases.set(rawTargetKey, targetKey);
|
||||
return existing.cleanup!;
|
||||
}
|
||||
|
|
@ -407,8 +464,7 @@ class RuntimeAdmissionCleanup {
|
|||
const fallbackHandle =
|
||||
this.#registeredTargets.get(fallbackTargetKey)?.handle;
|
||||
if (
|
||||
fallbackHandle !== undefined &&
|
||||
fallbackHandle !== null &&
|
||||
fallbackHandle != null &&
|
||||
nonEmptyRuntimeIdentity(fallbackHandle.acpxRecordId) === undefined &&
|
||||
sameRuntimeAdmissionCleanupOwner(fallbackHandle, handle)
|
||||
) {
|
||||
|
|
@ -590,9 +646,232 @@ function runtimePort(
|
|||
runtime: AcpRuntime,
|
||||
handle: AcpRuntimeHandle,
|
||||
identity: AcpxRuntimePortIdentity,
|
||||
admissionCleanup: RuntimeAdmissionCleanup,
|
||||
children: SpawnedChildSet,
|
||||
runtimeCloseTimeoutMs: number,
|
||||
): AcpxRuntimePort {
|
||||
return {
|
||||
type RuntimeCloseAttempt = {
|
||||
readonly outcome: Promise<unknown | null>;
|
||||
readonly reconciliationGeneration: number;
|
||||
readonly origin:
|
||||
| { readonly kind: "external" }
|
||||
| {
|
||||
readonly kind: "reconciliation";
|
||||
readonly generation: number;
|
||||
readonly attemptNumber: number;
|
||||
};
|
||||
pendingExternalIntent: boolean;
|
||||
};
|
||||
let runtimeClosed = false;
|
||||
let runtimeCloseAttempt: RuntimeCloseAttempt | undefined;
|
||||
let lateReconciliationOwner: Promise<void> | undefined;
|
||||
// Each independently observed late failure receives a bounded reconciliation
|
||||
// budget. Exhausting retries for an older generation must not prevent a newer
|
||||
// late failure from acquiring its own recovery owner.
|
||||
let lateReconciliationAttemptGeneration = 0;
|
||||
let lateReconciliationAttempts = 0;
|
||||
let lateFailureGeneration = 0;
|
||||
let reconciledLateFailureGeneration = 0;
|
||||
const watchedReleasedAttempts = new Set<RuntimeCloseAttempt>();
|
||||
|
||||
const hasUnreconciledLateFailure = (): boolean =>
|
||||
reconciledLateFailureGeneration < lateFailureGeneration;
|
||||
|
||||
const consumePendingExternalIntent = (
|
||||
attempt: RuntimeCloseAttempt,
|
||||
failed: boolean,
|
||||
): boolean => {
|
||||
const pending = attempt.pendingExternalIntent;
|
||||
attempt.pendingExternalIntent = false;
|
||||
return (
|
||||
pending &&
|
||||
failed &&
|
||||
attempt.origin.kind === "reconciliation" &&
|
||||
attempt.origin.attemptNumber >=
|
||||
MAX_LATE_RUNTIME_CLEANUP_RECONCILIATION_ATTEMPTS
|
||||
);
|
||||
};
|
||||
|
||||
const scheduleLateFailureReconciliation = (): void => {
|
||||
if (
|
||||
runtimeCloseAttempt ||
|
||||
lateReconciliationOwner ||
|
||||
!hasUnreconciledLateFailure()
|
||||
) {
|
||||
return;
|
||||
}
|
||||
if (lateReconciliationAttemptGeneration !== lateFailureGeneration) {
|
||||
lateReconciliationAttemptGeneration = lateFailureGeneration;
|
||||
lateReconciliationAttempts = 0;
|
||||
}
|
||||
if (
|
||||
lateReconciliationAttempts >=
|
||||
MAX_LATE_RUNTIME_CLEANUP_RECONCILIATION_ATTEMPTS
|
||||
) {
|
||||
return;
|
||||
}
|
||||
const attemptGeneration = lateFailureGeneration;
|
||||
const attemptNumber = lateReconciliationAttempts + 1;
|
||||
lateReconciliationAttempts = attemptNumber;
|
||||
let retry = false;
|
||||
const reconciliation = closeRuntime({
|
||||
reason: `ACPX late protocol cleanup reconciliation ${attemptNumber}`,
|
||||
reconciliation: {
|
||||
generation: attemptGeneration,
|
||||
attemptNumber,
|
||||
},
|
||||
}).then(
|
||||
() => {
|
||||
if (hasUnreconciledLateFailure()) {
|
||||
retry =
|
||||
lateFailureGeneration === attemptGeneration &&
|
||||
attemptNumber < MAX_LATE_RUNTIME_CLEANUP_RECONCILIATION_ATTEMPTS;
|
||||
}
|
||||
},
|
||||
() => {
|
||||
retry =
|
||||
lateFailureGeneration === attemptGeneration &&
|
||||
attemptNumber < MAX_LATE_RUNTIME_CLEANUP_RECONCILIATION_ATTEMPTS;
|
||||
},
|
||||
);
|
||||
const owner = reconciliation.finally(() => {
|
||||
if (lateReconciliationOwner === owner)
|
||||
lateReconciliationOwner = undefined;
|
||||
if (retry || hasUnreconciledLateFailure()) {
|
||||
queueMicrotask(scheduleLateFailureReconciliation);
|
||||
}
|
||||
});
|
||||
lateReconciliationOwner = owner;
|
||||
retainRuntimeCleanupOwner(owner);
|
||||
};
|
||||
|
||||
const watchPendingAttempt = (
|
||||
attempt: RuntimeCloseAttempt,
|
||||
processCleanupSucceeded: boolean,
|
||||
): void => {
|
||||
if (watchedReleasedAttempts.has(attempt)) return;
|
||||
watchedReleasedAttempts.add(attempt);
|
||||
void attempt.outcome.then((error) => {
|
||||
watchedReleasedAttempts.delete(attempt);
|
||||
if (runtimeCloseAttempt === attempt) runtimeCloseAttempt = undefined;
|
||||
const renewForExternalIntent = consumePendingExternalIntent(
|
||||
attempt,
|
||||
error !== null || !processCleanupSucceeded,
|
||||
);
|
||||
if (renewForExternalIntent) lateFailureGeneration += 1;
|
||||
if (error === null) {
|
||||
if (processCleanupSucceeded) {
|
||||
reconciledLateFailureGeneration = Math.max(
|
||||
reconciledLateFailureGeneration,
|
||||
attempt.reconciliationGeneration,
|
||||
);
|
||||
runtimeClosed = !hasUnreconciledLateFailure();
|
||||
}
|
||||
scheduleLateFailureReconciliation();
|
||||
return;
|
||||
}
|
||||
// A newer successful close cannot erase an older outcome that had not
|
||||
// settled yet. Re-open cleanup state and autonomously create a bounded
|
||||
// reconciliation generation so the late failure is not suppression-only.
|
||||
// An autonomous failure remains charged to the budget of the generation
|
||||
// that created it. If external callers coalesced onto the exhausted final
|
||||
// attempt, their single batched intent creates exactly one new generation;
|
||||
// joins on earlier attempts are satisfied by the remaining same-generation
|
||||
// retries.
|
||||
if (attempt.origin.kind === "external") {
|
||||
lateFailureGeneration += 1;
|
||||
}
|
||||
runtimeClosed = false;
|
||||
scheduleLateFailureReconciliation();
|
||||
});
|
||||
};
|
||||
|
||||
async function closeRuntime(input: {
|
||||
reason: string;
|
||||
reconciliation?: {
|
||||
generation: number;
|
||||
attemptNumber: number;
|
||||
};
|
||||
}): Promise<void> {
|
||||
if (runtimeClosed) return;
|
||||
if (
|
||||
runtimeCloseAttempt?.origin.kind === "reconciliation" &&
|
||||
input.reconciliation === undefined
|
||||
) {
|
||||
// Preserve the attempt's autonomous origin while remembering that one or
|
||||
// more external callers requested a fresh cleanup observation. The exact
|
||||
// outcome consumes this bit, so coalesced callers cannot mint generations
|
||||
// independently.
|
||||
runtimeCloseAttempt.pendingExternalIntent = true;
|
||||
}
|
||||
if (!runtimeCloseAttempt) {
|
||||
// A close can reconcile only failures already known when its protocol
|
||||
// attempt begins. A released older attempt may reject while this one is
|
||||
// in flight; that later generation must trigger a subsequent close.
|
||||
runtimeCloseAttempt = {
|
||||
outcome: ownedRuntimeCloseOutcome(runtime, handle, input.reason),
|
||||
reconciliationGeneration:
|
||||
input.reconciliation?.generation ?? lateFailureGeneration,
|
||||
origin:
|
||||
input.reconciliation === undefined
|
||||
? { kind: "external" }
|
||||
: {
|
||||
kind: "reconciliation",
|
||||
generation: input.reconciliation.generation,
|
||||
attemptNumber: input.reconciliation.attemptNumber,
|
||||
},
|
||||
pendingExternalIntent: false,
|
||||
};
|
||||
}
|
||||
const observedAttempt = runtimeCloseAttempt;
|
||||
const processCleanup = terminateChildrenAfterCloseBound(
|
||||
observedAttempt.outcome,
|
||||
children,
|
||||
runtimeCloseTimeoutMs,
|
||||
);
|
||||
// The caller may stop waiting, but the exact ACPX protocol cleanup stays
|
||||
// owned and remains this handle's sole close attempt until it settles.
|
||||
// Provider termination still proceeds at the deadline.
|
||||
const [closeError, processErrors] = await Promise.all([
|
||||
boundedCloseOutcome(observedAttempt.outcome, runtimeCloseTimeoutMs),
|
||||
processCleanup,
|
||||
]);
|
||||
if (closeError instanceof AcpxRuntimeCloseTimeoutError) {
|
||||
watchPendingAttempt(observedAttempt, processErrors.length === 0);
|
||||
} else {
|
||||
if (
|
||||
consumePendingExternalIntent(
|
||||
observedAttempt,
|
||||
closeError !== null || processErrors.length > 0,
|
||||
)
|
||||
) {
|
||||
lateFailureGeneration += 1;
|
||||
}
|
||||
if (runtimeCloseAttempt === observedAttempt) {
|
||||
runtimeCloseAttempt = undefined;
|
||||
}
|
||||
}
|
||||
if (processErrors.length === 0 && closeError === null) {
|
||||
reconciledLateFailureGeneration = Math.max(
|
||||
reconciledLateFailureGeneration,
|
||||
observedAttempt.reconciliationGeneration,
|
||||
);
|
||||
runtimeClosed = !hasUnreconciledLateFailure();
|
||||
} else {
|
||||
runtimeClosed = false;
|
||||
}
|
||||
scheduleLateFailureReconciliation();
|
||||
if (closeError !== null || processErrors.length > 0) {
|
||||
const errors = [closeError, ...processErrors].filter(
|
||||
(error): error is unknown => error !== null,
|
||||
);
|
||||
throw new AggregateError(
|
||||
errors,
|
||||
"ACPX runtime and provider cleanup failed",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
const port: AcpxRuntimePort = {
|
||||
async identity() {
|
||||
return structuredClone(identity);
|
||||
},
|
||||
|
|
@ -622,16 +901,9 @@ function runtimePort(
|
|||
...(input.signal ? { signal: input.signal } : {}),
|
||||
});
|
||||
},
|
||||
async close(input) {
|
||||
const errors = await admissionCleanup.run(handle, input.reason);
|
||||
if (errors.length > 0) {
|
||||
throw new AggregateError(
|
||||
errors,
|
||||
"ACPX runtime and provider cleanup failed",
|
||||
);
|
||||
}
|
||||
},
|
||||
close: closeRuntime,
|
||||
};
|
||||
return port;
|
||||
}
|
||||
|
||||
function runtimeCloseOutcome(
|
||||
|
|
@ -663,6 +935,70 @@ async function closeOutcomeWithin(
|
|||
return outcome;
|
||||
}
|
||||
|
||||
function ownedRuntimeCloseOutcome(
|
||||
runtime: AcpRuntime,
|
||||
handle: AcpRuntimeHandle,
|
||||
reason: string,
|
||||
): Promise<unknown | null> {
|
||||
const cleanup = Promise.resolve()
|
||||
.then(() =>
|
||||
runtime.close({ handle, reason, discardPersistentState: false }),
|
||||
)
|
||||
.then(
|
||||
() => null,
|
||||
(error: unknown) => error,
|
||||
);
|
||||
return retainRuntimeCleanupOwner(cleanup);
|
||||
}
|
||||
|
||||
function retainRuntimeCleanupOwner<T>(cleanup: Promise<T>): Promise<T> {
|
||||
activeRuntimeCleanupOwners.add(cleanup);
|
||||
void cleanup
|
||||
.finally(() => activeRuntimeCleanupOwners.delete(cleanup))
|
||||
.catch(() => undefined);
|
||||
return cleanup;
|
||||
}
|
||||
|
||||
async function terminateChildrenAfterCloseBound(
|
||||
closeOutcome: Promise<unknown | null>,
|
||||
children: SpawnedChildSet,
|
||||
timeoutMs: number,
|
||||
): Promise<unknown[]> {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
try {
|
||||
await Promise.race([
|
||||
closeOutcome.then(() => undefined),
|
||||
new Promise<void>((resolve) => {
|
||||
timer = setTimeout(resolve, Math.max(1, Math.floor(timeoutMs)));
|
||||
timer.unref();
|
||||
}),
|
||||
]);
|
||||
} finally {
|
||||
if (timer) clearTimeout(timer);
|
||||
}
|
||||
return await children.terminate();
|
||||
}
|
||||
|
||||
async function boundedCloseOutcome(
|
||||
closeOutcome: Promise<unknown | null>,
|
||||
timeoutMs: number,
|
||||
): Promise<unknown | null> {
|
||||
const boundedTimeoutMs = Math.max(1, timeoutMs);
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const outcome = await Promise.race([
|
||||
closeOutcome.then((error) => ({ error })),
|
||||
new Promise<{ error: unknown }>((resolve) => {
|
||||
timer = setTimeout(
|
||||
() => resolve({ error: new AcpxRuntimeCloseTimeoutError() }),
|
||||
boundedTimeoutMs,
|
||||
);
|
||||
timer.unref();
|
||||
}),
|
||||
]);
|
||||
if (timer) clearTimeout(timer);
|
||||
return outcome.error;
|
||||
}
|
||||
|
||||
function delay(timeoutMs: number): Promise<void> {
|
||||
return new Promise((resolve) => {
|
||||
const timer = setTimeout(resolve, timeoutMs);
|
||||
|
|
@ -703,7 +1039,7 @@ class SpawnedChildSet {
|
|||
const terminateOutcome = await signalAndWaitForExit(
|
||||
tracked,
|
||||
"SIGTERM",
|
||||
2_000,
|
||||
PROVIDER_TERM_EXIT_TIMEOUT_MS,
|
||||
);
|
||||
if (terminateOutcome.error !== undefined) {
|
||||
pushUnique(errors, terminateOutcome.error);
|
||||
|
|
@ -712,7 +1048,7 @@ class SpawnedChildSet {
|
|||
const killOutcome = await signalAndWaitForExit(
|
||||
tracked,
|
||||
"SIGKILL",
|
||||
2_000,
|
||||
PROVIDER_KILL_EXIT_TIMEOUT_MS,
|
||||
);
|
||||
if (killOutcome.error !== undefined) {
|
||||
pushUnique(errors, killOutcome.error);
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { join } from "node:path";
|
|||
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { openCodexAcpxRuntime } from "./codex-runtime-adapter.js";
|
||||
import { stageManagedCodexCredential } from "./codex-credentials.js";
|
||||
import type {
|
||||
VerifiedAcpxCommandLease,
|
||||
|
|
@ -14,6 +15,7 @@ import {
|
|||
AcpxRuntimeHost,
|
||||
type AcpxRuntimeHostDependencies,
|
||||
type AcpxRuntimePort,
|
||||
type AcpxRuntimePortOpenOptions,
|
||||
type AcpxRuntimeTurn,
|
||||
} from "./runtime-host.js";
|
||||
|
||||
|
|
@ -59,6 +61,65 @@ describe("ACPX runtime host", () => {
|
|||
expect(fixture.commandClose).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("scrubs credentials when abort wins before the adapter body starts", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const controller = new AbortController();
|
||||
const cancellation = new Error("runtime admission cancelled before entry");
|
||||
const createRuntime = vi.fn();
|
||||
let credentialHome = "";
|
||||
const openRuntime = vi.fn((options: AcpxRuntimePortOpenOptions) => {
|
||||
credentialHome = options.launchEnvironment.CODEX_HOME!;
|
||||
// The host has scheduled its openRuntime callback and transferred the
|
||||
// staged credential to that pending admission. Abort before entering the
|
||||
// adapter so its pre-entry path must publish a completed cleanup proof.
|
||||
controller.abort(cancellation);
|
||||
return openCodexAcpxRuntime(options, { createRuntime });
|
||||
});
|
||||
|
||||
await expect(
|
||||
AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "deny-all",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
signal: controller.signal,
|
||||
},
|
||||
fixture.dependencies({ openRuntime }),
|
||||
),
|
||||
).rejects.toBe(cancellation);
|
||||
|
||||
expect(openRuntime).toHaveBeenCalledOnce();
|
||||
expect(createRuntime).not.toHaveBeenCalled();
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
const authPath = join(credentialHome, "auth.json");
|
||||
const contender = await vi.waitFor(() =>
|
||||
stageManagedCodexCredential({
|
||||
agentHomeDirectory: credentialHome,
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
}),
|
||||
);
|
||||
await contender.close();
|
||||
await expect(readFile(authPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
|
||||
const retryRuntime = runtimePort();
|
||||
const retryHost = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "deny-all",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => retryRuntime }),
|
||||
);
|
||||
await retryHost.close({ reason: "retry admission complete" });
|
||||
expect(retryRuntime.close).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("composes admission, isolation, model verification, and cleanup", async () => {
|
||||
const fixture = await hostFixture();
|
||||
let capturedEnvironment: Readonly<NodeJS.ProcessEnv> = {};
|
||||
|
|
@ -301,6 +362,77 @@ describe("ACPX runtime host", () => {
|
|||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("retains failed admission cleanup until provider shutdown permits credential scrub", async () => {
|
||||
const fixture = await hostFixture();
|
||||
let authPath = "";
|
||||
let credentialHome = "";
|
||||
let resolveRetryClose!: () => void;
|
||||
const retryClose = new Promise<void>((resolve) => {
|
||||
resolveRetryClose = resolve;
|
||||
});
|
||||
const runtime = runtimePort({
|
||||
getStatus: async () => ({
|
||||
models: {
|
||||
currentModelId: "wrong-model",
|
||||
availableModelIds: ["wrong-model"],
|
||||
},
|
||||
}),
|
||||
onClose: vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error("first admission cleanup failed"))
|
||||
.mockImplementationOnce(() => retryClose),
|
||||
});
|
||||
|
||||
await expect(
|
||||
AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-all",
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET:
|
||||
'{"owner":"failed-admission"}',
|
||||
},
|
||||
},
|
||||
fixture.dependencies({
|
||||
openRuntime: async (options) => {
|
||||
credentialHome = options.launchEnvironment.CODEX_HOME!;
|
||||
authPath = join(credentialHome, "auth.json");
|
||||
return runtime;
|
||||
},
|
||||
}),
|
||||
),
|
||||
).rejects.toThrow(/initialization and cleanup failed/);
|
||||
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledTimes(2));
|
||||
await expect(readFile(authPath, "utf8")).resolves.toContain(
|
||||
"failed-admission",
|
||||
);
|
||||
await expect(
|
||||
stageManagedCodexCredential({
|
||||
agentHomeDirectory: credentialHome,
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("already has an active lease");
|
||||
|
||||
resolveRetryClose();
|
||||
await vi.waitFor(async () => {
|
||||
await expect(readFile(authPath)).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
});
|
||||
const contender = await stageManagedCodexCredential({
|
||||
agentHomeDirectory: credentialHome,
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
});
|
||||
await contender.close();
|
||||
});
|
||||
|
||||
it("retains credential ownership when runtime shutdown fails until retry succeeds", async () => {
|
||||
const fixture = await hostFixture();
|
||||
let failClose = true;
|
||||
|
|
@ -402,6 +534,70 @@ describe("ACPX runtime host", () => {
|
|||
await contender.close();
|
||||
});
|
||||
|
||||
it("retains the exact pending cleanup while independent resources close", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const firstClose = new Promise<void>(() => undefined);
|
||||
const runtime = runtimePort({
|
||||
onClose: vi
|
||||
.fn()
|
||||
.mockImplementationOnce(() => firstClose)
|
||||
.mockResolvedValueOnce(undefined),
|
||||
});
|
||||
const host = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-all",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => runtime }),
|
||||
);
|
||||
|
||||
const first = host.close({ reason: "first close stalls" });
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledOnce());
|
||||
const second = host.close({ reason: "same pending owner" });
|
||||
let settled = false;
|
||||
void Promise.all([first, second]).finally(() => {
|
||||
settled = true;
|
||||
});
|
||||
await Promise.resolve();
|
||||
expect(settled).toBe(false);
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("retries only after the exact close outcome settles with failure", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const runtime = runtimePort({
|
||||
onClose: vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error("runtime close failed"))
|
||||
.mockResolvedValueOnce(undefined),
|
||||
});
|
||||
const host = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-all",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => runtime }),
|
||||
);
|
||||
|
||||
await expect(host.close({ reason: "first close" })).rejects.toThrow(
|
||||
/cleanup failed/,
|
||||
);
|
||||
await expect(
|
||||
host.close({ reason: "fresh attempt" }),
|
||||
).resolves.toBeUndefined();
|
||||
await expect(
|
||||
host.close({ reason: "ownership released" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(runtime.close).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("admits one bounded turn and cancels it before shutdown", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const turn = runtimeTurn();
|
||||
|
|
@ -429,6 +625,9 @@ describe("ACPX runtime host", () => {
|
|||
host.startTurn({ text: "Concurrent", requestId: "turn-2" }),
|
||||
).toThrow("already has an active turn");
|
||||
|
||||
await host.interruptActiveTurn("user interrupt");
|
||||
expect(turn.cancel).toHaveBeenCalledWith({ reason: "user interrupt" });
|
||||
|
||||
await host.close({ reason: "shutdown" });
|
||||
expect(turn.cancel).toHaveBeenCalledWith({ reason: "shutdown" });
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
|
|
@ -693,13 +892,23 @@ describe("ACPX runtime host", () => {
|
|||
expect(fixture.commandClose).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("forwards cancellation and closes a runtime that resolves after abort", async () => {
|
||||
it("retains managed credentials until an aborted late runtime is closed", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const runtimeAdmission = deferred<AcpxRuntimePort>();
|
||||
const lateRuntime = runtimePort();
|
||||
const retryClose = deferred<void>();
|
||||
const lateRuntime = runtimePort({
|
||||
onClose: vi
|
||||
.fn()
|
||||
.mockRejectedValueOnce(new Error("late runtime close failed"))
|
||||
.mockImplementationOnce(() => retryClose.promise),
|
||||
});
|
||||
let receivedSignal: AbortSignal | undefined;
|
||||
let credentialHome = "";
|
||||
let bridgeUrl = "";
|
||||
const openRuntime = vi.fn((options) => {
|
||||
receivedSignal = options.signal;
|
||||
credentialHome = options.launchEnvironment.CODEX_HOME!;
|
||||
bridgeUrl = options.mcpServers[0]!.url;
|
||||
return runtimeAdmission.promise;
|
||||
});
|
||||
const controller = new AbortController();
|
||||
|
|
@ -714,6 +923,10 @@ describe("ACPX runtime host", () => {
|
|||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}",
|
||||
},
|
||||
signal: controller.signal,
|
||||
semanticTools: {
|
||||
tools: [],
|
||||
handler: async () => ({ ok: true }),
|
||||
},
|
||||
},
|
||||
fixture.dependencies({ openRuntime }),
|
||||
);
|
||||
|
|
@ -723,13 +936,93 @@ describe("ACPX runtime host", () => {
|
|||
controller.abort(cancellation);
|
||||
await expect(opening).rejects.toBe(cancellation);
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
runtimeAdmission.resolve(lateRuntime);
|
||||
await expect(fetch(bridgeUrl)).rejects.toThrow();
|
||||
const authPath = join(credentialHome, "auth.json");
|
||||
await expect(readFile(authPath, "utf8")).resolves.toBe("{}");
|
||||
await expect(
|
||||
stageManagedCodexCredential({
|
||||
agentHomeDirectory: credentialHome,
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("already has an active lease");
|
||||
|
||||
await vi.waitFor(() =>
|
||||
expect(lateRuntime.close).toHaveBeenCalledWith({
|
||||
reason: "ACPX runtime admission aborted",
|
||||
runtimeAdmission.resolve(lateRuntime);
|
||||
await vi.waitFor(() => expect(lateRuntime.close).toHaveBeenCalledTimes(2));
|
||||
expect(lateRuntime.close).toHaveBeenNthCalledWith(1, {
|
||||
reason: "ACPX runtime admission aborted",
|
||||
});
|
||||
await expect(readFile(authPath, "utf8")).resolves.toBe("{}");
|
||||
await expect(
|
||||
stageManagedCodexCredential({
|
||||
agentHomeDirectory: credentialHome,
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("already has an active lease");
|
||||
|
||||
retryClose.resolve(undefined);
|
||||
await vi.waitFor(async () => {
|
||||
await expect(readFile(authPath)).rejects.toMatchObject({
|
||||
code: "ENOENT",
|
||||
});
|
||||
});
|
||||
// File removal precedes kernel lease release. Wait for the lease itself so
|
||||
// this assertion cannot race between those two ordered cleanup steps.
|
||||
const contender = await vi.waitFor(() =>
|
||||
stageManagedCodexCredential({
|
||||
agentHomeDirectory: credentialHome,
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
}),
|
||||
);
|
||||
await contender.close();
|
||||
});
|
||||
|
||||
it("scrubs credentials after rejected runtime cleanup is proven", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const runtimeAdmission = deferred<AcpxRuntimePort>();
|
||||
const providerCleanup = deferred<void>();
|
||||
const credentialClose = vi.fn(async () => undefined);
|
||||
const controller = new AbortController();
|
||||
const cancellation = new Error("runtime admission cancelled");
|
||||
const openRuntime = vi.fn((options: AcpxRuntimePortOpenOptions) => {
|
||||
options.retainFailedAdmissionCleanup(providerCleanup.promise);
|
||||
return runtimeAdmission.promise;
|
||||
});
|
||||
const opening = AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "deny-all",
|
||||
signal: controller.signal,
|
||||
},
|
||||
{
|
||||
...fixture.dependencies({ openRuntime }),
|
||||
stageCredential: async () => ({
|
||||
path: join(fixture.root, "auth.json"),
|
||||
mode: "inline_json",
|
||||
close: credentialClose,
|
||||
}),
|
||||
},
|
||||
);
|
||||
await vi.waitFor(() => expect(openRuntime).toHaveBeenCalledOnce());
|
||||
|
||||
controller.abort(cancellation);
|
||||
await expect(opening).rejects.toBe(cancellation);
|
||||
const cleanupFailure = new Error("provider survived forced cleanup");
|
||||
runtimeAdmission.reject(new AggregateError([cancellation, cleanupFailure]));
|
||||
await new Promise((resolve) => setImmediate(resolve));
|
||||
|
||||
expect(credentialClose).not.toHaveBeenCalled();
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
|
||||
providerCleanup.resolve(undefined);
|
||||
await vi.waitFor(() => expect(credentialClose).toHaveBeenCalledOnce());
|
||||
});
|
||||
});
|
||||
|
||||
|
|
@ -824,10 +1117,13 @@ async function hostFixture() {
|
|||
function deferred<T>(): {
|
||||
promise: Promise<T>;
|
||||
resolve(value: T): void;
|
||||
reject(reason: unknown): void;
|
||||
} {
|
||||
let resolve!: (value: T) => void;
|
||||
const promise = new Promise<T>((settle) => {
|
||||
let reject!: (reason: unknown) => void;
|
||||
const promise = new Promise<T>((settle, fail) => {
|
||||
resolve = settle;
|
||||
reject = fail;
|
||||
});
|
||||
return { promise, resolve };
|
||||
return { promise, resolve, reject };
|
||||
}
|
||||
|
|
|
|||
|
|
@ -38,7 +38,11 @@ import {
|
|||
} from "./runtime-sandbox.js";
|
||||
import type { AcpxExpectedSessionIdentity } from "./sidecar-protocol.js";
|
||||
|
||||
const TURN_CANCELLATION_TIMEOUT_MS = 2_000;
|
||||
export const ACPX_TURN_CANCELLATION_SHUTDOWN_BOUND_MS = 2_000;
|
||||
|
||||
const ACPX_ADMISSION_CLEANUP_BATCH_ATTEMPTS = 8;
|
||||
const ACPX_ADMISSION_CLEANUP_RETRY_DELAY_MS = 10;
|
||||
const ACPX_ADMISSION_CLEANUP_RESCHEDULE_MS = 1_000;
|
||||
|
||||
export interface AcpxRuntimePortIdentity {
|
||||
acpxRecordId: string;
|
||||
|
|
@ -83,6 +87,11 @@ export interface AcpxRuntimePortOpenOptions {
|
|||
/** Abort provider admission and clean any runtime that resolves too late. */
|
||||
signal?: AbortSignal;
|
||||
mcpServers: readonly AcpxMcpServerBinding[];
|
||||
/**
|
||||
* Transfer the provider cleanup proof before a failed open settles. The host
|
||||
* keeps credentials fenced until this exact cleanup succeeds.
|
||||
*/
|
||||
retainFailedAdmissionCleanup(cleanup: Promise<void>): void;
|
||||
}
|
||||
|
||||
export interface AcpxRetainedCleanupFailure {
|
||||
|
|
@ -134,6 +143,35 @@ const activeRuntimeHostCleanupOwners = new Set<Promise<unknown>>();
|
|||
const RETAINED_CLEANUP_RETRY_INITIAL_DELAY_MS = 10;
|
||||
const RETAINED_CLEANUP_RETRY_MAX_DELAY_MS = 1_000;
|
||||
|
||||
interface RetainedRejectedRuntimeAdmission {
|
||||
readonly credential: ManagedCodexCredentialLease;
|
||||
cleanup: Promise<void>;
|
||||
}
|
||||
|
||||
// A rejected provider open is not itself proof that provider processes are
|
||||
// gone. Retain the credential with the adapter's exact cleanup proof and scrub
|
||||
// it only after that proof succeeds. Terminal cleanup failure stays inert and
|
||||
// fenced instead of exposing the credential to a replacement admission.
|
||||
const retainedRejectedRuntimeAdmissions =
|
||||
new Set<RetainedRejectedRuntimeAdmission>();
|
||||
|
||||
interface RetainedAcpxAdmissionCleanup {
|
||||
readonly runtime: AcpxRuntimePort | null;
|
||||
readonly toolBridge: RunnerToolBridge | null;
|
||||
readonly credential: ManagedCodexCredentialLease | null;
|
||||
readonly command: VerifiedAcpxCommandLease | null;
|
||||
readonly reason: string;
|
||||
recovery: Promise<void> | null;
|
||||
timer: ReturnType<typeof setTimeout> | null;
|
||||
}
|
||||
|
||||
// An admission failure has no AcpxRuntimeHost instance for a caller to close.
|
||||
// Retain those resources here until provider shutdown succeeds and the staged
|
||||
// credential can consequently be scrubbed. Each recovery batch is finite and
|
||||
// sequential; an unref'd timer rate-limits later autonomous attempts without
|
||||
// allowing the failed admission's ownership to become unreachable.
|
||||
const retainedAcpxAdmissionCleanups = new Set<RetainedAcpxAdmissionCleanup>();
|
||||
|
||||
export class AcpxRuntimeHost {
|
||||
readonly #runtime: AcpxRuntimePort;
|
||||
readonly #binding: AcpxRecoveryBinding;
|
||||
|
|
@ -205,6 +243,37 @@ export class AcpxRuntimeHost {
|
|||
let credential: ManagedCodexCredentialLease | null = null;
|
||||
let toolBridge: RunnerToolBridge | null = null;
|
||||
let runtime: AcpxRuntimePort | null = null;
|
||||
let pendingRuntimeOwnsCredential = false;
|
||||
let failedAdmissionCleanupTransferred = false;
|
||||
let resolveFailedAdmissionCleanupTransfer!: () => void;
|
||||
const failedAdmissionCleanupTransfer = new Promise<void>((resolve) => {
|
||||
resolveFailedAdmissionCleanupTransfer = resolve;
|
||||
});
|
||||
const retainFailedAdmissionCleanup = (cleanup: Promise<void>): void => {
|
||||
if (failedAdmissionCleanupTransferred) return;
|
||||
failedAdmissionCleanupTransferred = true;
|
||||
pendingRuntimeOwnsCredential = credential !== null;
|
||||
resolveFailedAdmissionCleanupTransfer();
|
||||
if (credential === null) {
|
||||
retainRuntimeHostCleanup(cleanup);
|
||||
return;
|
||||
}
|
||||
const retained: RetainedRejectedRuntimeAdmission = {
|
||||
credential,
|
||||
cleanup: Promise.resolve(),
|
||||
};
|
||||
const ownedCleanup = cleanup.then(async () => {
|
||||
await cleanupAbortedRuntimeAdmission(
|
||||
null,
|
||||
retained.credential,
|
||||
"ACPX rejected runtime admission cleanup confirmed",
|
||||
);
|
||||
retainedRejectedRuntimeAdmissions.delete(retained);
|
||||
});
|
||||
retained.cleanup = ownedCleanup;
|
||||
retainedRejectedRuntimeAdmissions.add(retained);
|
||||
retainRuntimeHostCleanup(ownedCleanup);
|
||||
};
|
||||
try {
|
||||
const sandbox = await runAbortableAdmissionStage(options.signal, () =>
|
||||
prepareAcpxRuntimeSandbox({
|
||||
|
|
@ -272,6 +341,7 @@ export class AcpxRuntimeHost {
|
|||
},
|
||||
]
|
||||
: [],
|
||||
retainFailedAdmissionCleanup,
|
||||
}),
|
||||
resource: "runtime",
|
||||
releaseLate: (lateRuntime) =>
|
||||
|
|
@ -280,6 +350,19 @@ export class AcpxRuntimeHost {
|
|||
}),
|
||||
reportFailure: (failure) =>
|
||||
dependencies.reportRetainedCleanupFailure(failure),
|
||||
onAbortedPending: (pendingRuntime) => {
|
||||
// A provider may already be running even though openRuntime has not
|
||||
// returned its port. Keep its credential lease with that exact
|
||||
// admission while the ordinary catch path revokes tools and releases
|
||||
// the consumed command snapshot without delaying cancellation.
|
||||
pendingRuntimeOwnsCredential = credential !== null;
|
||||
retainAbortedRuntimeAdmissionCleanup({
|
||||
pendingRuntime,
|
||||
credential,
|
||||
reason: "ACPX runtime admission aborted",
|
||||
failedAdmissionCleanupTransfer,
|
||||
});
|
||||
},
|
||||
});
|
||||
await runAbortableAdmissionStage(options.signal, () =>
|
||||
requireVerifiedAcpxModel(runtime!, profile),
|
||||
|
|
@ -316,11 +399,18 @@ export class AcpxRuntimeHost {
|
|||
const cleanupError = await cleanupRuntimeResources(
|
||||
runtime,
|
||||
toolBridge,
|
||||
credential,
|
||||
pendingRuntimeOwnsCredential ? null : credential,
|
||||
command,
|
||||
"ACPX runtime initialization failed",
|
||||
);
|
||||
if (cleanupError) {
|
||||
retainFailedAcpxAdmissionCleanup({
|
||||
runtime,
|
||||
toolBridge,
|
||||
credential: pendingRuntimeOwnsCredential ? null : credential,
|
||||
command,
|
||||
reason: "ACPX runtime initialization failed",
|
||||
});
|
||||
throw new AggregateError(
|
||||
[error, ...cleanupError.errors],
|
||||
"ACPX runtime initialization and cleanup failed",
|
||||
|
|
@ -346,6 +436,10 @@ export class AcpxRuntimeHost {
|
|||
return Object.freeze({ ...this.#sandbox.persistedEnvironment });
|
||||
}
|
||||
|
||||
async status(): Promise<AcpxModelStatus> {
|
||||
return structuredClone(await this.#runtime.getStatus());
|
||||
}
|
||||
|
||||
startTurn(input: AcpxRuntimeTurnInput): AcpxRuntimeTurn {
|
||||
if (this.#closed || this.#closingStarted) {
|
||||
throw new Error("ACPX runtime host is closing");
|
||||
|
|
@ -374,9 +468,20 @@ export class AcpxRuntimeHost {
|
|||
return turn;
|
||||
}
|
||||
|
||||
async interruptActiveTurn(reason: string): Promise<void> {
|
||||
const turn = this.#activeTurn;
|
||||
if (!turn) throw new Error("ACPX runtime host has no active turn");
|
||||
const cancellationError = await boundedCancellation(
|
||||
turn.cancel({ reason: boundedReason(reason) }),
|
||||
);
|
||||
if (cancellationError) throw cancellationError;
|
||||
}
|
||||
|
||||
async close(input: { reason: string }): Promise<void> {
|
||||
if (this.#closed) return;
|
||||
if (this.#closePromise) return await this.#closePromise;
|
||||
if (this.#closePromise) {
|
||||
return await this.#closePromise;
|
||||
}
|
||||
this.#closingStarted = true;
|
||||
const closePromise = this.#close(boundedReason(input.reason));
|
||||
this.#closePromise = closePromise;
|
||||
|
|
@ -435,6 +540,7 @@ async function acquireAbortableAdmissionResource<T>(input: {
|
|||
resource: AcpxRetainedCleanupFailure["resource"];
|
||||
releaseLate: (resource: T) => Promise<void>;
|
||||
reportFailure: (failure: AcpxRetainedCleanupFailure) => void;
|
||||
onAbortedPending?: (pending: Promise<T>) => void;
|
||||
}): Promise<T> {
|
||||
if (input.signal === undefined) return await input.acquire();
|
||||
input.signal.throwIfAborted();
|
||||
|
|
@ -443,16 +549,20 @@ async function acquireAbortableAdmissionResource<T>(input: {
|
|||
return await raceAdmissionWithAbort(pending, input.signal);
|
||||
} catch (error) {
|
||||
if (input.signal.aborted) {
|
||||
retainRuntimeHostCleanup(
|
||||
pending.then((resource) =>
|
||||
releaseRetainedAdmissionResource({
|
||||
resource,
|
||||
resourceKind: input.resource,
|
||||
release: input.releaseLate,
|
||||
reportFailure: input.reportFailure,
|
||||
}),
|
||||
),
|
||||
);
|
||||
if (input.onAbortedPending) {
|
||||
input.onAbortedPending(pending);
|
||||
} else {
|
||||
retainRuntimeHostCleanup(
|
||||
pending.then((resource) =>
|
||||
releaseRetainedAdmissionResource({
|
||||
resource,
|
||||
resourceKind: input.resource,
|
||||
release: input.releaseLate,
|
||||
reportFailure: input.reportFailure,
|
||||
}),
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
|
|
@ -530,6 +640,123 @@ function retainRuntimeHostCleanup(cleanup: Promise<unknown>): void {
|
|||
.catch(() => undefined);
|
||||
}
|
||||
|
||||
function retainAbortedRuntimeAdmissionCleanup(input: {
|
||||
pendingRuntime: Promise<AcpxRuntimePort>;
|
||||
credential: ManagedCodexCredentialLease | null;
|
||||
reason: string;
|
||||
failedAdmissionCleanupTransfer: Promise<void>;
|
||||
}): void {
|
||||
const cleanup = input.pendingRuntime.then(
|
||||
(runtime) =>
|
||||
cleanupAbortedRuntimeAdmission(runtime, input.credential, input.reason),
|
||||
() => input.failedAdmissionCleanupTransfer,
|
||||
);
|
||||
retainRuntimeHostCleanup(cleanup);
|
||||
}
|
||||
|
||||
async function cleanupAbortedRuntimeAdmission(
|
||||
runtime: AcpxRuntimePort | null,
|
||||
credential: ManagedCodexCredentialLease | null,
|
||||
reason: string,
|
||||
): Promise<void> {
|
||||
const cleanupError = await cleanupRuntimeResources(
|
||||
runtime,
|
||||
null,
|
||||
credential,
|
||||
null,
|
||||
reason,
|
||||
);
|
||||
if (!cleanupError) return;
|
||||
retainFailedAcpxAdmissionCleanup({
|
||||
runtime,
|
||||
toolBridge: null,
|
||||
credential,
|
||||
command: null,
|
||||
reason,
|
||||
});
|
||||
}
|
||||
|
||||
function retainFailedAcpxAdmissionCleanup(input: {
|
||||
runtime: AcpxRuntimePort | null;
|
||||
toolBridge: RunnerToolBridge | null;
|
||||
credential: ManagedCodexCredentialLease | null;
|
||||
command: VerifiedAcpxCommandLease | null;
|
||||
reason: string;
|
||||
}): void {
|
||||
const cleanup: RetainedAcpxAdmissionCleanup = {
|
||||
...input,
|
||||
recovery: null,
|
||||
timer: null,
|
||||
};
|
||||
retainedAcpxAdmissionCleanups.add(cleanup);
|
||||
startRetainedAcpxAdmissionCleanup(cleanup);
|
||||
}
|
||||
|
||||
function startRetainedAcpxAdmissionCleanup(
|
||||
cleanup: RetainedAcpxAdmissionCleanup,
|
||||
): Promise<void> {
|
||||
if (cleanup.recovery) return cleanup.recovery;
|
||||
const recovery = (async () => {
|
||||
let retryDelayMs = ACPX_ADMISSION_CLEANUP_RETRY_DELAY_MS;
|
||||
for (
|
||||
let attempt = 1;
|
||||
attempt <= ACPX_ADMISSION_CLEANUP_BATCH_ATTEMPTS;
|
||||
attempt += 1
|
||||
) {
|
||||
const cleanupError = await cleanupRuntimeResources(
|
||||
cleanup.runtime,
|
||||
cleanup.toolBridge,
|
||||
cleanup.credential,
|
||||
cleanup.command,
|
||||
`${cleanup.reason} (automatic cleanup recovery ${attempt})`,
|
||||
);
|
||||
if (!cleanupError) {
|
||||
retainedAcpxAdmissionCleanups.delete(cleanup);
|
||||
if (cleanup.timer) clearTimeout(cleanup.timer);
|
||||
cleanup.timer = null;
|
||||
return;
|
||||
}
|
||||
if (attempt < ACPX_ADMISSION_CLEANUP_BATCH_ATTEMPTS) {
|
||||
await waitForAdmissionCleanupRetry(retryDelayMs);
|
||||
retryDelayMs = Math.min(retryDelayMs * 2, 1_000);
|
||||
}
|
||||
}
|
||||
})();
|
||||
cleanup.recovery = recovery;
|
||||
void recovery
|
||||
.finally(() => {
|
||||
if (cleanup.recovery === recovery) cleanup.recovery = null;
|
||||
scheduleRetainedAcpxAdmissionCleanup(cleanup);
|
||||
})
|
||||
.catch(() => undefined);
|
||||
return recovery;
|
||||
}
|
||||
|
||||
function scheduleRetainedAcpxAdmissionCleanup(
|
||||
cleanup: RetainedAcpxAdmissionCleanup,
|
||||
): void {
|
||||
if (
|
||||
!retainedAcpxAdmissionCleanups.has(cleanup) ||
|
||||
cleanup.recovery ||
|
||||
cleanup.timer
|
||||
) {
|
||||
return;
|
||||
}
|
||||
cleanup.timer = setTimeout(() => {
|
||||
cleanup.timer = null;
|
||||
if (!retainedAcpxAdmissionCleanups.has(cleanup)) return;
|
||||
startRetainedAcpxAdmissionCleanup(cleanup);
|
||||
}, ACPX_ADMISSION_CLEANUP_RESCHEDULE_MS);
|
||||
cleanup.timer.unref?.();
|
||||
}
|
||||
|
||||
async function waitForAdmissionCleanupRetry(delayMs: number): Promise<void> {
|
||||
await new Promise<void>((resolve) => {
|
||||
const timer = setTimeout(resolve, delayMs);
|
||||
timer.unref?.();
|
||||
});
|
||||
}
|
||||
|
||||
async function boundedCancellation(
|
||||
cancellation: Promise<void>,
|
||||
): Promise<unknown | null> {
|
||||
|
|
@ -547,7 +774,7 @@ async function boundedCancellation(
|
|||
"ACPX turn cancellation exceeded its shutdown timeout",
|
||||
),
|
||||
}),
|
||||
TURN_CANCELLATION_TIMEOUT_MS,
|
||||
ACPX_TURN_CANCELLATION_SHUTDOWN_BOUND_MS,
|
||||
);
|
||||
}),
|
||||
]);
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue