feat(runner): add bounded ACPX turn lifecycle (#12403)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - The admitted Codex ACPX runtime can open and recover a verified session. > - It cannot yet accept a provider prompt through the narrow host port. > - Turn admission must bound durable identity and prompt payloads. > - Shutdown must cancel an active turn and release runtime resources in a bounded way. > - A cancellation timeout must not leave the runtime, command lease, or staged credentials alive indefinitely. > - This pull request adds the package-local turn lifecycle and its cleanup rules without production wiring. > - The benefit is one explicit and testable prompt boundary for the later harness driver. ## Linked Issues or Issue Description Refs #12402 **What would you like to improve?** The package-local ACPX host stops at session admission. A later driver needs to submit a prompt, consume typed ACP events, wait for the terminal result, cancel work, and close the session. Passing the full third-party runtime through the host would bypass the existing trust and cleanup boundary. **Why is this important?** Provider prompts can be large. Turn identifiers participate in durable correlation. Concurrent turns can make replay and cancellation ambiguous. Shutdown must also stop an active prompt before credentials and verified command resources are released. A provider that does not finish cancellation must not block all remaining cleanup indefinitely. **Suggested approach** Add a minimal turn interface to the admitted runtime port. Accept one prompt turn at a time. Bound the request identity and text before the runtime sees them. Map the call to ACPX prompt mode with the admitted session handle. Track the active turn and request cancellation before ordered runtime cleanup. Bound the cancellation wait. Continue runtime and command cleanup after that timeout. Release staged credentials only after the exact runtime close succeeds. **Additional context** This pull request builds on #12402. It does not attach semantic tools, normalize provider events, create a harness driver, start runnerd, register production execution, or change server, UI, or direct-adapter behavior. ## What Changed - Add a narrow ACPX turn input and result and event lifecycle to the admitted runtime port. - Map prompt turns to the exact persistent ACPX session handle. - Support abort signals without adding steering or attachments. - Reject empty, whitespace-normalized, or oversized request identities. - Reject prompt text larger than one MiB before third-party code executes. - Permit only one active turn per host. - Clear the active turn only after the canonical ACPX result settles. - Reject new turns as soon as shutdown starts. - Cancel an active turn before runtime, credential, and command cleanup. - Bound the cancellation wait to two seconds. - Continue runtime and command cleanup when turn cancellation fails or reaches its timeout. - Release staged credentials only after the exact runtime close succeeds. - Keep the cancellation handle and credential lease when runtime cleanup remains retryable. - Coalesce concurrent close calls and report all cleanup failures in one aggregate error. - Add focused host and adapter tests for turn mapping, bounds, concurrency, cancellation, timeout cleanup, credential retention, and late-turn rejection. ## Verification - Exact corrected head: `57e1edfcfc496bd9688c1ecf22f2d402c6bb2079`. - The pull request delta contains four files: - `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` - `git diff --check` passed for the exact corrected delta. - This delta does not change dependencies, `pnpm-lock.yaml`, workflows, migrations, server selection, UI behavior, or production runner wiring. - Full GitHub PR workflow passed in [run 33343457544](https://github.com/paperclipai/paperclip/actions/runs/33343457544): 28 successful checks, including runner verification/build, typecheck, all test shards, canary, and e2e; Storybook skipped by path as expected. - Greptile is 5/5 on the exact corrected head with no blocking failure and zero unresolved review threads. - Superagent Security, Snyk, contributor trust, and commitperclip passed on the exact corrected head. - No local test result is claimed. GitHub Actions is the authoritative verification environment for this revision. ## Risks The primary risk is ambiguous concurrent execution. The host admits only one active turn and releases that slot from the canonical ACPX terminal result. Another risk is partial shutdown. The host requests cancellation first and waits for at most two seconds. It then attempts runtime and command cleanup even if cancellation fails or reaches the timeout. It releases staged credentials only after the exact runtime close succeeds. If runtime cleanup fails, the host keeps the cancellation handle and credentials for a later cleanup attempt. This pull request does not register the runtime for production use. ## Model Used OpenAI Codex with GPT-5 and repository tool use. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [ ] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Dotta
GitHub
parent
9036d3c484
commit
57449579ca
|
|
@ -248,6 +248,36 @@ describe("Codex ACPX runtime adapter", () => {
|
|||
});
|
||||
});
|
||||
|
||||
it("maps prompt turns to the admitted ACPX handle", async () => {
|
||||
const runtime = fakeRuntime();
|
||||
const turn = {
|
||||
requestId: "turn-1",
|
||||
promptStarted: Promise.resolve(),
|
||||
events: { async *[Symbol.asyncIterator]() {} },
|
||||
result: Promise.resolve({ status: "completed" as const }),
|
||||
cancel: vi.fn(),
|
||||
closeStream: vi.fn(),
|
||||
};
|
||||
vi.mocked(runtime.startTurn).mockReturnValue(turn);
|
||||
const port = await openCodexAcpxRuntime(openOptions(fakeCommand()), {
|
||||
createRegistry: () => registry(),
|
||||
createStore: () => store(),
|
||||
createRuntime: () => runtime,
|
||||
});
|
||||
const signal = new AbortController().signal;
|
||||
|
||||
expect(
|
||||
port.startTurn({ text: "Complete the task.", requestId: "turn-1", signal }),
|
||||
).toBe(turn);
|
||||
expect(runtime.startTurn).toHaveBeenCalledWith({
|
||||
handle: HANDLE,
|
||||
text: "Complete the task.",
|
||||
mode: "prompt",
|
||||
requestId: "turn-1",
|
||||
signal,
|
||||
});
|
||||
});
|
||||
|
||||
it("fails closed and closes the session when ACPX omits recovery identity", async () => {
|
||||
const runtime = fakeRuntime({ ...HANDLE, agentSessionId: undefined });
|
||||
await expect(
|
||||
|
|
|
|||
|
|
@ -585,6 +585,15 @@ function runtimePort(
|
|||
},
|
||||
}
|
||||
: {}),
|
||||
startTurn(input) {
|
||||
return runtime.startTurn({
|
||||
handle,
|
||||
text: input.text,
|
||||
mode: "prompt",
|
||||
requestId: input.requestId,
|
||||
...(input.signal ? { signal: input.signal } : {}),
|
||||
});
|
||||
},
|
||||
async close(input) {
|
||||
const errors = await admissionCleanup.run(handle, input.reason);
|
||||
if (errors.length > 0) {
|
||||
|
|
|
|||
|
|
@ -4,6 +4,7 @@ import { join } from "node:path";
|
|||
|
||||
import { afterEach, describe, expect, it, vi } from "vitest";
|
||||
|
||||
import { stageManagedCodexCredential } from "./codex-credentials.js";
|
||||
import type {
|
||||
VerifiedAcpxCommandLease,
|
||||
VerifiedAcpxInstallation,
|
||||
|
|
@ -13,6 +14,7 @@ import {
|
|||
AcpxRuntimeHost,
|
||||
type AcpxRuntimeHostDependencies,
|
||||
type AcpxRuntimePort,
|
||||
type AcpxRuntimeTurn,
|
||||
} from "./runtime-host.js";
|
||||
|
||||
const temporaryDirectories: string[] = [];
|
||||
|
|
@ -227,7 +229,7 @@ describe("ACPX runtime host", () => {
|
|||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
});
|
||||
|
||||
it("attempts every cleanup when runtime shutdown fails", async () => {
|
||||
it("retains credential ownership when runtime shutdown fails until retry succeeds", async () => {
|
||||
const fixture = await hostFixture();
|
||||
let failClose = true;
|
||||
const runtime = runtimePort({
|
||||
|
|
@ -252,12 +254,263 @@ describe("ACPX runtime host", () => {
|
|||
await expect(host.close({ reason: "first close" })).rejects.toThrow(
|
||||
/cleanup failed/,
|
||||
);
|
||||
await expect(readFile(authPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
await expect(readFile(authPath, "utf8")).resolves.toBe("{}");
|
||||
await expect(
|
||||
stageManagedCodexCredential({
|
||||
agentHomeDirectory: join(host.runtimeRoot(), "codex-home"),
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
}),
|
||||
).rejects.toThrow("already has an active lease");
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
failClose = false;
|
||||
await expect(
|
||||
host.close({ reason: "retry close" }),
|
||||
).resolves.toBeUndefined();
|
||||
await expect(readFile(authPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
const contender = await stageManagedCodexCredential({
|
||||
agentHomeDirectory: join(host.runtimeRoot(), "codex-home"),
|
||||
environment: {
|
||||
PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: '{"owner":"contender"}',
|
||||
},
|
||||
});
|
||||
await contender.close();
|
||||
});
|
||||
|
||||
it("scrubs credentials only after the exact pending runtime close resolves", async () => {
|
||||
const fixture = await hostFixture();
|
||||
let resolveRuntimeClose!: () => void;
|
||||
const runtimeClose = new Promise<void>((resolve) => {
|
||||
resolveRuntimeClose = resolve;
|
||||
});
|
||||
const runtime = runtimePort({
|
||||
onClose: vi.fn(() => runtimeClose),
|
||||
});
|
||||
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 credentialHome = join(host.runtimeRoot(), "codex-home");
|
||||
const authPath = join(credentialHome, "auth.json");
|
||||
|
||||
const first = host.close({ reason: "runtime close pending" });
|
||||
await vi.waitFor(() => expect(runtime.close).toHaveBeenCalledOnce());
|
||||
await vi.waitFor(() => expect(fixture.commandClose).toHaveBeenCalledOnce());
|
||||
const second = host.close({ reason: "same exact close" });
|
||||
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");
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
|
||||
resolveRuntimeClose();
|
||||
await expect(Promise.all([first, second])).resolves.toEqual([
|
||||
undefined,
|
||||
undefined,
|
||||
]);
|
||||
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("admits one bounded turn and cancels it before shutdown", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const turn = runtimeTurn();
|
||||
const startTurn = vi.fn(() => turn);
|
||||
const runtime = runtimePort({ startTurn });
|
||||
const host = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-reads",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => runtime }),
|
||||
);
|
||||
|
||||
expect(
|
||||
host.startTurn({ text: "Complete the task.", requestId: "turn-1" }),
|
||||
).toBe(turn);
|
||||
expect(startTurn).toHaveBeenCalledWith({
|
||||
text: "Complete the task.",
|
||||
requestId: "turn-1",
|
||||
});
|
||||
expect(() =>
|
||||
host.startTurn({ text: "Concurrent", requestId: "turn-2" }),
|
||||
).toThrow("already has an active turn");
|
||||
|
||||
await host.close({ reason: "shutdown" });
|
||||
expect(turn.cancel).toHaveBeenCalledWith({ reason: "shutdown" });
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
expect(() => host.startTurn({ text: "Late", requestId: "turn-3" })).toThrow(
|
||||
"is closing",
|
||||
);
|
||||
});
|
||||
|
||||
it("rejects oversized turn inputs before calling the runtime", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const runtime = runtimePort();
|
||||
const host = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-reads",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => runtime }),
|
||||
);
|
||||
|
||||
expect(() => host.startTurn({ text: "ok", requestId: " " })).toThrow(
|
||||
"request id",
|
||||
);
|
||||
expect(() =>
|
||||
host.startTurn({ text: "x".repeat(1024 * 1024 + 1), requestId: "turn" }),
|
||||
).toThrow("turn text");
|
||||
expect(runtime.startTurn).not.toHaveBeenCalled();
|
||||
await host.close({ reason: "complete" });
|
||||
});
|
||||
|
||||
it("cleans runtime resources after a turn cancellation timeout", async () => {
|
||||
vi.useFakeTimers();
|
||||
try {
|
||||
const fixture = await hostFixture();
|
||||
const turn = runtimeTurn();
|
||||
turn.cancel.mockImplementation(() => new Promise(() => undefined));
|
||||
const runtime = runtimePort({ startTurn: () => turn });
|
||||
const host = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-reads",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => runtime }),
|
||||
);
|
||||
host.startTurn({ text: "Complete the task.", requestId: "turn-1" });
|
||||
|
||||
const closing = host.close({ reason: "shutdown" });
|
||||
await vi.advanceTimersByTimeAsync(2_000);
|
||||
await expect(closing).rejects.toThrow(/cleanup failed/);
|
||||
expect(turn.cancel).toHaveBeenCalledOnce();
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
await expect(
|
||||
host.close({ reason: "observe completed cleanup" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(turn.cancel).toHaveBeenCalledOnce();
|
||||
expect(runtime.close).toHaveBeenCalledOnce();
|
||||
expect(fixture.commandClose).toHaveBeenCalledOnce();
|
||||
} finally {
|
||||
vi.useRealTimers();
|
||||
}
|
||||
});
|
||||
|
||||
it("retains the cancellation handle while runtime cleanup remains retryable", async () => {
|
||||
const fixture = await hostFixture();
|
||||
const turn = runtimeTurn();
|
||||
let failRuntimeClose = true;
|
||||
const runtime = runtimePort({
|
||||
startTurn: () => turn,
|
||||
onClose: async () => {
|
||||
if (failRuntimeClose) throw new Error("runtime cleanup failed");
|
||||
},
|
||||
});
|
||||
const host = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-reads",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => runtime }),
|
||||
);
|
||||
host.startTurn({ text: "Complete the task.", requestId: "turn-1" });
|
||||
const credentialHome = join(host.runtimeRoot(), "codex-home");
|
||||
const authPath = join(credentialHome, "auth.json");
|
||||
|
||||
await expect(host.close({ reason: "first shutdown" })).rejects.toThrow(
|
||||
/cleanup failed/,
|
||||
);
|
||||
expect(turn.cancel).toHaveBeenCalledOnce();
|
||||
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");
|
||||
|
||||
failRuntimeClose = false;
|
||||
await expect(
|
||||
host.close({ reason: "retry shutdown" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(turn.cancel).toHaveBeenCalledTimes(2);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(2);
|
||||
await expect(readFile(authPath)).rejects.toMatchObject({ code: "ENOENT" });
|
||||
});
|
||||
|
||||
it("retains a settled turn when shutdown cleanup must be retried", async () => {
|
||||
const fixture = await hostFixture();
|
||||
let settleTurn!: (value: { stopReason: string }) => void;
|
||||
const turn = {
|
||||
...runtimeTurn(),
|
||||
result: new Promise<{ stopReason: string }>((resolve) => {
|
||||
settleTurn = resolve;
|
||||
}),
|
||||
};
|
||||
let failRuntimeClose = true;
|
||||
const runtime = runtimePort({
|
||||
startTurn: () => turn,
|
||||
onClose: async () => {
|
||||
if (failRuntimeClose) throw new Error("runtime cleanup failed");
|
||||
},
|
||||
});
|
||||
const host = await AcpxRuntimeHost.open(
|
||||
{
|
||||
...fixture.options,
|
||||
agent: "codex",
|
||||
model: "gpt-5.6-sol",
|
||||
permissionMode: "approve-reads",
|
||||
environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" },
|
||||
},
|
||||
fixture.dependencies({ openRuntime: async () => runtime }),
|
||||
);
|
||||
host.startTurn({ text: "Complete the task.", requestId: "turn-1" });
|
||||
|
||||
const firstClose = host.close({ reason: "first shutdown" });
|
||||
settleTurn({ stopReason: "end_turn" });
|
||||
await expect(firstClose).rejects.toThrow(/cleanup failed/);
|
||||
expect(turn.cancel).toHaveBeenCalledOnce();
|
||||
|
||||
failRuntimeClose = false;
|
||||
await expect(
|
||||
host.close({ reason: "retry shutdown" }),
|
||||
).resolves.toBeUndefined();
|
||||
expect(turn.cancel).toHaveBeenCalledTimes(2);
|
||||
expect(runtime.close).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
it("closes a command lease that resolves after admission is aborted", async () => {
|
||||
|
|
@ -412,6 +665,7 @@ function runtimePort(
|
|||
input: {
|
||||
getStatus?: AcpxRuntimePort["getStatus"];
|
||||
setModel?: NonNullable<AcpxRuntimePort["setModel"]>;
|
||||
startTurn?: AcpxRuntimePort["startTurn"];
|
||||
onClose?: AcpxRuntimePort["close"];
|
||||
} = {},
|
||||
): AcpxRuntimePort & { close: ReturnType<typeof vi.fn> } {
|
||||
|
|
@ -430,10 +684,28 @@ function runtimePort(
|
|||
},
|
||||
})),
|
||||
...(input.setModel ? { setModel: input.setModel } : {}),
|
||||
startTurn: vi.fn(input.startTurn ?? (() => runtimeTurn())),
|
||||
close: vi.fn(input.onClose ?? (async () => undefined)),
|
||||
};
|
||||
}
|
||||
|
||||
function runtimeTurn(): AcpxRuntimeTurn & {
|
||||
cancel: ReturnType<typeof vi.fn>;
|
||||
} {
|
||||
return {
|
||||
requestId: "turn-1",
|
||||
promptStarted: Promise.resolve(),
|
||||
events: {
|
||||
async *[Symbol.asyncIterator]() {
|
||||
yield { type: "text_delta" as const, text: "done" };
|
||||
},
|
||||
},
|
||||
result: new Promise(() => undefined),
|
||||
cancel: vi.fn(async () => undefined),
|
||||
closeStream: vi.fn(async () => undefined),
|
||||
};
|
||||
}
|
||||
|
||||
async function hostFixture() {
|
||||
const root = await mkdtemp(join(tmpdir(), "paperclip-acpx-host-"));
|
||||
temporaryDirectories.push(root);
|
||||
|
|
|
|||
|
|
@ -1,3 +1,5 @@
|
|||
import type { AcpRuntimeEvent, AcpRuntimeTurnResult } from "acpx/runtime";
|
||||
|
||||
import type { NativeAcpxPermissionMode } from "../../contracts/native-execution.js";
|
||||
import {
|
||||
stageManagedCodexCredential,
|
||||
|
|
@ -31,17 +33,35 @@ import {
|
|||
} from "./runtime-sandbox.js";
|
||||
import type { AcpxExpectedSessionIdentity } from "./sidecar-protocol.js";
|
||||
|
||||
const TURN_CANCELLATION_TIMEOUT_MS = 2_000;
|
||||
|
||||
export interface AcpxRuntimePortIdentity {
|
||||
acpxRecordId: string;
|
||||
backendSessionId: string;
|
||||
agentSessionId: string;
|
||||
}
|
||||
|
||||
export interface AcpxRuntimeTurnInput {
|
||||
text: string;
|
||||
requestId: string;
|
||||
signal?: AbortSignal;
|
||||
}
|
||||
|
||||
export interface AcpxRuntimeTurn {
|
||||
readonly requestId: string;
|
||||
readonly promptStarted: Promise<void>;
|
||||
readonly events: AsyncIterable<AcpRuntimeEvent>;
|
||||
readonly result: Promise<AcpRuntimeTurnResult>;
|
||||
cancel(input?: { reason?: string }): Promise<void>;
|
||||
closeStream(input?: { reason?: string }): Promise<void>;
|
||||
}
|
||||
|
||||
/** Minimal third-party ACP runtime surface admitted by the host boundary. */
|
||||
export interface AcpxRuntimePort {
|
||||
identity(): Promise<AcpxRuntimePortIdentity>;
|
||||
getStatus(): Promise<AcpxModelStatus>;
|
||||
setModel?(model: string): Promise<void>;
|
||||
startTurn(input: AcpxRuntimeTurnInput): AcpxRuntimeTurn;
|
||||
close(input: { reason: string }): Promise<void>;
|
||||
}
|
||||
|
||||
|
|
@ -105,6 +125,9 @@ export class AcpxRuntimeHost {
|
|||
readonly #sandbox: AcpxRuntimeSandbox;
|
||||
readonly #credential: ManagedCodexCredentialLease | null;
|
||||
readonly #command: VerifiedAcpxCommandLease;
|
||||
#activeTurn: AcpxRuntimeTurn | null = null;
|
||||
#closingStarted = false;
|
||||
#closePromise: Promise<void> | null = null;
|
||||
#closed = false;
|
||||
|
||||
private constructor(input: {
|
||||
|
|
@ -281,16 +304,75 @@ export class AcpxRuntimeHost {
|
|||
return Object.freeze({ ...this.#sandbox.persistedEnvironment });
|
||||
}
|
||||
|
||||
startTurn(input: AcpxRuntimeTurnInput): AcpxRuntimeTurn {
|
||||
if (this.#closed || this.#closingStarted) {
|
||||
throw new Error("ACPX runtime host is closing");
|
||||
}
|
||||
if (this.#activeTurn) {
|
||||
throw new Error("ACPX runtime host already has an active turn");
|
||||
}
|
||||
const requestId = boundedRequestId(input.requestId);
|
||||
const text = boundedTurnText(input.text);
|
||||
const turn = this.#runtime.startTurn({
|
||||
text,
|
||||
requestId,
|
||||
...(input.signal ? { signal: input.signal } : {}),
|
||||
});
|
||||
this.#activeTurn = turn;
|
||||
void turn.result
|
||||
.finally(() => {
|
||||
// Once shutdown owns this turn, retain its cancellation handle until
|
||||
// runtime cleanup succeeds. The result may settle while cleanup is
|
||||
// failing, and a later close must still be able to retry cancellation.
|
||||
if (this.#activeTurn === turn && !this.#closingStarted) {
|
||||
this.#activeTurn = null;
|
||||
}
|
||||
})
|
||||
.catch(() => undefined);
|
||||
return turn;
|
||||
}
|
||||
|
||||
async close(input: { reason: string }): Promise<void> {
|
||||
if (this.#closed) return;
|
||||
const error = await cleanupRuntimeResources(
|
||||
if (this.#closePromise) return await this.#closePromise;
|
||||
this.#closingStarted = true;
|
||||
const closePromise = this.#close(boundedReason(input.reason));
|
||||
this.#closePromise = closePromise;
|
||||
try {
|
||||
await closePromise;
|
||||
this.#closed = true;
|
||||
} finally {
|
||||
if (this.#closePromise === closePromise) this.#closePromise = null;
|
||||
}
|
||||
}
|
||||
|
||||
async #close(reason: string): Promise<void> {
|
||||
const errors: unknown[] = [];
|
||||
const activeTurn = this.#activeTurn;
|
||||
if (activeTurn) {
|
||||
try {
|
||||
const cancellationError = await boundedCancellation(
|
||||
activeTurn.cancel({ reason }),
|
||||
);
|
||||
if (cancellationError !== null) errors.push(cancellationError);
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
}
|
||||
}
|
||||
const cleanupError = await cleanupRuntimeResources(
|
||||
this.#runtime,
|
||||
this.#credential,
|
||||
this.#command,
|
||||
boundedReason(input.reason),
|
||||
reason,
|
||||
);
|
||||
if (error) throw error;
|
||||
this.#closed = true;
|
||||
if (!cleanupError) {
|
||||
if (this.#activeTurn === activeTurn) this.#activeTurn = null;
|
||||
this.#closed = true;
|
||||
}
|
||||
if (cleanupError) errors.push(...cleanupError.errors);
|
||||
if (errors.length > 0) {
|
||||
throw new AggregateError(errors, "ACPX runtime cleanup failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -405,25 +487,69 @@ function retainRuntimeHostCleanup(cleanup: Promise<unknown>): void {
|
|||
.catch(() => undefined);
|
||||
}
|
||||
|
||||
async function boundedCancellation(
|
||||
cancellation: Promise<void>,
|
||||
): Promise<unknown | null> {
|
||||
let timer: ReturnType<typeof setTimeout> | undefined;
|
||||
const outcome = await Promise.race([
|
||||
cancellation.then(
|
||||
() => ({ error: null }),
|
||||
(error: unknown) => ({ error }),
|
||||
),
|
||||
new Promise<{ error: unknown }>((resolve) => {
|
||||
timer = setTimeout(
|
||||
() =>
|
||||
resolve({
|
||||
error: new Error(
|
||||
"ACPX turn cancellation exceeded its shutdown timeout",
|
||||
),
|
||||
}),
|
||||
TURN_CANCELLATION_TIMEOUT_MS,
|
||||
);
|
||||
}),
|
||||
]);
|
||||
if (timer) clearTimeout(timer);
|
||||
return outcome.error;
|
||||
}
|
||||
|
||||
async function cleanupRuntimeResources(
|
||||
runtime: AcpxRuntimePort | null,
|
||||
credential: ManagedCodexCredentialLease | null,
|
||||
command: VerifiedAcpxCommandLease | null,
|
||||
reason: string,
|
||||
): Promise<AggregateError | null> {
|
||||
const errors: unknown[] = [];
|
||||
for (const close of [
|
||||
runtime ? () => runtime.close({ reason }) : null,
|
||||
credential ? () => credential.close() : null,
|
||||
command ? () => command.close() : null,
|
||||
]) {
|
||||
if (!close) continue;
|
||||
const settle = async (
|
||||
close: () => Promise<void>,
|
||||
): Promise<unknown | null> => {
|
||||
try {
|
||||
await close();
|
||||
return null;
|
||||
} catch (error) {
|
||||
errors.push(error);
|
||||
return error;
|
||||
}
|
||||
}
|
||||
};
|
||||
// The command lease owns only the already-consumed verified launch
|
||||
// snapshot, so it can be released as shutdown starts. The credential is
|
||||
// different: a provider whose exact runtime close is pending or failed may
|
||||
// still read or rewrite its home. Retain both the staged bytes and the
|
||||
// exclusive home lease until that exact close succeeds.
|
||||
const runtimeOutcome = runtime
|
||||
? settle(() => runtime.close({ reason }))
|
||||
: Promise.resolve(null);
|
||||
const commandOutcome = command
|
||||
? settle(() => command.close())
|
||||
: Promise.resolve(null);
|
||||
const credentialOutcome = (async (): Promise<unknown | null> => {
|
||||
const runtimeError = await runtimeOutcome;
|
||||
if (runtimeError !== null || credential === null) return null;
|
||||
return await settle(() => credential.close());
|
||||
})();
|
||||
const outcomes = await Promise.all([
|
||||
runtimeOutcome,
|
||||
commandOutcome,
|
||||
credentialOutcome,
|
||||
]);
|
||||
const errors = outcomes.filter((error): error is unknown => error !== null);
|
||||
return errors.length > 0
|
||||
? new AggregateError(errors, "ACPX runtime cleanup failed")
|
||||
: null;
|
||||
|
|
@ -441,3 +567,22 @@ function boundedReason(value: string): string {
|
|||
const reason = value.trim().slice(0, 1_000);
|
||||
return reason || "ACPX runtime closed";
|
||||
}
|
||||
|
||||
function boundedRequestId(value: string): string {
|
||||
const requestId = value.trim();
|
||||
if (
|
||||
requestId.length === 0 ||
|
||||
requestId !== value ||
|
||||
Buffer.byteLength(requestId) > 1_024
|
||||
) {
|
||||
throw new Error("ACPX turn request id is outside its bounded size");
|
||||
}
|
||||
return requestId;
|
||||
}
|
||||
|
||||
function boundedTurnText(value: string): string {
|
||||
if (Buffer.byteLength(value) > 1024 * 1024) {
|
||||
throw new Error("ACPX turn text exceeds its bounded size");
|
||||
}
|
||||
return value;
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue