Fence ACP extension callbacks and Pi controls to active turns

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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<void>((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,

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<void>;
+  onRequest?: AcpRuntimeOptions["onExtensionRequest"];
+  onNotification?: AcpRuntimeOptions["onExtensionNotification"];
+}
+
+interface AcpxRuntimeExtensionBoundary {
+  active: AcpxRuntimeExtensionTurn | null;
+  sessionIds: Set<string>;
+  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<string, unknown>): 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<void> },
   permissionBoundary: { active: AbortController | null; handler?: AcpRuntimeOptions["onPermissionRequest"] },
+  extensionBoundary: AcpxRuntimeExtensionBoundary,
 ): AcpxRuntimePort {
+  extensionBoundary.sessionIds = new Set([identity.backendSessionId]);
+  let extensionControls: Promise<void> = Promise.resolve();
+  const controlActiveTurn = (method: "pi/steer" | "pi/follow_up", text: string, requestId: string): Promise<void> => {
+    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<void> => {
+      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<unknown | null>;
     readonly reconciliationGeneration: number;
@@ -1042,6 +1122,7 @@ function runtimePort(
       attemptNumber: number;
     };
   }): Promise<void> {
+    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<string, unknown> {
     : {};
 }
 
+async function abortableExtensionResult<T>(result: Promise<T>, signal: AbortSignal): Promise<T> {
+  signal.throwIfAborted();
+  let rejectAbort: (() => void) | undefined;
+  const cancellation = new Promise<never>((_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 {

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<string, unknown> | 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();

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<AcpxRuntimePortIdentity>;
   getStatus(): Promise<AcpxModelStatus>;
   setModel?(model: string): Promise<void>;
+  steeringCapability?(): AcpxRuntimeSteeringCapability | null;
+  steerActiveTurn?(text: string, requestId: string): Promise<void>;
+  queueFollowUp?(text: string, requestId: string): Promise<void>;
   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<string, unknown>;
   command: VerifiedAcpxCommandLease;
   /** Replace a consumed launch snapshot after an ephemeral control session. */
   refreshConsumedCommand?: () => Promise<void>;
@@ -199,6 +211,8 @@ export interface AcpxRuntimeHostDependencies {
 }
 
 export interface OpenAcpxRuntimeHostOptions {
+  /** Never persisted as session options or recovery identity. */
+  clientCapabilities?: Record<string, unknown>;
   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<void> {
+    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<void> {
+    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