Add steering support for ACP providers that indicate support (#3921)

* Add steering support for ACP providers that indicate support

* fix(acp): keep steering inside the admitted turn

---------

Co-authored-by: paseo-bot[bot] <266920839+paseo-bot[bot]@users.noreply.github.com>

packages/server/src/server/agent/provider-registry.test.ts

@@ -7,6 +7,7 @@ import type {
   AgentFeature,
   AgentModelDefinition,
   AgentMode,
+  AgentSession,
   AgentSessionConfig,
   ProviderCatalog,
 } from "./agent-sdk-types.js";
@@ -532,6 +533,7 @@ import {
   AGENT_PROVIDER_DEFINITIONS,
   buildProviderRegistry,
   createAllClients,
+  wrapSessionProvider,
 } from "./provider-registry.js";
 import { FakeOmp } from "./providers/omp/test-utils/fake-omp.js";
 
@@ -1831,3 +1833,57 @@ describe("fetchCatalog", () => {
     expect(catalog.modes.map((mode) => mode.id)).toEqual(["ask"]);
   });
 });
+
+describe("wrapSessionProvider", () => {
+  function makeFakeSession(overrides?: Partial<AgentSession>): AgentSession {
+    return {
+      provider: "acp",
+      id: "inner-session-id",
+      capabilities: {},
+      run: vi.fn(),
+      startTurn: vi.fn(),
+      subscribe: vi.fn(() => () => {}),
+      streamHistory: (async function* () {})(),
+      getRuntimeInfo: vi.fn(),
+      getAvailableModes: vi.fn(),
+      getCurrentMode: vi.fn(),
+      setMode: vi.fn(),
+      getPendingPermissions: vi.fn(() => []),
+      respondToPermission: vi.fn(),
+      describePersistence: vi.fn(() => null),
+      interrupt: vi.fn(),
+      close: vi.fn(),
+      ...overrides,
+    } satisfies Partial<AgentSession> as AgentSession;
+  }
+
+  // Any provider built on GenericACPAgentClient (cursor, kimi, kiro,
+  // traecli, or a user-configured `extends: "acp"` entry) reports its own
+  // `provider` as the base id ("acp"), which never matches the requested
+  // label, so createResolvedProviderClient always routes it through this
+  // wrapper. A forward missing here silently drops that capability for
+  // every such provider, with no type error, since AgentSession declares
+  // it optional.
+  test("forwards steerActiveTurn to the inner session when present", async () => {
+    const steerActiveTurn = vi.fn(async () => ({ status: "accepted" as const }));
+    const inner = makeFakeSession({ steerActiveTurn });
+
+    const wrapped = wrapSessionProvider("cursor", inner);
+
+    expect(wrapped.steerActiveTurn).toBeDefined();
+    const result = await wrapped.steerActiveTurn?.({ role: "user", content: [] } as never, {
+      expectedTurnId: "turn-1",
+    });
+    expect(result).toEqual({ status: "accepted" });
+    expect(steerActiveTurn).toHaveBeenCalledWith(
+      { role: "user", content: [] },
+      { expectedTurnId: "turn-1" },
+    );
+  });
+
+  test("leaves steerActiveTurn undefined when the inner session doesn't implement it", () => {
+    const inner = makeFakeSession();
+    const wrapped = wrapSessionProvider("cursor", inner);
+    expect(wrapped.steerActiveTurn).toBeUndefined();
+  });
+});

packages/server/src/server/agent/provider-registry.ts

@@ -461,6 +461,7 @@ export function wrapSessionProvider(provider: AgentProvider, inner: AgentSession
     close: () => inner.close(),
     listCommands: inner.listCommands?.bind(inner),
     setModel: inner.setModel?.bind(inner),
+    steerActiveTurn: inner.steerActiveTurn?.bind(inner),
     setThinkingOption: inner.setThinkingOption?.bind(inner),
     setFeature: inner.setFeature?.bind(inner),
     revertConversation: inner.revertConversation?.bind(inner),

packages/server/src/server/agent/providers/acp-agent.test.ts

@@ -93,6 +93,15 @@ interface ACPSessionInternals {
   acpMcpServers(): unknown[];
 }
 
+interface ACPSteerInternals {
+  sessionId: string | null;
+  connection: {
+    extMethod: (method: string, params: unknown) => Promise<{ outcome: string }>;
+  };
+  steeringSupported: boolean;
+  activeForegroundTurnId: string | null;
+}
+
 interface ACPModelSelectionInternals {
   sessionId: string | null;
   connection: {
@@ -148,6 +157,117 @@ function createSession(terminateProcess?: ProcessTerminator): ACPAgentSession {
   );
 }
 
+describe("ACPAgentSession.steerActiveTurn", () => {
+  function primeForSteer(
+    session: ACPAgentSession,
+    overrides: Partial<ACPSteerInternals> = {},
+  ): { internals: ACPSteerInternals; extMethod: ReturnType<typeof vi.fn> } {
+    const extMethod = vi.fn(async () => ({ outcome: "injected" }));
+    const internals = asInternals<ACPSteerInternals>(session);
+    internals.sessionId = "session-1";
+    internals.connection = { extMethod };
+    internals.steeringSupported = true;
+    internals.activeForegroundTurnId = "turn-1";
+    Object.assign(internals, overrides);
+    return { internals, extMethod };
+  }
+
+  test("returns unavailable when the agent never advertised steering support", async () => {
+    const session = createSession();
+    primeForSteer(session, { steeringSupported: false });
+
+    const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" });
+
+    expect(result).toEqual({ status: "unavailable" });
+  });
+
+  test("returns unavailable when the caller's expected turn doesn't match the live one", async () => {
+    // Guards against steering into the wrong turn: the caller's belief about
+    // what's running raced with the agent's own turn boundary.
+    const session = createSession();
+    const { extMethod } = primeForSteer(session, { activeForegroundTurnId: "turn-2" });
+
+    const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" });
+
+    expect(result).toEqual({ status: "unavailable" });
+    expect(extMethod).not.toHaveBeenCalled();
+  });
+
+  test("returns unavailable when there is no live connection", async () => {
+    const session = createSession();
+    primeForSteer(session, { connection: undefined as never });
+
+    const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" });
+
+    expect(result).toEqual({ status: "unavailable" });
+  });
+
+  test("forwards to _session/steering with the session id and prompt content", async () => {
+    const session = createSession();
+    const { extMethod } = primeForSteer(session);
+
+    await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" });
+
+    expect(extMethod).toHaveBeenCalledWith("_session/steering", {
+      sessionId: "session-1",
+      prompt: [{ type: "text", text: "steer this" }],
+      _meta: { steering: { idleBehavior: "promptRequired" } },
+    });
+  });
+
+  test.each([
+    ["injected", "accepted"],
+    ["startedNewTurn", "unavailable"],
+    ["promptRequired", "unavailable"],
+    ["failed", "unavailable"],
+  ] as const)("maps agent outcome %s to status %s", async (outcome, status) => {
+    const session = createSession();
+    const { extMethod } = primeForSteer(session);
+    extMethod.mockResolvedValue({ outcome });
+
+    const result = await session.steerActiveTurn("steer this", { expectedTurnId: "turn-1" });
+
+    expect(result).toEqual({ status });
+  });
+
+  test("denies pending permissions after the steer is accepted", async () => {
+    const session = createSession();
+    primeForSteer(session);
+    const events: AgentStreamEvent[] = [];
+    session.subscribe((event) => events.push(event));
+    const permission = session.requestPermission({
+      sessionId: "session-1",
+      toolCall: {
+        toolCallId: "tool-1",
+        title: "Run command",
+        kind: "execute",
+        status: "pending",
+      },
+      options: [
+        { optionId: "allow-once", name: "Allow", kind: "allow_once" },
+        { optionId: "reject-once", name: "Reject", kind: "reject_once" },
+      ],
+    } satisfies RequestPermissionRequest);
+    await Promise.resolve();
+
+    await session.steerActiveTurn("steer this", {
+      expectedTurnId: "turn-1",
+      clearPendingPermissions: true,
+    });
+
+    await expect(permission).resolves.toEqual({
+      outcome: { outcome: "selected", optionId: "reject-once" },
+    });
+    expect(session.getPendingPermissions()).toEqual([]);
+    expect(events).toContainEqual(
+      expect.objectContaining({
+        type: "permission_resolved",
+        resolution: expect.objectContaining({ behavior: "deny" }),
+      }),
+    );
+  });
+});
+
 // Typed substitute for the real tree-kill terminator. Records which child
 // processes it was asked to terminate, so tests assert on observable state
 // instead of spying on the production function. In "deferred" mode the

packages/server/src/server/agent/providers/acp-agent.ts

@@ -94,6 +94,8 @@ import {
   type ProviderCatalog,
   type ResolveAgentCreateConfigInput,
   type ResolveAgentCreateConfigResult,
+  type SteerActiveTurnOptions,
+  type SteerResult,
   type ToolCallDetail,
   type ToolCallTimelineItem,
 } from "../agent-sdk-types.js";
@@ -1456,6 +1458,7 @@ export class ACPAgentSession implements AgentSession, ACPClient {
   private readonly extensionCommandsParser?: ACPExtensionCommandsParser;
   private currentTurnUsage: AgentUsage | undefined;
   private activeForegroundTurnId: string | null = null;
+  private steeringSupported = false;
   private fallbackAssistantMessageId: string | null = null;
   private closed = false;
   private historyPending = false;
@@ -1506,6 +1509,8 @@ export class ACPAgentSession implements AgentSession, ACPClient {
       this.child = spawned.child;
       this.connection = spawned.connection;
       this.agentCapabilities = spawned.initialize.agentCapabilities ?? null;
+      this.steeringSupported =
+        readRecord(readRecord(spawned.initialize._meta)?.steering)?.supported === true;
 
       const response = await this.runACPRequest(() =>
         this.connection!.newSession({
@@ -1540,6 +1545,8 @@ export class ACPAgentSession implements AgentSession, ACPClient {
       this.child = spawned.child;
       this.connection = spawned.connection;
       this.agentCapabilities = spawned.initialize.agentCapabilities ?? null;
+      this.steeringSupported =
+        readRecord(readRecord(spawned.initialize._meta)?.steering)?.supported === true;
       this.sessionId = handle.sessionId;
       this.bootstrapThreadEventPending = true;
 
@@ -1654,6 +1661,45 @@ export class ACPAgentSession implements AgentSession, ACPClient {
     return { turnId };
   }
 
+  async steerActiveTurn(
+    prompt: AgentPromptInput,
+    options: SteerActiveTurnOptions,
+  ): Promise<SteerResult> {
+    if (
+      !this.connection ||
+      !this.sessionId ||
+      !this.steeringSupported ||
+      this.activeForegroundTurnId !== options.expectedTurnId
+    ) {
+      return { status: "unavailable" };
+    }
+    const response = await this.runACPRequest(() =>
+      this.connection!.extMethod("_session/steering", {
+        sessionId: this.sessionId,
+        prompt: toACPContentBlocks(prompt),
+        _meta: { steering: { idleBehavior: "promptRequired" } },
+      }),
+    );
+    if (response.outcome !== "injected") {
+      return { status: "unavailable" };
+    }
+    if (options.clearPendingPermissions) {
+      await this.clearPendingPermissionsForSteer();
+    }
+    return { status: "accepted" };
+  }
+
+  private async clearPendingPermissionsForSteer(): Promise<void> {
+    const requestIds = Array.from(this.pendingPermissions.keys());
+    for (const requestId of requestIds) {
+      if (!this.pendingPermissions.has(requestId)) continue;
+      await this.respondToPermission(requestId, {
+        behavior: "deny",
+        message: "The user answered with a message instead of approving. Their message follows.",
+      });
+    }
+  }
+
   subscribe(callback: (event: AgentStreamEvent) => void): () => void {
     this.subscribers.add(callback);
     if (this.sessionId) {