diff --git a/packages/cli/src/acp/agent.ts b/packages/cli/src/acp/agent.ts index bb2f7c809fd2..bbfd4b2537e9 100644 --- a/packages/cli/src/acp/agent.ts +++ b/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((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" diff --git a/packages/cli/src/acp/service.ts b/packages/cli/src/acp/service.ts index 70b723bd37a4..5bd9e326045f 100644 --- a/packages/cli/src/acp/service.ts +++ b/packages/cli/src/acp/service.ts @@ -62,6 +62,7 @@ export interface Interface { ) => Effect.Effect readonly setSessionMode: (input: SetSessionModeRequest) => Effect.Effect readonly prompt: (input: PromptRequest, signal: AbortSignal) => Effect.Effect + readonly steer: (input: ACPTurn.SteeringRequest) => Effect.Effect readonly cancel: (input: CancelNotification) => Effect.Effect } @@ -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, } } diff --git a/packages/cli/src/acp/turn.ts b/packages/cli/src/acp/turn.ts index c02aeac4ef51..f43d82eb2264 100644 --- a/packages/cli/src/acp/turn.ts +++ b/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 + /** Redirects a running turn; an idle session is left alone so a steer never starts a turn. */ + readonly steer: (input: SteeringRequest) => Effect.Effect readonly cancel: (input: CancelNotification) => Effect.Effect /** Unlike `cancel`, interrupts an idle session too, since server work can outlive its turn. */ readonly close: (sessionID: string) => Effect.Effect @@ -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) }), diff --git a/packages/cli/test/acp/session.test.ts b/packages/cli/test/acp/session.test.ts index bcb3a9137da2..7abf031a0022 100644 --- a/packages/cli/test/acp/session.test.ts +++ b/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" }, diff --git a/packages/cli/test/acp/turn.test.ts b/packages/cli/test/acp/turn.test.ts index 2a49099a9db3..54bfa5a8aa75 100644 --- a/packages/cli/test/acp/turn.test.ts +++ b/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[]) { diff --git a/packages/cli/test/acp/wire-fixture.ts b/packages/cli/test/acp/wire-fixture.ts index 67ad107231bb..0af8843b03b3 100644 --- a/packages/cli/test/acp/wire-fixture.ts +++ b/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 => + 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: 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(), 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)