diff --git a/src/bundled/goose/goose.ts b/src/bundled/goose/goose.ts index 965a309756..7c55dbc77a 100644 --- a/src/bundled/goose/goose.ts +++ b/src/bundled/goose/goose.ts @@ -53,6 +53,8 @@ export const GOOSE: AcpHooks = { program: "goose", missing: "goose-acp is missing from this Buzz build.", env: ENV, + // goose is asked to close every session, advertised or not. + closeSession: "always", newSessionMeta: (buzz, { title, provider }) => ({ // The session's whole tool set: goose's developer tools and the Buzz // tools, so extensions in the owner's goose config and plugins never load diff --git a/src/bundled/goose/runtime.ts b/src/bundled/goose/runtime.ts index 491ea66b63..1293c9c007 100644 --- a/src/bundled/goose/runtime.ts +++ b/src/bundled/goose/runtime.ts @@ -37,7 +37,7 @@ import { turnPrompt, } from "../claude-code/prompt"; import type { ServeMcp, Spawn, ToolServer } from "./goose"; -import { AgentSessions, ServerPool, SharedProcess } from "./sessions"; +import { AgentSessions, SharedProcess } from "./sessions"; export type Config = Readonly<{ /** A model the provider offers; empty uses the provider's default, or the @@ -108,8 +108,6 @@ type Entry = { export class GooseRuntime { private readonly agents = new Map(); - /** The MCP servers every agent's sessions share. */ - private readonly pool = new ServerPool(); /** The goose every agent's sessions open in. */ private readonly goose = new SharedProcess( (id, options) => this.spawn(id, options), @@ -502,7 +500,6 @@ export class GooseRuntime { spawn: (id, options) => this.spawn(id, options), tools, process: this.goose, - pool: this.pool, onChange: () => this.notify(), // Read as each turn starts, so a change needs no restart. choice: () => ({ diff --git a/src/features/agent-harness/acp/agent-sessions.ts b/src/features/agent-harness/acp/agent-sessions.ts index a575ff1cd6..a8c2a67a93 100644 --- a/src/features/agent-harness/acp/agent-sessions.ts +++ b/src/features/agent-harness/acp/agent-sessions.ts @@ -3,8 +3,8 @@ // conversation's session is closed; its next message opens a new one, shown // the conversation again. A one-off turn (a watch or timer wake) gets a // session of its own, closed when it ends. Every session holds one of the -// host's MCP servers, in a pool the plugin's agents share, and the least -// recently used idle session closes to make room. Nothing is saved: sessions +// host's MCP servers, in one pool every ACP plugin's agents share, and the +// least recently used idle session closes to make room, whichever plugin's. Nothing is saved: sessions // live as long as the app does. import type { AcpProcess } from "./process"; import type { AcpSession } from "./session"; @@ -18,8 +18,8 @@ import type { /** A conversation's session closes after this long without a message. */ export const IDLE_MS = 15 * 60_000; -/** The most MCP servers a pool's sessions hold at once, below the host's 64 for - * every plugin, so other plugins keep room. */ +/** The most MCP servers the pool's sessions hold at once, below the host's 64 + * for every plugin, so plugins not on this client keep room. */ export const POOL_LIMIT = 48; /** A session's hold on one MCP server in the pool. */ @@ -30,8 +30,7 @@ export type Slot = { evict(): Promise; }; -/** The MCP servers a plugin's agents' sessions hold, least recently used - * first. One runtime shares one. */ +/** The MCP servers agents' sessions hold, least recently used first. */ export class ServerPool { private readonly slots = new Set(); constructor(private readonly limit = POOL_LIMIT) {} @@ -71,6 +70,9 @@ export class ServerPool { } } +/** The pool every ACP plugin's sessions share. */ +export const SHARED_POOL = new ServerPool(); + /** What a session is opened with. */ export type Launch = Readonly<{ /** The agent's working directory, absolute or `~/…`. */ @@ -86,8 +88,8 @@ export type SessionsOptions = Readonly<{ tools: ToolServer; /** The process its sessions open in, started when needed. */ process: Readonly<{ get(): Promise }>; - /** The servers its plugin's sessions hold. */ - pool: ServerPool; + /** The servers its sessions hold: the shared pool unless given. */ + pool?: ServerPool; /** The agent's current settings, for a new session. */ launch(): Promise; /** The agent's provider and model now, which each new turn runs on. */ @@ -309,6 +311,9 @@ export class AgentSessions { for (const live of this.once.values()) if (live) void this.close(live); } + private get pool() { + return this.options.pool ?? SHARED_POOL; + } private stale(live: Live) { return live.fingerprint !== this.options.fingerprint(); } @@ -320,7 +325,8 @@ export class AgentSessions { title: string | undefined, use: (live: Live) => unknown, ): Promise { - const { pool, spawn } = this.options; + const { spawn } = this.options; + const { pool } = this; // Read before `launch` reads the settings, so it never describes newer ones. const fingerprint = this.options.fingerprint(); const launch = await this.options.launch(); @@ -408,7 +414,7 @@ export class AgentSessions { } /** Notes what the session has been shown and restarts its idle countdown. */ private touch(key: string, live: Live, seen?: number) { - this.options.pool.touch(live.slot); + this.pool.touch(live.slot); clearTimeout(live.timer); live.timer = setTimeout(() => { if (live.session.busy) this.touch(key, live); diff --git a/src/features/agent-harness/acp/client.test.ts b/src/features/agent-harness/acp/client.test.ts new file mode 100644 index 0000000000..05af182120 --- /dev/null +++ b/src/features/agent-harness/acp/client.test.ts @@ -0,0 +1,373 @@ +// The shared ACP client driving a plain-ACP harness, one without hooks. +import { afterEach, describe, expect, it, vi } from "vitest"; +import { + fakeSpawn as fakeGoose, + flush, +} from "../../../bundled/goose/goose-testing"; +import { SharedProcess } from "../../../bundled/goose/sessions"; +import type { + HostMcpServer, + HostProcess, + HostProcessOptions, +} from "../../host/service"; +import { AgentSessions, ServerPool, SHARED_POOL } from "./agent-sessions"; +import { AcpProcess, REQUEST_TIMEOUT_MS } from "./process"; +import type { AcpHooks, ToolServer } from "./types"; + +type Message = { + id?: number | string; + method?: string; + params?: Record; +}; + +const HOOKS: AcpHooks = { + label: "Harness", + program: "harness", + missing: "Install the harness CLI.", + env: { HOOK: "hook", SHARED: "hook" }, +}; + +const tools: ToolServer = async (_conversation, message) => ({ + jsonrpc: "2.0", + id: message.id, + result: {}, +}); + +/** A plain-ACP harness: `capabilities` is what `initialize` advertises. */ +function fakeHarness( + options: Readonly<{ + capabilities?: Record; + /** Requests it never answers. */ + silent?: readonly string[]; + /** Exits with this code as soon as it starts, as `env` does for a + * missing program. */ + exitCode?: number; + /** Turns stay open until finished. */ + hold?: boolean; + }> = {}, +) { + const started: { options: HostProcessOptions; killed: boolean }[] = []; + const received: Message[] = []; + const turns: { sessionId: string; text: string; finish(): void }[] = []; + let servers = 0; + const serve = async (): Promise => { + const index = ++servers; + let stopped!: () => void; + const closed = new Promise((resolve) => (stopped = resolve)); + return { + url: `http://127.0.0.1:${40_000 + index}/mcp`, + headers: [{ name: "Authorization", value: `Bearer token-${index}` }], + close: async () => stopped(), + closed, + }; + }; + const spawn = async ( + _id: string, + process: HostProcessOptions = {}, + ): Promise => { + let exit!: (code: number | null) => void; + const exited = new Promise((resolve) => (exit = resolve)); + let alive = true; + const record = { options: process, killed: false }; + started.push(record); + const reply = (id: Message["id"], result: unknown) => + queueMicrotask(() => { + if (alive) + process.onStdout?.( + `${JSON.stringify({ jsonrpc: "2.0", id, result })}\n`, + ); + }); + const end = (code: number | null) => { + if (!alive) return; + alive = false; + exit(code); + }; + if (options.exitCode !== undefined) { + const code = options.exitCode; + queueMicrotask(() => { + process.onStderr?.("env: harness: No such file or directory\n"); + end(code); + }); + } + let sessions = 0; + return { + write: async (data) => { + if (!alive) throw new Error("Broken pipe"); + for (const line of data.split("\n").filter(Boolean)) { + const message = JSON.parse(line) as Message; + received.push(message); + const { id, method, params = {} } = message; + if (!method || options.silent?.includes(method)) continue; + if (method === "initialize") + reply(id, { + protocolVersion: 1, + agentCapabilities: options.capabilities ?? { + mcpCapabilities: { http: true }, + }, + }); + else if (method === "session/new") + reply(id, { sessionId: `session-${++sessions}` }); + else if (method === "session/close") reply(id, {}); + else if (method === "session/prompt") { + const text = (params.prompt as { text: string }[])[0]?.text ?? ""; + const finish = () => reply(id, { stopReason: "end_turn" }); + turns.push({ sessionId: String(params.sessionId), text, finish }); + if (!options.hold) finish(); + } + } + }, + end: async () => undefined, + kill: async () => { + record.killed = true; + end(null); + }, + exited, + }; + }; + const sent = (method: string) => + received.filter((message) => message.method === method); + return { spawn, serve, started, received, turns, sent }; +} + +const start = ( + harness: ReturnType, + launch: Partial[2]> = {}, + hooks: AcpHooks = HOOKS, +) => + AcpProcess.start(harness.spawn, hooks, { + cwd: "/work", + serve: harness.serve, + ...launch, + }); + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("starting a harness", () => { + it("passes the plugin's argv unchanged, with the Buzz identity removed last", async () => { + const harness = fakeHarness(); + await start(harness, { + args: ["cursor-agent", "acp", "--model", "x y"], + env: { + SHARED: "launch", + BUZZ_PRIVATE_KEY: "nsec1leak", + BUZZ_RELAY_URL: "wss://leak", + }, + }); + const options = harness.started[0]?.options; + expect(options?.args).toEqual(["cursor-agent", "acp", "--model", "x y"]); + expect(options?.env).toEqual({ + HOOK: "hook", + SHARED: "launch", + BUZZ_PRIVATE_KEY: null, + BUZZ_AUTH_TAG: null, + BUZZ_RELAY_URL: null, + }); + }); + + it("refuses a harness that cannot take Buzz tools over HTTP MCP", async () => { + for (const http of [false, undefined, "yes"]) { + const harness = fakeHarness({ + capabilities: { mcpCapabilities: { http } }, + }); + await expect(start(harness)).rejects.toThrow( + "This Harness cannot reach Buzz tools (no HTTP MCP support)", + ); + expect(harness.started[0]?.killed).toBe(true); + } + }); + + it("names the install hint when the program is missing or cannot run", async () => { + for (const exitCode of [127, 126]) { + const harness = fakeHarness({ exitCode }); + await expect(start(harness)).rejects.toThrow("Install the harness CLI."); + } + }); + + it("reports another early exit as itself", async () => { + const harness = fakeHarness({ exitCode: 1 }); + await expect(start(harness)).rejects.toThrow( + "env: harness: No such file or directory", + ); + }); + + it("stops a harness that never answers initialize", async () => { + vi.useFakeTimers(); + const harness = fakeHarness({ silent: ["initialize"] }); + const starting = start(harness); + const failed = expect(starting).rejects.toThrow( + "Harness did not answer initialize in 60s", + ); + await vi.advanceTimersByTimeAsync(REQUEST_TIMEOUT_MS); + await failed; + expect(harness.started[0]?.killed).toBe(true); + }); + + it("stops the harness when cancelled while it starts", async () => { + const harness = fakeHarness({ silent: ["initialize"] }); + const controller = new AbortController(); + const starting = start(harness, { signal: controller.signal }); + await flush(); + controller.abort(new Error("Check was cancelled")); + await expect(starting).rejects.toThrow("Check was cancelled"); + expect(harness.started[0]?.killed).toBe(true); + }); + + it("starts nothing when already cancelled", async () => { + const harness = fakeHarness(); + const controller = new AbortController(); + controller.abort(new Error("Removed")); + await expect(start(harness, { signal: controller.signal })).rejects.toThrow( + "Removed", + ); + expect(harness.started).toEqual([]); + }); + + it("stops within its time limit, for a check that opens no session", async () => { + const harness = fakeHarness(); + const acp = await start(harness); + expect(acp.agentCapabilities).toEqual({ mcpCapabilities: { http: true } }); + await acp.stop(); + expect(harness.started[0]?.killed).toBe(true); + expect(acp.running).toBe(false); + }); +}); + +describe("a session without hooks", () => { + it("leads the first prompt with the instructions, once, when ACP's systemPrompt is not advertised", async () => { + const harness = fakeHarness(); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "Be Buzz." }); + const [opened] = harness.sent("session/new"); + expect(opened?.params).not.toHaveProperty("systemPrompt"); + expect(opened?.params).not.toHaveProperty("_meta"); + await session.send("first"); + await session.send("second"); + expect(harness.turns.map((turn) => turn.text)).toEqual([ + "Be Buzz.\n\nfirst", + "second", + ]); + }); + + it("sends the instructions as ACP's systemPrompt when advertised", async () => { + const harness = fakeHarness({ + capabilities: { + mcpCapabilities: { http: true }, + sessionCapabilities: { systemPrompt: {} }, + }, + }); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "Be Buzz." }); + expect(harness.sent("session/new")[0]?.params?.systemPrompt).toEqual({ + mode: "append", + content: [{ type: "text", text: "Be Buzz." }], + }); + await session.send("first"); + expect(harness.turns.map((turn) => turn.text)).toEqual(["first"]); + }); + + it("queues a message sent mid-turn as the next turn", async () => { + const harness = fakeHarness({ hold: true }); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "" }); + const first = session.send("first"); + await flush(); + const second = session.send("second"); + await flush(); + expect(harness.turns.map((turn) => turn.text)).toEqual(["first"]); + harness.turns[0]?.finish(); + await flush(); + expect(harness.turns.map((turn) => turn.text)).toEqual(["first", "second"]); + harness.turns[1]?.finish(); + await expect(Promise.all([first, second])).resolves.toEqual([ + { ok: true }, + { ok: true }, + ]); + }); + + it("asks the harness to close a session only when it advertises close", async () => { + for (const close of [undefined, {}]) { + const harness = fakeHarness({ + capabilities: { + mcpCapabilities: { http: true }, + sessionCapabilities: close ? { close } : {}, + }, + }); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "" }); + await acp.close(session); + expect(harness.sent("session/cancel")).toHaveLength(1); + expect(harness.sent("session/close")).toHaveLength(close ? 1 : 0); + expect(session.open).toBe(false); + } + }); +}); + +describe("the shared server pool", () => { + it("is used by every agent's sessions unless one is given", async () => { + const harness = fakeHarness(); + const acp = await start(harness); + const sessions = new AgentSessions({ + label: "Harness", + spawn: harness.spawn, + tools, + process: { get: async () => acp }, + launch: async () => ({ workspace: "/work", systemPrompt: "" }), + choice: () => ({ provider: "", model: "" }), + fingerprint: () => "", + }); + const before = SHARED_POOL.size; + await sessions.deliver("conversation", "hello", 1); + expect(SHARED_POOL.size).toBe(before + 1); + sessions.dispose(); + await flush(); + expect(SHARED_POOL.size).toBe(before); + }); + + it("closes the idle session of either plugin to make room for the other", async () => { + const pool = new ServerPool(1); + const goose = fakeGoose({ home: "/home/test" }); + const harness = fakeHarness(); + const launch = async () => ({ workspace: "/work", systemPrompt: "" }); + const choice = () => ({ provider: "", model: "" }); + const gooseSessions = new AgentSessions({ + label: "goose", + spawn: goose.spawn, + tools, + process: new SharedProcess(goose.spawn, goose.serve), + pool, + launch, + choice, + fingerprint: () => "", + }); + const acp = await start(harness); + const genericSessions = new AgentSessions({ + label: "Harness", + spawn: harness.spawn, + tools, + process: { get: async () => acp }, + pool, + launch, + choice, + fingerprint: () => "", + }); + + await gooseSessions.deliver("goose chat", "hello", 1); + expect(gooseSessions.snapshot().map((item) => item.key)).toEqual([ + "goose chat", + ]); + await genericSessions.deliver("generic chat", "hello", 1); + expect(gooseSessions.snapshot()).toEqual([]); + expect(genericSessions.snapshot().map((item) => item.key)).toEqual([ + "generic chat", + ]); + expect(goose.geese[0]?.closed).toEqual(["session-1"]); + + await gooseSessions.deliver("goose chat", "again", 2); + expect(genericSessions.snapshot()).toEqual([]); + expect(pool.size).toBe(1); + gooseSessions.dispose(); + genericSessions.dispose(); + }); +}); diff --git a/src/features/agent-harness/acp/process.ts b/src/features/agent-harness/acp/process.ts index 5540c5ec90..23ca795622 100644 --- a/src/features/agent-harness/acp/process.ts +++ b/src/features/agent-harness/acp/process.ts @@ -30,11 +30,22 @@ export const REQUEST_TIMEOUT_MS = 60_000; * `buzz__`. */ export const TOOL_SERVER = "buzz"; +/** How long a harness that failed to start has to exit once stopped. */ +const STOP_TIMEOUT_MS = 5_000; + const STDERR_LIMIT = 4_000; -/** The hooks every harness on this client provides so far. One without them - * is not supported yet. */ -export type Hooks = Required; +/** Never reaches a harness, whatever its hooks or launch say: its Buzz tools + * come over HTTP MCP, never a Buzz key. */ +export const PROTECTED_ENV: Readonly> = { + BUZZ_PRIVATE_KEY: null, + BUZZ_AUTH_TAG: null, + BUZZ_RELAY_URL: null, +}; + +/** Exit codes for a program that cannot be found or run, as `env` reports + * them. */ +const MISSING_CODES = new Set([126, 127]); type Wire = { jsonrpc?: string; @@ -57,15 +68,17 @@ export class AcpProcess { private buffer = ""; private stderr = ""; private alive = true; + private code: number | null | undefined; private capabilities: Readonly> = {}; private constructor( private readonly process: HostProcess, - readonly hooks: Hooks, + readonly hooks: AcpHooks, private readonly serve: ServeMcp | undefined, ) { this.exited = process.exited.then((code) => { this.alive = false; + this.code = code; const error = new Error( this.stderr.trim().split("\n").at(-1) || `${hooks.label} exited (${code ?? "signal"})`, @@ -81,33 +94,43 @@ export class AcpProcess { }); } - /** Starts the harness and checks it can reach the Buzz tools. Sessions - * with tools need `serve`, which starts their MCP server. */ + /** Starts the harness with `args` and checks it can reach the Buzz tools. + * Sessions with tools need `serve`, which starts their MCP server. Once + * `signal` aborts, it stops the harness and rejects. */ static async start( spawn: Spawn, hooks: AcpHooks, launch: Readonly<{ cwd: string; + args?: readonly string[]; env?: Readonly>; serve?: ServeMcp; + signal?: AbortSignal; }>, ) { - const supported = supportedHooks(hooks); + const { signal } = launch; + signal?.throwIfAborted(); let acp: AcpProcess | undefined; const early: string[] = []; + const earlyErrors: string[] = []; const process = await spawn(hooks.program, { cwd: launch.cwd, - env: { ...hooks.env, ...launch.env }, + ...(launch.args ? { args: launch.args } : {}), + env: { ...hooks.env, ...launch.env, ...PROTECTED_ENV }, onStdout: (data) => (acp ? acp.read(data) : early.push(data)), - onStderr: (data) => acp?.readError(data), + onStderr: (data) => (acp ? acp.readError(data) : earlyErrors.push(data)), }).catch((error: unknown) => { throw /os error 2|no such file|not found/i.test(errorMessage(error)) ? new Error(hooks.missing) : error; }); - acp = new AcpProcess(process, supported, launch.serve); + acp = new AcpProcess(process, hooks, launch.serve); + for (const data of earlyErrors) acp.readError(data); for (const data of early) acp.read(data); + const abort = () => void process.kill(); + signal?.addEventListener("abort", abort, { once: true }); try { + signal?.throwIfAborted(); const init = (await acp.request( "initialize", { @@ -121,14 +144,20 @@ export class AcpProcess { mcpCapabilities?: { http?: boolean }; }; }; + signal?.throwIfAborted(); acp.capabilities = init.agentCapabilities ?? {}; - if (!init.agentCapabilities?.mcpCapabilities?.http) + if (init.agentCapabilities?.mcpCapabilities?.http !== true) throw new Error( `This ${hooks.label} cannot reach Buzz tools (no HTTP MCP support)`, ); } catch (error) { - void process.kill(); + await acp.stop(STOP_TIMEOUT_MS); + signal?.throwIfAborted(); + // A program `env` cannot find or run exits before it answers. + if (MISSING_CODES.has(acp.code ?? 0)) throw new Error(hooks.missing); throw error; + } finally { + signal?.removeEventListener("abort", abort); } return acp; } @@ -143,6 +172,15 @@ export class AcpProcess { get agentCapabilities() { return this.capabilities; } + /** Whether `initialize` advertised `sessionCapabilities[name]`. */ + advertises(name: string) { + const session = this.capabilities.sessionCapabilities as + | Record + | null + | undefined; + const value = session?.[name]; + return typeof value === "object" && value !== null; + } /** Opens a new session in `cwd`, named `title`, whose system prompt has * `systemPrompt` appended. `tools` answers its Buzz tools; without them it @@ -159,6 +197,9 @@ export class AcpProcess { ): Promise { const { hooks } = this; const provider = launch.choice?.().provider ?? ""; + const { systemPrompt } = launch; + const prompting = !!systemPrompt && !hooks.systemPrompt; + const standard = prompting && this.advertises("systemPrompt"); let server: HostMcpServer | undefined; // The harness connects while it opens the session, before it is known here. let session: AcpSession | undefined; @@ -192,9 +233,17 @@ export class AcpProcess { { cwd: launch.cwd, mcpServers: buzz ? [buzz] : [], - _meta: hooks.newSessionMeta(buzz, { - ...(launch.title ? { title: launch.title } : {}), - provider, + ...(standard && { + systemPrompt: { + mode: "append", + content: [{ type: "text", text: systemPrompt }], + }, + }), + ...(hooks.newSessionMeta && { + _meta: hooks.newSessionMeta(buzz, { + ...(launch.title ? { title: launch.title } : {}), + provider, + }), }), }, REQUEST_TIMEOUT_MS, @@ -203,18 +252,22 @@ export class AcpProcess { void server?.close(); throw error; } - session = new AcpSession(this, sessionId, server, launch.choice, { - provider, - model: "", - }); + session = new AcpSession( + this, + sessionId, + server, + launch.choice, + { provider, model: "" }, + prompting && !standard ? systemPrompt : "", + ); this.sessions.set(sessionId, session); if (stopped) { void this.close(session); throw new Error(STOPPED); } try { - if (launch.systemPrompt) - await hooks.systemPrompt(this, sessionId, launch.systemPrompt); + if (systemPrompt && hooks.systemPrompt) + await hooks.systemPrompt(this, sessionId, systemPrompt); } catch (error) { void this.close(session); throw error; @@ -222,13 +275,16 @@ export class AcpProcess { return session; } - /** Ends a session. Its turn, if any, is cancelled first. */ + /** Ends a session. Its turn, if any, is cancelled first. The harness is + * asked to close it only when it says it can. */ async close(session: AcpSession) { if (!this.sessions.delete(session.id)) return; session.fail("The session was closed"); void session.server?.close(); if (!this.alive) return; this.notify("session/cancel", { sessionId: session.id }); + if (this.hooks.closeSession !== "always" && !this.advertises("close")) + return; await this.request( "session/close", { sessionId: session.id }, @@ -242,6 +298,20 @@ export class AcpProcess { return this.process.kill(); } + /** Stops the harness, and resolves once it has exited or `timeout` ms + * have passed, whichever is first. */ + async stop(timeout = REQUEST_TIMEOUT_MS) { + let timer: ReturnType | undefined; + await Promise.race([ + this.kill() + .catch(() => undefined) + .then(() => this.exited), + new Promise((resolve) => { + timer = setTimeout(resolve, timeout); + }), + ]).finally(() => clearTimeout(timer)); + } + /** Sends a request and resolves with its result. When the harness does not * answer within `timeout` ms it is stopped, which fails its sessions. */ request(method: string, params: Record, timeout?: number) { @@ -372,12 +442,3 @@ export class AcpProcess { return response; } } - -/** `hooks`, once it has every hook: a harness that lacks one is not - * supported yet. */ -function supportedHooks(hooks: AcpHooks): Hooks { - const { newSessionMeta, systemPrompt, steering, select } = hooks; - if (!newSessionMeta || !systemPrompt || !steering || !select) - throw new Error(`${hooks.label} is unsupported: it lacks an ACP hook`); - return { ...hooks, newSessionMeta, systemPrompt, steering, select }; -} diff --git a/src/features/agent-harness/acp/session.ts b/src/features/agent-harness/acp/session.ts index baca5e4425..1a2fceaca5 100644 --- a/src/features/agent-harness/acp/session.ts +++ b/src/features/agent-harness/acp/session.ts @@ -9,8 +9,8 @@ const WRITTEN_LIMIT = 20_000; type Waiter = (settled: Settled) => void; /** One conversation's ACP session. Messages sent while a turn runs are - * steered into it; one that arrives as the turn ends, or asks for another - * model, runs as the next turn. */ + * steered into it, when the harness can steer; one that arrives as the turn + * ends, or asks for another model, runs as the next turn. */ export class AcpSession { /** The conversation it serves. */ conversation: string | undefined; @@ -49,6 +49,9 @@ export class AcpSession { private readonly choice?: () => Choice, /** The provider and model it runs on, as chosen here. */ public chosen: Choice = { provider: "", model: "" }, + /** Instructions that lead its first prompt, for a harness that takes no + * system prompt. */ + private preamble = "", ) {} get busy() { @@ -77,6 +80,7 @@ export class AcpSession { // A steer cannot change the model, and must not pass a queued turn. if ( this.prompting && + this.acp.hooks.steering && !this.queue.length && (!model || model === this.model) ) @@ -110,7 +114,7 @@ export class AcpSession { if (typeof content?.text === "string") this.written = (this.written + content.text).slice(0, WRITTEN_LIMIT); } - const runId = this.acp.hooks.steering.runId(update); + const runId = this.acp.hooks.steering?.runId(update) ?? null; if (runId !== null) { this.runId = runId; if (runId) this.ranAs(runId); @@ -147,9 +151,10 @@ export class AcpSession { this.runWaiters.push(resolve), ) : undefined); - if (runId) { + const { steering } = this.acp.hooks; + if (runId && steering) { try { - await this.acp.hooks.steering.steer(this.acp, this.id, runId, text); + await steering.steer(this.acp, this.id, runId, text); this.acknowledge(activity, waiter); return; } catch { @@ -173,9 +178,13 @@ export class AcpSession { const { text, model } = this.queue.shift() ?? { text: "" }; try { await this.select(model); + const lead = this.preamble; + this.preamble = ""; const { stopReason } = (await this.acp.request("session/prompt", { sessionId: this.id, - prompt: [{ type: "text", text }], + prompt: [ + { type: "text", text: lead ? `${lead}\n\n${text}` : text }, + ], })) as { stopReason?: string }; if (stopReason === "refusal") this.lastError = `${this.acp.hooks.label} refused the request`; @@ -199,7 +208,7 @@ export class AcpSession { if (!chosen) return; const want = model ? { ...chosen, model } : chosen; this.model = want.model; - await this.acp.hooks.select(this.acp, this, want); + await this.acp.hooks.select?.(this.acp, this, want); } private ranAs(runId: string | undefined) { diff --git a/src/features/agent-harness/acp/types.ts b/src/features/agent-harness/acp/types.ts index afe2fdac81..fa1ea09e4e 100644 --- a/src/features/agent-harness/acp/types.ts +++ b/src/features/agent-harness/acp/types.ts @@ -50,28 +50,33 @@ export type McpServer = Readonly<{ headers: HostMcpServer["headers"]; }>; -/** What a harness adds to, or does in place of, plain ACP. */ +/** What a harness adds to, or does in place of, plain ACP. Each hook is + * optional: without it the client uses plain ACP's way, or none. */ export type AcpHooks = Readonly<{ /** Names the harness in errors. */ label: string; /** The manifest process it runs as. */ program: string; - /** The error when that program cannot be found. */ + /** The error when that program cannot be found, or cannot run. */ missing: string; - /** Its environment, under the launch's own. */ + /** Its environment, under the launch's own. The Buzz identity is always + * removed, whatever either says. */ env: Readonly>; /** `session/new`'s `_meta`. */ newSessionMeta?( buzz: McpServer | undefined, open: Readonly<{ title?: string; provider: string }>, ): Record; - /** Appends `text` to a new session's system prompt; a throw closes it. */ + /** Appends `text` to a new session's system prompt; a throw closes it. + * Without it, ACP's `systemPrompt` carries it when advertised, else it + * leads the session's first prompt. */ systemPrompt?( process: AcpProcess, sessionId: string, text: string, ): Promise; - /** Steers a message into the running turn. */ + /** Steers a message into the running turn. Without it, a message sent + * mid-turn runs as the next turn. */ steering?: Readonly<{ /** The running turn's id from an update: undefined when none runs, * null when the update says nothing about it. */ @@ -84,10 +89,13 @@ export type AcpHooks = Readonly<{ ): Promise; }>; /** Puts a session on `want` before a turn, keeping `session.chosen` to - * what it now runs on. */ + * what it now runs on. Without it, the harness's own pick stands. */ select?( process: AcpProcess, session: { readonly id: string; chosen: Choice }, want: Choice, ): Promise; + /** Whether `session/close` is sent: always, or only when + * `sessionCapabilities.close` is advertised (the default). */ + closeSession?: "always" | "advertised"; }>;