diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts index 9d53e2cab1b..f1c44314788 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.test.ts @@ -28,6 +28,91 @@ const HANDLE: AcpRuntimeHandle = { }; describe("Codex ACPX runtime adapter", () => { + it("routes only profile-allowed extensions to the owning turn and expires late responses", async () => { + const runtime = fakeRuntime(); + const first = pendingExtensionTurn("turn-1"); + const second = pendingExtensionTurn("turn-2"); + vi.mocked(runtime.startTurn).mockReturnValueOnce(first.turn).mockReturnValueOnce(second.turn); + let created!: AcpRuntimeOptions; + const options = openOptions(fakeCommand()); + options.profile = { ...options.profile, agent: "cursor" }; + options.clientCapabilities = { _meta: { cursor: { test: true } } }; + const port = await openCodexAcpxRuntime(options, { + createRegistry: () => registry(), createStore: () => store(), + createRuntime: (value) => { created = value; return runtime; }, + }); + expect(created.clientCapabilities).toEqual(options.clientCapabilities); + expect(JSON.stringify(vi.mocked(runtime.ensureSession).mock.calls)).not.toContain("clientCapabilities"); + const notification = vi.fn(); + const request = vi.fn(async (_method, _params, context) => ({ requestId: context.requestId })); + const admittedFirst = port.startTurn({ text: "First", requestId: "turn-1", onExtensionRequest: request, onExtensionNotification: notification }); + const signal = new AbortController().signal; + await expect(created.onExtensionRequest!("cursor/ask_question", { sessionId: "backend-1" }, { requestId: 0, signal })).resolves.toEqual({ requestId: 0 }); + await expect(created.onExtensionRequest!("pi/steer", {}, { requestId: 1, signal })).rejects.toThrow("admitted active turn"); + await expect(created.onExtensionRequest!("cursor/ask_question", { sessionId: "other" }, { requestId: 1, signal })).rejects.toThrow("admitted active turn"); + created.onExtensionNotification!("cursor/task", { sessionId: "backend-1", taskId: "t" }); + created.onExtensionNotification!("cursor/task", { sessionId: "other" }); + created.onExtensionNotification!("cursor/ask_question", { sessionId: "backend-1" }); + expect(notification).toHaveBeenCalledExactlyOnceWith("cursor/task", { sessionId: "backend-1", taskId: "t" }); + request.mockImplementationOnce(() => new Promise(() => undefined)); + const pending = created.onExtensionRequest!("cursor/ask_question", {}, { requestId: 2, signal }); + const rejected = expect(pending).rejects.toThrow("expired"); + first.settle(); await admittedFirst.result; await rejected; + const nextRequest = vi.fn(async () => ({ accepted: true })); + const admittedSecond = port.startTurn({ text: "Second", requestId: "turn-2", onExtensionRequest: nextRequest }); + expect(nextRequest).not.toHaveBeenCalled(); + await admittedSecond.cancel({ reason: "cancelled" }); + await expect(created.onExtensionRequest!("cursor/ask_question", {}, { requestId: 3, signal })).rejects.toThrow("admitted active turn"); + second.settle(); await admittedSecond.result; await port.close({ reason: "test finished" }); + }); + + it("requires the Pi capability handshake and acknowledgments, preserves follow-up order, and never starts a second prompt", async () => { + const pending = pendingExtensionTurn("turn-1"); + const runtime = Object.assign(fakeRuntime(), { requestExtension: vi.fn(async () => ({ accepted: true })) }); + vi.mocked(runtime.startTurn).mockReturnValue(pending.turn); + let created!: AcpRuntimeOptions & { onAgentInitialize?: (result: unknown) => void }; + const options = openOptions(fakeCommand()); + options.profile = { ...options.profile, agent: "pi" }; + const port = await openCodexAcpxRuntime(options, { + createRegistry: () => registry(), createStore: () => store(), + createRuntime: (value) => { created = value; return runtime; }, + }); + const turn = port.startTurn({ text: "Initial", requestId: "turn-1" }); + await expect(port.steerActiveTurn!("Update", "turn-1")).rejects.toThrow("supported active turn"); + created.onAgentInitialize!({ agentCapabilities: { _meta: { paperclipPi: { version: 1, steering: true, queuedFollowUp: true } } } }); + expect(port.steeringCapability!()).toEqual({ steering: true, queuedFollowUp: true }); + await expect(port.steerActiveTurn!("Update", "wrong")).rejects.toThrow("supported active turn"); + await port.steerActiveTurn!("Update", "turn-1"); + await Promise.all([port.queueFollowUp!("One", "turn-1"), port.queueFollowUp!("Two", "turn-1")]); + expect(runtime.requestExtension.mock.calls.map((call) => call[0])).toEqual([ + { handle: HANDLE, method: "pi/steer", params: { sessionId: "backend-1", message: "Update" }, sessionMode: "persistent" }, + { handle: HANDLE, method: "pi/follow_up", params: { sessionId: "backend-1", message: "One" }, sessionMode: "persistent" }, + { handle: HANDLE, method: "pi/follow_up", params: { sessionId: "backend-1", message: "Two" }, sessionMode: "persistent" }, + ]); + expect(runtime.startTurn).toHaveBeenCalledTimes(1); + runtime.requestExtension.mockResolvedValueOnce({ accepted: false }); + await expect(port.steerActiveTurn!("Denied", "turn-1")).rejects.toThrow("not acknowledged"); + runtime.requestExtension.mockImplementationOnce(() => new Promise(() => undefined)); + const stale = port.queueFollowUp!("Stale", "turn-1"); + const rejected = expect(stale).rejects.toThrow("expired"); + pending.settle(); await turn.result; await rejected; + await expect(port.queueFollowUp!("Late", "turn-1")).rejects.toThrow("supported active turn"); + await port.close({ reason: "test finished" }); + }); + + it("does not grant steering to Copilot or Cursor when an untrusted initialize result claims Pi support", async () => { + const options = openOptions(fakeCommand()); options.profile = { ...options.profile, agent: "copilot" }; + const runtime = fakeRuntime(); + const port = await openCodexAcpxRuntime(options, { + createRegistry: () => registry(), createStore: () => store(), + createRuntime: (created) => { + (created as AcpRuntimeOptions & { onAgentInitialize: (result: unknown) => void }).onAgentInitialize({ agentCapabilities: { _meta: { paperclipPi: { version: 1, steering: true, queuedFollowUp: true } } } }); + return runtime; + }, + }); + expect(port.steeringCapability!()).toBeNull(); + await port.close({ reason: "test finished" }); + }); it.each(["approve-reads", "approve-paperclip"] as const)("%s stops unassigned operations when approval has no handler", async (permissionMode) => { const runtime = fakeRuntime(); let runtimeOptions: AcpRuntimeOptions | undefined; @@ -2807,6 +2892,17 @@ function fakeRuntime(handle: AcpRuntimeHandle = HANDLE): AcpRuntime { }; } +function pendingExtensionTurn(requestId: string) { + let settle!: () => void; + const result = new Promise((resolve) => { settle = resolve; }); + return { settle, turn: { + requestId, promptStarted: Promise.resolve(), + events: (async function* () { await result; })(), + result: result.then(() => ({ status: "completed" as const })), + cancel: async () => undefined, closeStream: async () => undefined, + } }; +} + function runtimeWithProvider( runtime: AcpRuntime, options: AcpRuntimeOptions, diff --git a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts index 0297f538243..f4fc95a27c1 100644 --- a/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts +++ b/packages/paperclip-runner/src/drivers/acpx/codex-runtime-adapter.ts @@ -16,6 +16,7 @@ import { import type { AcpxRuntimeGoalCapability, AcpxRuntimeGoalSnapshot, + AcpxRuntimeSteeringCapability, AcpxRuntimePort, AcpxRuntimePortIdentity, AcpxRuntimePortOpenOptions, @@ -28,6 +29,7 @@ import { } from "./installation-integrity.js"; import type { AcpxModelStatus } from "./model-verification.js"; import { AcpxApprovalRequiredError, decideAcpxPermission } from "./permission-policy.js"; +import { ACPX_CAPABILITY_PROFILES } from "./capability-profiles.js"; const VERIFIED_COMMAND_SENTINEL = "paperclip-verified-acpx-command"; const DEFAULT_RUNTIME_CLOSE_TIMEOUT_MS = 2_000; @@ -80,6 +82,21 @@ interface AcpxRuntimeGoalState { observedSnapshot: boolean; } +interface AcpxRuntimeExtensionTurn { + requestId: string; + controller: AbortController; + signal: AbortSignal; + promptStarted: Promise; + onRequest?: AcpRuntimeOptions["onExtensionRequest"]; + onNotification?: AcpRuntimeOptions["onExtensionNotification"]; +} + +interface AcpxRuntimeExtensionBoundary { + active: AcpxRuntimeExtensionTurn | null; + sessionIds: Set; + steering: AcpxRuntimeSteeringCapability | null; +} + class AcpxRuntimeCloseTimeoutError extends Error { constructor() { super("ACPX runtime close timed out"); @@ -249,6 +266,15 @@ export async function openQualifiedAcpxRuntime( active: AbortController | null; handler?: AcpRuntimeOptions["onPermissionRequest"]; } = { active: null }; + const extensionProfile = ACPX_CAPABILITY_PROFILES[options.profile.agent]; + const extensionRequests = new Set(extensionProfile.extensionRequests); + const extensionNotifications = new Set(extensionProfile.extensionNotifications); + const extensionBoundary: AcpxRuntimeExtensionBoundary = { + active: null, sessionIds: new Set(), steering: null, + }; + const ownsExtensionTurn = (active: AcpxRuntimeExtensionTurn, params: Record): boolean => + extensionBoundary.active === active && !active.signal.aborted + && (params.sessionId === undefined || (typeof params.sessionId === "string" && extensionBoundary.sessionIds.has(params.sessionId))); const goalState: AcpxRuntimeGoalState = { capability: null, snapshot: null, @@ -286,6 +312,24 @@ export async function openQualifiedAcpxRuntime( ? "approve-reads" : options.permissionMode, elicitationModes: ["form"], + ...(options.clientCapabilities === undefined ? {} : { clientCapabilities: structuredClone(options.clientCapabilities) }), + extensionMethods: [...new Set([...extensionRequests, ...extensionNotifications])], + onExtensionRequest: async (method, params, context) => { + const active = extensionBoundary.active; + if (!extensionRequests.has(method) || !active?.onRequest || !ownsExtensionTurn(active, params) || context.signal.aborted) { + throw new Error("ACPX extension request does not own an admitted active turn"); + } + const signal = AbortSignal.any([active.signal, context.signal]); + const response = await abortableExtensionResult(active.onRequest(method, params, { requestId: context.requestId, signal }), signal); + signal.throwIfAborted(); + if (!ownsExtensionTurn(active, params)) throw new Error("ACPX extension request turn expired"); + return response; + }, + onExtensionNotification: (method, params) => { + const active = extensionBoundary.active; + if (!extensionNotifications.has(method) || !active?.onNotification || !ownsExtensionTurn(active, params)) return; + active.onNotification(method, params); + }, nonInteractivePermissions: "fail", permissionPolicy: { ...options.permissionPolicy, @@ -341,6 +385,12 @@ export async function openQualifiedAcpxRuntime( onAgentInitialize: (result) => { const capability = goalCapabilityFromAcpMessage({ result }); if (capability) goalState.capability = capability; + // Only the runner-owned Pi wrapper has an admitted steering contract. + // Never infer it from generic ACP prompt support or native-only features. + const pi = objectRecord(objectRecord(objectRecord(objectRecord(result).agentCapabilities)._meta).paperclipPi); + extensionBoundary.steering = options.profile.agent === "pi" && pi.version === 1 + ? { steering: pi.steering === true, queuedFollowUp: pi.queuedFollowUp === true } + : null; }, onSessionNotification: acceptGoalNotification, spawnEnvironment: () => ({ @@ -470,6 +520,7 @@ export async function openQualifiedAcpxRuntime( goalState, commandLaunches, permissionBoundary, + extensionBoundary, ); } catch (error) { const cleanupReason = "ACPX runtime identity validation failed"; @@ -898,7 +949,36 @@ function runtimePort( goalState: AcpxRuntimeGoalState, commandLaunches: { count: number; refreshConsumedCommand?: () => Promise }, permissionBoundary: { active: AbortController | null; handler?: AcpRuntimeOptions["onPermissionRequest"] }, + extensionBoundary: AcpxRuntimeExtensionBoundary, ): AcpxRuntimePort { + extensionBoundary.sessionIds = new Set([identity.backendSessionId]); + let extensionControls: Promise = Promise.resolve(); + const controlActiveTurn = (method: "pi/steer" | "pi/follow_up", text: string, requestId: string): Promise => { + const active = extensionBoundary.active; + const supported = method === "pi/steer" + ? extensionBoundary.steering?.steering : extensionBoundary.steering?.queuedFollowUp; + if (!supported || !runtime.requestExtension || !active || active.requestId !== requestId || active.signal.aborted) { + return Promise.reject(new Error("ACPX extension control does not own a supported active turn")); + } + if (!text.trim() || Buffer.byteLength(text, "utf8") > 65_536 || text.includes("\0")) { + return Promise.reject(new Error("ACPX extension control message is invalid")); + } + const send = async (): Promise => { + await abortableExtensionResult(active.promptStarted, active.signal); + active.signal.throwIfAborted(); + if (extensionBoundary.active !== active) throw new Error("ACPX extension control turn expired"); + const response = await abortableExtensionResult(runtime.requestExtension!({ + handle, method, params: { sessionId: identity.backendSessionId, message: text }, sessionMode: "persistent", + }), active.signal); + active.signal.throwIfAborted(); + if (extensionBoundary.active !== active) throw new Error("ACPX extension control turn expired"); + if (response.accepted !== true) throw new Error("ACPX extension control was not acknowledged"); + }; + const delivery = extensionControls.then(send); + // Preserve caller ordering without letting one rejection deadlock later calls. + extensionControls = delivery.catch(() => undefined); + return delivery; + }; type RuntimeCloseAttempt = { readonly outcome: Promise; readonly reconciliationGeneration: number; @@ -1042,6 +1122,7 @@ function runtimePort( attemptNumber: number; }; }): Promise { + extensionBoundary.active?.controller.abort(new Error("ACPX runtime is closing")); if (runtimeClosed) return; if ( runtimeCloseAttempt?.origin.kind === "reconciliation" && @@ -1128,6 +1209,11 @@ function runtimePort( async getStatus() { return await persistedRuntimeStatus(sessionStore, handle, identity); }, + steeringCapability() { + return extensionBoundary.steering === null ? null : structuredClone(extensionBoundary.steering); + }, + steerActiveTurn: (text, requestId) => controlActiveTurn("pi/steer", text, requestId), + queueFollowUp: (text, requestId) => controlActiveTurn("pi/follow_up", text, requestId), goalCapability() { return goalState.capability === null ? null @@ -1221,7 +1307,20 @@ function runtimePort( } : {}), startTurn(input) { + if (extensionBoundary.active) throw new Error("ACPX runtime already has an active turn"); const approval = new AbortController(); + const controller = new AbortController(); + const extensionTurn: AcpxRuntimeExtensionTurn = { + requestId: input.requestId, controller, + signal: input.signal ? AbortSignal.any([controller.signal, input.signal, approval.signal]) : AbortSignal.any([controller.signal, approval.signal]), + promptStarted: Promise.resolve(), + onRequest: input.onExtensionRequest, onNotification: input.onExtensionNotification, + }; + extensionBoundary.active = extensionTurn; + const releaseExtensionTurn = (): void => { + controller.abort(new Error("ACPX extension turn expired")); + if (extensionBoundary.active === extensionTurn) extensionBoundary.active = null; + }; permissionBoundary.active = approval; permissionBoundary.handler = input.onPermissionRequest; const finishOwnershipAdmission = @@ -1241,21 +1340,32 @@ function runtimePort( : {}), }); } catch (error) { + releaseExtensionTurn(); if (permissionBoundary.active === approval) { permissionBoundary.active = null; permissionBoundary.handler = undefined; } void finishOwnershipAdmission().catch(() => undefined); throw error; } const guarded = turnWithVerifiedLifetimeOwnership(turn, finishOwnershipAdmission); + extensionTurn.promptStarted = guarded.promptStarted; const result = guarded.result.then( (value) => { approval.signal.throwIfAborted(); return value; }, (error: unknown) => { approval.signal.throwIfAborted(); throw error; }, ).finally(() => { + releaseExtensionTurn(); if (permissionBoundary.active === approval) { permissionBoundary.active = null; permissionBoundary.handler = undefined; approval.abort(); } }); void result.catch(() => undefined); return { ...guarded, result, + cancel: (input) => { + controller.abort(new Error("ACPX extension turn cancelled")); + return guarded.cancel(input); + }, + closeStream: (input) => { + controller.abort(new Error("ACPX extension stream closed")); + return guarded.closeStream(input); + }, events: (async function* () { try { for await (const event of guarded.events) { @@ -1811,6 +1921,17 @@ function objectRecord(value: unknown): Record { : {}; } +async function abortableExtensionResult(result: Promise, signal: AbortSignal): Promise { + signal.throwIfAborted(); + let rejectAbort: (() => void) | undefined; + const cancellation = new Promise((_resolve, reject) => { + rejectAbort = () => reject(signal.reason ?? new Error("ACPX extension turn expired")); + signal.addEventListener("abort", rejectAbort, { once: true }); + }); + try { return await Promise.race([result, cancellation]); } + finally { if (rejectAbort) signal.removeEventListener("abort", rejectAbort); } +} + function goalCapabilityFromAcpMessage( message: unknown, ): AcpxRuntimeGoalCapability | null { diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts index d7aeda29e7f..b0ed749af0f 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.test.ts @@ -1176,6 +1176,8 @@ describe("ACPX runtime host", () => { const turn = runtimeTurn(); const startTurn = vi.fn(() => turn); const onElicitation = vi.fn(); + const onExtensionRequest = vi.fn(); + const onExtensionNotification = vi.fn(); const runtime = runtimePort({ startTurn }); const host = await AcpxRuntimeHost.open( { @@ -1193,12 +1195,16 @@ describe("ACPX runtime host", () => { text: "Complete the task.", requestId: "turn-1", onElicitation, + onExtensionRequest, + onExtensionNotification, }), ).toBe(turn); expect(startTurn).toHaveBeenCalledWith({ text: "Complete the task.", requestId: "turn-1", onElicitation, + onExtensionRequest, + onExtensionNotification, }); expect(() => host.startTurn({ text: "Concurrent", requestId: "turn-2" }), @@ -1215,6 +1221,34 @@ describe("ACPX runtime host", () => { ); }); + it("clones ephemeral capabilities and fences steering controls to an acknowledged active turn", async () => { + const fixture = await hostFixture(); + const turn = runtimeTurn(); + const runtime = Object.assign(runtimePort({ startTurn: () => turn }), { + steeringCapability: () => ({ steering: true, queuedFollowUp: true }), + steerActiveTurn: vi.fn(async () => undefined), + queueFollowUp: vi.fn(async () => undefined), + }); + const clientCapabilities = { _meta: { fixture: { enabled: true } } }; + let observed: Record | undefined; + const host = await AcpxRuntimeHost.open({ + ...fixture.options, agent: "codex", model: "gpt-5.6-sol", permissionMode: "approve-reads", + environment: { PAPERCLIP_ACPX_CODEX_AUTH_JSON_SECRET: "{}" }, clientCapabilities, + }, fixture.dependencies({ openRuntime: async (options) => { observed = options.clientCapabilities; return runtime; } })); + expect(observed).toEqual(clientCapabilities); + expect(observed).not.toBe(clientCapabilities); + await expect(host.steerActiveTurn("Early")).rejects.toThrow("active turn"); + host.startTurn({ text: "Start", requestId: "turn-1" }); + await expect(host.steerActiveTurn("Wrong turn", "turn-other")).rejects.toThrow("active turn"); + await host.steerActiveTurn("Steer", "turn-1"); + await host.queueFollowUp("Next", "turn-1"); + expect(runtime.steerActiveTurn).toHaveBeenCalledExactlyOnceWith("Steer", "turn-1"); + expect(runtime.queueFollowUp).toHaveBeenCalledExactlyOnceWith("Next", "turn-1"); + expect(runtime.startTurn).toHaveBeenCalledOnce(); + await host.close({ reason: "controls tested" }); + await expect(host.queueFollowUp("Too late", "turn-1")).rejects.toThrow("active turn"); + }); + it("rejects oversized turn inputs before calling the runtime", async () => { const fixture = await hostFixture(); const runtime = runtimePort(); diff --git a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts index 9f4b39d867f..37cbc74e84e 100644 --- a/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts +++ b/packages/paperclip-runner/src/drivers/acpx/runtime-host.ts @@ -83,6 +83,11 @@ export interface AcpxRuntimeGoalCapability { actions: Array<"set" | "pause" | "resume" | "clear">; } +export interface AcpxRuntimeSteeringCapability { + steering: boolean; + queuedFollowUp: boolean; +} + export interface AcpxRuntimeGoalSnapshot { objective: string; status: "active" | "paused" | "blocked" | "limited" | "complete"; @@ -101,6 +106,8 @@ export interface AcpxRuntimeTurnInput { signal?: AbortSignal; onElicitation?: AcpElicitationHandler; onPermissionRequest?: AcpRuntimeOptions["onPermissionRequest"]; + onExtensionRequest?: AcpRuntimeOptions["onExtensionRequest"]; + onExtensionNotification?: AcpRuntimeOptions["onExtensionNotification"]; } export interface AcpxRuntimeTurn { @@ -117,6 +124,9 @@ export interface AcpxRuntimePort { identity(): Promise; getStatus(): Promise; setModel?(model: string): Promise; + steeringCapability?(): AcpxRuntimeSteeringCapability | null; + steerActiveTurn?(text: string, requestId: string): Promise; + queueFollowUp?(text: string, requestId: string): Promise; goalCapability?(): AcpxRuntimeGoalCapability | null; goalSnapshot?(): AcpxRuntimeGoalSnapshot | null; controlGoal?( @@ -128,6 +138,8 @@ export interface AcpxRuntimePort { } export interface AcpxRuntimePortOpenOptions { + /** Ephemeral provider extension metadata; mandatory ACP caps remain host-owned. */ + clientCapabilities?: Record; command: VerifiedAcpxCommandLease; /** Replace a consumed launch snapshot after an ephemeral control session. */ refreshConsumedCommand?: () => Promise; @@ -199,6 +211,8 @@ export interface AcpxRuntimeHostDependencies { } export interface OpenAcpxRuntimeHostOptions { + /** Never persisted as session options or recovery identity. */ + clientCapabilities?: Record; runtimeDirectory: string; normalizedSessionId: string; workingDirectory: string; @@ -463,6 +477,9 @@ export class AcpxRuntimeHost { acquire: () => { options.assertWorkspaceHeld?.(); return dependencies.openRuntime({ + ...(options.clientCapabilities === undefined ? {} : { + clientCapabilities: structuredClone(options.clientCapabilities), + }), command: command!, refreshConsumedCommand: commandOwner.refreshConsumedCommand, profile, @@ -629,6 +646,36 @@ export class AcpxRuntimeHost { return this.#runtime.goalCapability?.() ?? null; } + steeringCapability(): AcpxRuntimeSteeringCapability | null { + return this.#runtime.steeringCapability?.() ?? null; + } + + async steerActiveTurn(text: string, turnId?: string): Promise { + const turn = this.#activeTurn; + if (this.#closed || this.#closingStarted || !turn || (turnId !== undefined && turn.requestId !== turnId)) { + throw new Error("ACPX steering does not own an active turn"); + } + if (!this.steeringCapability()?.steering || !this.#runtime.steerActiveTurn) { + throw new Error("ACPX active steering is unavailable"); + } + await turn.promptStarted; + if (this.#activeTurn !== turn || this.#closingStarted) throw new Error("ACPX steering turn expired"); + await this.#runtime.steerActiveTurn(boundedTurnText(text), turn.requestId); + } + + async queueFollowUp(text: string, turnId?: string): Promise { + const turn = this.#activeTurn; + if (this.#closed || this.#closingStarted || !turn || (turnId !== undefined && turn.requestId !== turnId)) { + throw new Error("ACPX follow-up does not own an active turn"); + } + if (!this.steeringCapability()?.queuedFollowUp || !this.#runtime.queueFollowUp) { + throw new Error("ACPX queued follow-up is unavailable"); + } + await turn.promptStarted; + if (this.#activeTurn !== turn || this.#closingStarted) throw new Error("ACPX follow-up turn expired"); + await this.#runtime.queueFollowUp(boundedTurnText(text), turn.requestId); + } + goalSnapshot(): AcpxRuntimeGoalSnapshot | null { return this.#runtime.goalSnapshot?.() ?? null; } @@ -663,6 +710,8 @@ export class AcpxRuntimeHost { ...(input.signal ? { signal: input.signal } : {}), ...(input.onElicitation ? { onElicitation: input.onElicitation } : {}), ...(input.onPermissionRequest ? { onPermissionRequest: input.onPermissionRequest } : {}), + ...(input.onExtensionRequest ? { onExtensionRequest: input.onExtensionRequest } : {}), + ...(input.onExtensionNotification ? { onExtensionNotification: input.onExtensionNotification } : {}), }); this.#activeTurn = turn; void turn.result