feat(acp): support _session/steering (#54394)

Co-authored-by: Dan Smolsky <dannysmo@gmail.com>

packages/cli/src/acp/agent.ts

@@ -10,7 +10,7 @@ import {
   type Stream,
 } from "@agentclientprotocol/sdk"
 import type { OpenCodeClient } from "@opencode/client/effect"
-import { Cause, Deferred, Effect, Ref, type Scope } from "effect"
+import { Cause, Deferred, Effect, Ref, Result, Schema, type Scope } from "effect"
 import { ACPCapabilities } from "./capabilities"
 import { ACPCatalog } from "./catalog"
 import { ACPClient } from "./client"
@@ -117,6 +117,14 @@ export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stre
     "session/cancel",
     handle((service, ctx) => service.cancel(ctx.params)),
   )
+  // Interim extension until ACP has native mid-turn steering.
+  app.onRequest(
+    "_session/steering",
+    parseSteering,
+    handle<ACPTurn.SteeringRequest, ACPTurn.SteeringResponse>((service, ctx) => service.steer(ctx.params))(
+      spanName("_session/steering"),
+    ),
+  )
   const acp = ACPConnection.make(app, stream)
   const connection = acp.connection
   const sessions = yield* ACPSessions.make({ client, connection, catalog })
@@ -129,6 +137,14 @@ export const connect = Effect.fnUntraced(function* (client: OpenCodeClient, stre
   return acp.agent
 })
 
+const Steering = Schema.Struct({ sessionId: Schema.String, prompt: Schema.mutable(Schema.Array(Schema.Any)) })
+
+function parseSteering(params: unknown) {
+  return Schema.decodeUnknownResult(Steering)(params).pipe(
+    Result.getOrThrowWith((error) => RequestError.invalidParams({ details: error.message })),
+  )
+}
+
 const spanName = (method: string) => `cli.acp.${method.replaceAll("/", ".")}`
 
 export * as ACP from "./agent"

packages/cli/src/acp/service.ts

@@ -62,6 +62,7 @@ export interface Interface {
   ) => Effect.Effect<SetSessionConfigOptionResponse, ACPError.Failure>
   readonly setSessionMode: (input: SetSessionModeRequest) => Effect.Effect<SetSessionModeResponse, ACPError.Failure>
   readonly prompt: (input: PromptRequest, signal: AbortSignal) => Effect.Effect<PromptResponse, ACPError.Failure>
+  readonly steer: (input: ACPTurn.SteeringRequest) => Effect.Effect<ACPTurn.SteeringResponse, ACPError.Failure>
   readonly cancel: (input: CancelNotification) => Effect.Effect<void>
 }
 
@@ -139,6 +140,7 @@ export function make(input: {
           _meta: { [ACPCapabilities.ChildSessionUpdates]: true },
         },
         authMethods: [authMethod],
+        _meta: { steering: { supported: true } },
         agentInfo: { name: "OpenCode", version: OPENCODE_VERSION },
       }
     }),
@@ -256,6 +258,7 @@ export function make(input: {
         unprompted.delete(params.sessionId)
         return input.turn.prompt(params, signal)
       }),
+    steer: input.turn.steer,
     cancel: input.turn.cancel,
   }
 }

packages/cli/src/acp/turn.ts

@@ -29,8 +29,15 @@ import { ACPPrompt } from "./prompt"
 import type { ACPSessions, Attached } from "./sessions"
 import { ACPTranslate } from "./translate"
 
+export type SteeringRequest = { readonly sessionId: string; readonly prompt: PromptRequest["prompt"] }
+export type SteeringResponse =
+  | { readonly outcome: "injected" }
+  | { readonly outcome: "promptRequired"; readonly reason: "noRunningTurn" }
+
 export interface Interface {
   readonly prompt: (input: PromptRequest, signal: AbortSignal) => Effect.Effect<PromptResponse, ACPError.Failure>
+  /** Redirects a running turn; an idle session is left alone so a steer never starts a turn. */
+  readonly steer: (input: SteeringRequest) => Effect.Effect<SteeringResponse, ACPError.Failure>
   readonly cancel: (input: CancelNotification) => Effect.Effect<void>
   /** Unlike `cancel`, interrupts an idle session too, since server work can outlive its turn. */
   readonly close: (sessionID: string) => Effect.Effect<void, ACPError.Error | RequestError>
@@ -223,7 +230,7 @@ export const make = Effect.fnUntraced(function* (input: {
     }
   })
 
-  const submit = Effect.fnUntraced(function* (attached: Attached, prompt: ACPPrompt.Prepared) {
+  const submitSynthetic = Effect.fnUntraced(function* (attached: Attached, prompt: ACPPrompt.Prepared) {
     const sessionID = attached.id
     if (prompt.synthetic.length > 0) {
       yield* input.client.session
@@ -236,6 +243,11 @@ export const make = Effect.fnUntraced(function* (input: {
         })
         .pipe(Effect.catch(ACPClient.classify))
     }
+  })
+
+  const submit = Effect.fnUntraced(function* (attached: Attached, prompt: ACPPrompt.Prepared) {
+    const sessionID = attached.id
+    yield* submitSynthetic(attached, prompt)
     if (prompt.start.type === "compaction") {
       yield* input.client.session.compact({ sessionID, id: prompt.start.id }).pipe(Effect.catch(ACPClient.classify))
       return
@@ -391,6 +403,25 @@ export const make = Effect.fnUntraced(function* (input: {
       yield* aborted(signal).pipe(Effect.andThen(Fiber.interrupt(turn)), Effect.forkChild)
       return yield* Fiber.join(turn)
     }),
+    steer: Effect.fn("cli.acp.turn.steer")(function* (params) {
+      const attached = yield* input.sessions.require(params.sessionId)
+      const active = yield* input.client.session.active().pipe(Effect.catch(ACPClient.classify))
+      if (!(attached.id in active)) return { outcome: "promptRequired", reason: "noRunningTurn" } as const
+      const prompt = yield* ACPPrompt.prepare(yield* input.catalog.get(attached.cwd), params.prompt)
+      yield* submitSynthetic(attached, prompt)
+      // Without resume, a session that went idle after the check keeps the steer queued instead of starting a turn.
+      yield* input.client.session
+        .prompt({
+          sessionID: attached.id,
+          id: prompt.start.id,
+          text: prompt.text,
+          files: prompt.files,
+          delivery: "steer",
+          resume: false,
+        })
+        .pipe(Effect.catch(ACPClient.classify))
+      return { outcome: "injected" } as const
+    }),
     cancel: Effect.fnUntraced(function* (params) {
       yield* FiberMap.remove(turns, params.sessionId)
     }),

packages/cli/test/acp/session.test.ts

@@ -25,6 +25,7 @@ describe("acp session lifecycle over the wire", () => {
         _meta: { "opencode/child-session-updates": true },
       },
       agentInfo: { name: "OpenCode" },
+      _meta: { steering: { supported: true } },
     })
     expect(plain.authMethods).toEqual([
       { id: "opencode-login", name: "Login with opencode", description: "Run `opencode auth login` in the terminal" },

packages/cli/test/acp/turn.test.ts

@@ -162,6 +162,19 @@ describe("acp turn events over the wire", () => {
     ])
     expect(acp.elicitations).toEqual([])
   })
+
+  test("steers a running session and leaves an idle one alone", async () => {
+    await using acp = await startSession()
+    const steer = () =>
+      acp.requestExt("_session/steering", { sessionId: acp.sessionId, prompt: [{ type: "text", text: "redirect" }] })
+
+    expect(await steer()).toEqual({ outcome: "promptRequired", reason: "noRunningTurn" })
+    expect(acp.server.submissions).toEqual([])
+
+    acp.server.active.add(acp.sessionId)
+    expect(await steer()).toEqual({ outcome: "injected" })
+    expect(acp.server.prompts).toEqual([expect.objectContaining({ text: "redirect", delivery: "steer", resume: false })])
+  })
 })
 
 function turnUpdates(updates: readonly SessionNotification[]) {

packages/cli/test/acp/wire-fixture.ts

@@ -63,7 +63,13 @@ type ServerRequest = {
 
 const Files = Schema.Array(Schema.Struct({ uri: Schema.String, name: Schema.optional(Schema.String) }))
 const Delivery = Schema.optional(Schema.String)
-const PromptBody = Schema.Struct({ id: Schema.String, text: Schema.String, files: Files, delivery: Delivery })
+const PromptBody = Schema.Struct({
+  id: Schema.String,
+  text: Schema.String,
+  files: Files,
+  delivery: Delivery,
+  resume: Schema.optional(Schema.Boolean),
+})
 const CommandBody = Schema.Struct({ name: Schema.String, text: Schema.String, files: Files, delivery: Delivery })
 const CompactBody = Schema.Struct({ id: Schema.String })
 const SyntheticBody = Schema.Struct({
@@ -499,6 +505,12 @@ export async function startWire(options: WireOptions = {}) {
       return until(() => counts.handled >= target, "client handlers for every agent message (is one missing?)")
     })
 
+  const requestExt = (method: string, params: unknown): Promise<unknown> =>
+    connection.agent.request(method, params).finally(() => {
+      const target = counts.sent
+      return until(() => counts.handled >= target, "client handlers for every agent message (is one missing?)")
+    })
+
   const initialize = (capabilities: InitializeOptions = {}) =>
     request("initialize", {
       protocolVersion: 1,
@@ -522,6 +534,7 @@ export async function startWire(options: WireOptions = {}) {
     elicitations,
     logs,
     request,
+    requestExt,
     until,
     initialize,
     notify: <Method extends AgentNotificationMethod>(method: Method, params: AgentNotificationParamsByMethod[Method]) =>
@@ -599,6 +612,7 @@ function startServer(options: WireOptions, changed: () => void) {
       return submissions.filter((item): item is PromptSubmission => item.kind === "prompt")
     },
     selections,
+    active: new Set<string>(),
     interrupts,
     replies,
     cancelledForms,
@@ -736,6 +750,9 @@ function startServer(options: WireOptions, changed: () => void) {
           }),
         ),
       },
+      "/api/session/active": {
+        GET: route(() => Response.json({ data: Object.fromEntries([...fake.active].map((id) => [id, { type: "running" }])) })),
+      },
       "/api/session/:sessionID": {
         GET: route((req) => {
           const session = fake.sessions.get(req.params.sessionID)