diff --git a/src/bundled/goose/goose.ts b/src/bundled/goose/goose.ts index 66d44b02d7..54cf26f3d7 100644 --- a/src/bundled/goose/goose.ts +++ b/src/bundled/goose/goose.ts @@ -1,77 +1,43 @@ -// One `goose-acp` process, driven over ACP's line-delimited JSON-RPC. Each -// conversation an agent is in is one ACP session in it, in the agent's -// workspace. Each session gets its own loopback HTTP MCP server for its agent's -// Buzz tools, which the app serves and answers: goose is given its address and -// token in `session/new`, never a Buzz key. -import { steerPrompt } from "../claude-code/prompt"; +// goose on the shared ACP client: one `goose-acp` process, with goose's +// extensions for the system prompt, steering, and per-turn provider and model +// picks. Each session's Buzz tools reach goose over HTTP MCP, never a Buzz key. +import { + AcpProcess, + REQUEST_TIMEOUT_MS, +} from "../../features/agent-harness/acp/process"; +import type { AcpSession } from "../../features/agent-harness/acp/session"; +import { errorMessage } from "../../features/agent-harness/acp/steps"; import type { - HostMcpServer, - HostProcess, - HostProcessOptions, - McpHandler, -} from "../../features/host/service"; - -export type Spawn = ( - id: string, - options?: HostProcessOptions, -) => Promise; -export type ServeMcp = (handler: McpHandler) => Promise; + AcpHooks, + ServeMcp, + Spawn, +} from "../../features/agent-harness/acp/types"; + +export { + REQUEST_TIMEOUT_MS, + TOOL_SERVER, +} from "../../features/agent-harness/acp/process"; +export { stepLabel } from "../../features/agent-harness/acp/steps"; +export type { + ActivityReceipt, + Choice, + ServeMcp, + Settled, + Spawn, + ToolServer, +} from "../../features/agent-harness/acp/types"; + +export type GooseProcess = AcpProcess; +export type GooseSession = AcpSession; /** The desktop host's `spawn` and `resolveDirectory`, which goose starts with. */ export type GooseHost = Readonly<{ spawn: Spawn; resolveDirectory(path: string): Promise; }>; -/** The most of a turn's text kept, to post when it answers nowhere in Buzz. */ -const WRITTEN_LIMIT = 20_000; -/** The Buzz tools that answer where people see it. A read, or a send that - * failed, still leaves goose's text to post. */ -const ANSWERS = new Set(["send", "edit", "react"]); -const STOPPED = - "Buzz stopped this session's tools: another program used their token"; - -/** How long goose has to answer a request that starts it or opens or closes - * a session. One that does not is stopped. A turn's prompt is never timed. */ -export const REQUEST_TIMEOUT_MS = 60_000; - -/** How a burst of work ended: every message sent so far has been answered, - * or the session or its process is gone. One message of the burst owns its - * `error` and `text`, to post them once; the others get only `ok`. */ -export type Settled = Readonly<{ - ok: boolean; - error?: string; - /** What goose wrote, when it answered nowhere in Buzz. Nobody saw it. */ - text?: string; - /** The message gave up before any turn took it in. */ - undelivered?: true; - /** The owner's interrupt ended the turn. */ - stopped?: true; -}>; - -export type ActivityReceipt = Readonly<{ - current?: () => boolean; - /** Rebuilds an Attention prompt after startup or queue waits. */ - refresh?: (fresh?: boolean) => Promise; - /** The prompt was handed to goose, after cancellation and provider checks. */ - delivered?: () => void; - /** Steps show on the latest acknowledged message of each scope. */ - activityScope?: string; - step?: (tool: string, target?: string) => void; - acknowledged?: () => void; - superseded?: () => void; -}>; - -/** Answers one MCP message from a session serving `conversation` with its - * JSON-RPC response. */ -export type ToolServer = ( - conversation: string | undefined, - message: Readonly<{ id: string | number; method: string; params?: unknown }>, -) => Promise; - -/** The HTTP MCP server every session declares; goose names its tools - * `buzz__`. */ -export const TOOL_SERVER = "buzz"; const DEVELOPER = { type: "platform", name: "developer" }; +/** goose's provider id for the provider and model its config picks. */ +const CONFIGURED = "goose"; /** What every `goose-acp` runs with: tools run without asking, and no Buzz * identity reaches it from the app's environment. Nor does a provider or @@ -91,916 +57,132 @@ const databricks = (): Record => { return host ? { DATABRICKS_HOST: host } : {}; }; -/** The provider, model and thinking level a session's turns run on; empty is - * what goose's config picks. */ -export type Choice = Readonly<{ - provider: string; - model: string; - effort?: string; -}>; -/** goose's provider id for the provider and model its config picks. */ -const CONFIGURED = "goose"; - -const STDERR_LIMIT = 4_000; -const STEP_KEYS = [ - "description", - "path", - "file_path", - "pattern", - "url", - "query", - "command", -]; - -type Wire = { - jsonrpc?: string; - id?: number | string | null; - method?: string; - params?: Record; - result?: unknown; - error?: { code?: number; message: string; data?: unknown }; -}; -type Waiter = (settled: Settled) => void; -type Pending = { - resolve: (result: unknown) => void; - reject: (error: Error) => void; -}; - -export class GooseProcess { - readonly exited: Promise; - private readonly sessions = new Map(); - private readonly calls = new Map(); - private requests = 0; - private buffer = ""; - private stderr = ""; - private alive = true; - /** goose names an untitled session with a model call, unless its config - * sets `GOOSE_DISABLE_SESSION_NAMING`; `names` reads it once. */ - private naming: Promise | undefined; - - private constructor( - private readonly process: HostProcess, - private readonly serve: ServeMcp | undefined, - ) { - this.exited = process.exited.then((code) => { - this.alive = false; - const error = new Error( - this.stderr.trim().split("\n").at(-1) || - `goose exited (${code ?? "signal"})`, - ); - for (const call of this.calls.values()) call.reject(error); - this.calls.clear(); - for (const session of this.sessions.values()) { - session.fail(error.message); - void session.server?.close(); - } - this.sessions.clear(); - return code; - }); - } - - /** Starts `goose-acp` and checks it can reach the Buzz tools. Sessions - * with tools need `serve`, which starts their MCP server. */ - static async start( - spawn: Spawn, - launch: Readonly<{ - cwd: string; - env?: Readonly>; - serve?: ServeMcp; - }>, - ) { - let goose: GooseProcess | undefined; - const early: string[] = []; - const process = await spawn("goose", { - cwd: launch.cwd, - env: { ...ENV, ...databricks(), ...launch.env }, - onStdout: (data) => (goose ? goose.read(data) : early.push(data)), - onStderr: (data) => goose?.readError(data), - }).catch((error: unknown) => { - throw /os error 2|no such file|not found/i.test(errorMessage(error)) - ? new Error("goose-acp is missing from this Buzz build.") - : error; - }); - goose = new GooseProcess(process, launch.serve); - for (const data of early) goose.read(data); - try { - const init = (await goose.request( - "initialize", - { - protocolVersion: 1, - clientCapabilities: {}, - clientInfo: { name: "buzz", version: "1" }, - }, +export const GOOSE: AcpHooks = { + label: "goose", + 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", + // goose names an untitled session with a model call, unless its config + // sets `GOOSE_DISABLE_SESSION_NAMING`. + names: (goose) => + goose + .request( + "_goose/unstable/config/read", + { key: "GOOSE_DISABLE_SESSION_NAMING", isSecret: false }, REQUEST_TIMEOUT_MS, - )) as { - agentCapabilities?: { mcpCapabilities?: { http?: boolean } }; - }; - if (!init.agentCapabilities?.mcpCapabilities?.http) - throw new Error( - "This goose cannot reach Buzz tools (no HTTP MCP support)", - ); - } catch (error) { - void process.kill(); - throw error; - } - return goose; - } - - get running() { - return this.alive; - } - get busy() { - return [...this.sessions.values()].some((session) => session.busy); - } - - /** Opens a new session in `cwd`, named `title`, whose system prompt has - * `systemPrompt` appended. With `named`, goose names it instead when it - * names sessions, and `named` takes that name. `tools` answers its Buzz - * tools; without them it has no tools at all. Each turn runs on the - * provider and model `choice` gives as it starts. */ - async open( - launch: Readonly<{ - cwd: string; - systemPrompt: string; - title?: string; - named?: (title: string) => void; - tools?: ToolServer; - choice?: () => Choice; - }>, - ): Promise { - const named = - launch.named && (await this.names()) ? launch.named : undefined; - const title = named ? undefined : launch.title; - const provider = launch.choice?.().provider ?? ""; - let server: HostMcpServer | undefined; - // goose connects while it opens the session, before it is known here. - let session: GooseSession | undefined; - const { tools } = launch; - if (tools) { - if (!this.serve) throw new Error("This goose serves no Buzz tools"); - server = await this.serve((message) => this.mcp(tools, session, message)); - } - const buzz = server && { - type: "http", - name: TOOL_SERVER, - url: server.url, - headers: server.headers, - }; - // The host stops a server whose token a second client used, so the - // session's tools may be someone else's. A session closed here has - // already left `sessions`. - let stopped = false; - void server?.closed.then(() => { - stopped = true; - if (session && this.sessions.get(session.id) === session) { - session.fail(STOPPED); - void this.close(session); - } - }); - let sessionId: string; - try { - ({ sessionId } = (await this.request( - "session/new", - { - cwd: launch.cwd, - mcpServers: buzz ? [buzz] : [], - _meta: { - // 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 into a Buzz agent. goose stores this list - // with the session. Since aaif-goose/goose#12908 it leaves out - // the Buzz entry and holds the `mcpServers` copy in memory, so - // the token stays out of sessions.db; older goose uses this one. - enabledExtensions: buzz - ? [DEVELOPER, { type: "mcp", server: buzz }] - : [], - ...(title ? { sessionTitle: title } : {}), - // On the provider's default model, whatever goose's config holds. - ...(provider ? { provider } : {}), - }, + ) + .then( + (result) => { + const { value } = result as { value?: unknown }; + return value !== true && value !== "true"; }, - REQUEST_TIMEOUT_MS, - )) as { sessionId: string }); - } catch (error) { - void server?.close(); - throw error; - } - session = new GooseSession(this, sessionId, server, launch.choice, { - provider, - model: "", - }); - session.named = named; - this.sessions.set(sessionId, session); - if (stopped) { - void this.close(session); - throw new Error(STOPPED); - } - try { - // Appended under goose's own prompt, which carries each extension's - // instructions; `set` would replace it, and render ours as a template. - if (launch.systemPrompt) - await this.request( - "_goose/unstable/session/system-prompt/set", - { sessionId, mode: "append", key: "buzz", text: launch.systemPrompt }, - REQUEST_TIMEOUT_MS, - ); - } catch (error) { - void this.close(session); - throw error; - } - return session; - } - - private names() { - this.naming ??= this.request( - "_goose/unstable/config/read", - { key: "GOOSE_DISABLE_SESSION_NAMING", isSecret: false }, + () => true, + ), + 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 + // into a Buzz agent. goose stores this list with the session. Since + // aaif-goose/goose#12908 it leaves out the Buzz entry and holds the + // `mcpServers` copy in memory, so the token stays out of sessions.db; + // older goose uses this one. + enabledExtensions: buzz ? [DEVELOPER, { type: "mcp", server: buzz }] : [], + ...(title ? { sessionTitle: title } : {}), + // On the provider's default model, whatever goose's config holds. + ...(provider ? { provider } : {}), + }), + // Appended under goose's own prompt, which carries each extension's + // instructions; `set` would replace it, and render ours as a template. + systemPrompt: async (goose, sessionId, text) => { + await goose.request( + "_goose/unstable/session/system-prompt/set", + { sessionId, mode: "append", key: "buzz", text }, REQUEST_TIMEOUT_MS, - ).then( - (result) => { - const { value } = result as { value?: unknown }; - return value !== true && value !== "true"; - }, - () => true, ); - return this.naming; - } - - /** Ends a session. Its turn, if any, is cancelled first. */ - async close(session: GooseSession) { - 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 }); - await this.request( - "session/close", - { sessionId: session.id }, - REQUEST_TIMEOUT_MS, - ).catch((error) => console.warn("goose did not close a session", error)); - } - - kill() { - return this.process.kill(); - } - - /** Sends a request and resolves with its result. When goose does not - * answer within `timeout` ms it is stopped, which fails its sessions. */ - request(method: string, params: Record, timeout?: number) { - if (!this.alive) return Promise.reject(new Error("goose is not running")); - const id = ++this.requests; - const answer = new Promise((resolve, reject) => { - if (timeout === undefined) return this.calls.set(id, { resolve, reject }); - const timer = setTimeout(() => { - this.calls.delete(id); - reject( - new Error( - `goose did not answer ${method} in ${Math.round(timeout / 1000)}s`, - ), - ); - void this.kill(); - }, timeout); - this.calls.set(id, { - resolve: (result) => { - clearTimeout(timer); - resolve(result); - }, - reject: (error) => { - clearTimeout(timer); - reject(error); - }, - }); - }); - this.write({ jsonrpc: "2.0", id, method, params }).catch((error) => { - this.calls.get(id)?.reject(new Error(String(error))); - this.calls.delete(id); - }); - return answer; - } - notify(method: string, params: Record) { - void this.write({ jsonrpc: "2.0", method, params }).catch(() => undefined); - } - - private write(message: Wire) { - return this.process.write(`${JSON.stringify(message)}\n`); - } - private readError(data: string) { - this.stderr = (this.stderr + data).slice(-STDERR_LIMIT); - } - private read(data: string) { - this.buffer += data; - let newline = this.buffer.indexOf("\n"); - while (newline >= 0) { - const line = this.buffer.slice(0, newline).trim(); - this.buffer = this.buffer.slice(newline + 1); - newline = this.buffer.indexOf("\n"); - if (!line) continue; - let message: Wire; - try { - message = JSON.parse(line); - } catch { - continue; - } - this.handle(message); - } - } - private handle(message: Wire) { - const { id, method } = message; - if (method === undefined) { - if (typeof id !== "number") return; - const call = this.calls.get(id); - this.calls.delete(id); - if (message.error) call?.reject(new Error(errorMessage(message.error))); - else call?.resolve(message.result); - return; - } - if (id === undefined || id === null) { - if (method === "session/update") this.update(message.params ?? {}); - return; - } - void this.answer(id, method, message.params ?? {}); - } - private update(params: Record) { - const session = this.sessions.get(String(params.sessionId)); - const update = params.update as Record | undefined; - if (session && update) session.update(update); - } - /** Answers what goose asks its client: since tools run without asking, - * any permission prompt with allow. */ - private async answer( - id: number | string, - method: string, - params: Record, - ) { - let reply: Wire; - if (method === "session/request_permission") - reply = { jsonrpc: "2.0", id, result: { outcome: allow(params) } }; - else - reply = { - jsonrpc: "2.0", - id, - error: { code: -32601, message: `Buzz does not handle ${method}` }, - }; - return this.write(reply).catch(() => undefined); - } - /** One MCP message for a session's Buzz tools, answered with its JSON-RPC - * response; a notification gets none. A final `send` that succeeded ends - * its turn. The host answers a throw with an internal error. */ - private async mcp( - tools: ToolServer, - session: GooseSession | undefined, - message: unknown, - ): Promise { - const { id, method = "", params } = (message ?? {}) as Wire; - if (id === undefined || id === null) return undefined; - const response = (await tools(session?.conversation, { - id, - method, - params, - })) as Wire; - const call = (params ?? {}) as { - name?: unknown; - arguments?: { final?: unknown }; - }; - const worked = - method === "tools/call" && - !response.error && - !(response.result as { isError?: unknown } | undefined)?.isError; - if (worked && ANSWERS.has(String(call.name))) session?.answered(); - if (worked && call.name === "send" && call.arguments?.final === true) - session?.finished(); - return response; - } -} - -/** 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. */ -export class GooseSession { - /** The conversation it serves. */ - conversation: string | undefined; - /** Takes the first name goose generates for it; a titled session gets none. */ - named: ((title: string) => void) | undefined; - /** goose's id for the running turn, which a steer must name. */ - private runId: string | undefined; - /** Steers waiting for the running turn's id, which goose sends just after - * the turn starts. */ - private runWaiters: ((runId: string | undefined) => void)[] = []; - /** A `session/prompt` is in flight. */ - private prompting = false; - private readonly steering = new Set(); - private steered = Promise.resolve(); - /** Turns to run, each with the model and level it was asked for, if any. */ - private readonly queue: { - text: string; - model?: string; - effort?: string; - current?: () => boolean; - refresh?: ActivityReceipt["refresh"]; - waiter?: Waiter; - delivered?: () => void; - }[] = []; - /** The model the running turn runs on; empty is its provider's default. */ - private model = ""; - /** The thinking level the running turn runs on; empty is goose's own. */ - private level = ""; - /** Messages sent and not yet answered. */ - private pending = 0; - private waiters: Waiter[] = []; - /** The waiter that posts the burst's error or text: the latest message - * acknowledged, else the latest sent. */ - private owner: Waiter | undefined; - private readonly active = new Map(); - private lastStep = ""; - private lastError: string | undefined; - /** Whether this work answered in Buzz, and what goose wrote meanwhile: a - * provider failure comes back as text, and text is never posted. */ - private acted = false; - private written = ""; - private closed = false; - /** The work the owner stopped, while goose ends it: the messages it had - * taken, and their activity. What goose wrote meanwhile is not posted, and - * messages sent since wait for a turn of their own. `ended` is set once - * the cancel ended the turn, not when the turn finished on its own. */ - private stopped: - | { waiters: Waiter[]; receipts: Set; ended?: true } - | undefined; - private idlers: (() => void)[] = []; - - constructor( - private readonly goose: GooseProcess, - readonly id: string, - /** Serves this session's Buzz tools; closed with it. */ - readonly server?: HostMcpServer, - private readonly choice?: () => Choice, - /** The provider and model it runs on, as chosen here. */ - private chosen: Choice = { provider: "", model: "" }, - ) {} - - get busy() { - return this.pending > 0; - } - get working() { - return !this.quiet(); - } - get open() { - return !this.closed; - } - /** The thinking level it was put on; empty while it runs on goose's own. */ - get effort() { - return this.chosen.effort ?? ""; - } - /** Resolves once nothing runs, waits to run or waits for its answer: - * unlike `busy`, it waits out the model call goose makes after a final - * `send` too. */ - idle(): Promise { - if (this.quiet()) return Promise.resolve(); - return new Promise((resolve) => this.idlers.push(resolve)); - } - - /** Sends one user message, whose turn runs on `model` of the provider and - * thinking level `effort` when given. Resolves when it, and anything sent - * after it, has been answered, or a final `send` ended the work. */ - send( - text: string, - activity: ActivityReceipt = {}, - model?: string, - effort?: string, - ): Promise { - if (this.closed) - return Promise.resolve({ ok: false, error: "The session was closed" }); - this.pending++; - let waiter!: Waiter; - const settled = new Promise((resolve) => { - waiter = resolve; - this.waiters.push(resolve); - }); - // A steer cannot change the model or level, and must not pass a queued - // turn. - if ( - this.prompting && - !this.stopped && - !this.queue.length && - (!model || model === this.model) && - (!effort || effort === this.level) - ) - void this.steer(text, activity, waiter, model, effort); - else { - this.queue.push({ - text, - ...(activity.delivered ? { delivered: activity.delivered } : {}), - ...(activity.current ? { current: activity.current } : {}), - ...(activity.refresh ? { refresh: activity.refresh } : {}), - waiter, - ...(model ? { model } : {}), - ...(effort ? { effort } : {}), - }); - this.acknowledge(activity, waiter); - void this.pump(); - } - return settled; - } - - /** A final `send` succeeded: the work counts as done at once, unless - * another message is still waiting for its answer. goose still ends its turn - * with one more model call, whose text nobody posts. */ - finished() { - if (this.pending === 1 && !this.queue.length && !this.steering.size) - this.settle({ ok: true }); - } - - /** Stops the work: the running turn, as ACP's `session/cancel` does, and - * the messages waiting for a turn after it, which goose cannot stop apart. - * A turn not yet sent is never sent. The work then ends as goose ends it; - * a message sent before it has runs as a new turn after. */ - interrupt() { - if (!this.working || this.closed) return; - this.queue.length = 0; - this.steered = Promise.resolve(); - this.stopped = { - waiters: [...this.waiters], - receipts: new Set(this.active.values()), - }; - this.goose.notify("session/cancel", { sessionId: this.id }); - } - - /** A Buzz tool that answers, such as `send`, succeeded. */ - answered() { - this.acted = true; - } - - /** Called by the process with each `session/update` for this session. */ - update(update: Record) { - const kind = update.sessionUpdate; - if (kind === "agent_message_chunk" && !this.acted) { - const content = update.content as { text?: unknown } | undefined; - if (typeof content?.text === "string") - this.written = (this.written + content.text).slice(0, WRITTEN_LIMIT); - } - if (kind === "session_info_update") { - const title = typeof update.title === "string" ? update.title.trim() : ""; - const named = this.named; - if (title && named) { - this.named = undefined; - named(title); - } + }, + steering: { + runId: (update) => { + if (update.sessionUpdate !== "session_info_update") return null; const goose = (update._meta as { goose?: { activeRunId?: unknown } }) ?.goose; - if (goose && "activeRunId" in goose) { - this.runId = - typeof goose.activeRunId === "string" ? goose.activeRunId : undefined; - if (this.runId) this.ranAs(this.runId); - } - return; - } - const step = stepLabel(update); - if (!step) return; - const label = `${step.tool}\u0000${step.target ?? ""}`; - if (label === this.lastStep) return; - this.lastStep = label; - for (const receipt of this.active.values()) - receipt.step?.(step.tool, step.target); - } - - /** Ends the session's work with `error`; nothing more is sent to it. */ - fail(error: string) { - this.closed = true; - this.queue.length = 0; - this.settle({ ok: false, error }); - } - - private async steer( - text: string, - activity: ActivityReceipt, - waiter: Waiter, - model?: string, - effort?: string, - ) { - this.steering.add(waiter); - const prior = this.steered; - let release!: () => void; - this.steered = new Promise((resolve) => { - release = resolve; - }); - // Work the owner stopped, or that ended meanwhile, takes no more. - const live = () => - this.waiters.includes(waiter) && !this.stopped?.waiters.includes(waiter); - try { - await prior; - if (!live()) return; - const runId = - this.runId ?? - (this.prompting - ? await new Promise((resolve) => - this.runWaiters.push(resolve), - ) - : undefined); - if (activity.refresh) text = steerPrompt(await activity.refresh()); - if (!live()) return; - if (activity.current && !activity.current()) { - this.discard(waiter); - return; - } - if (runId) { - try { - await this.goose.request("_goose/unstable/session/steer", { - sessionId: this.id, - prompt: [{ type: "text", text }], - expectedRunId: runId, - }); - if (live()) { - activity.delivered?.(); - this.acknowledge(activity, waiter); - } - return; - } catch { - // The turn ended, or another took its place: run it as the next. - } - } - if (live()) { - const next = this.queue.findIndex( - (turn) => - turn.waiter && - this.waiters.indexOf(turn.waiter) > this.waiters.indexOf(waiter), - ); - this.queue.splice(next < 0 ? this.queue.length : next, 0, { - text, - ...(activity.delivered ? { delivered: activity.delivered } : {}), - ...(activity.current ? { current: activity.current } : {}), - ...(activity.refresh ? { refresh: activity.refresh } : {}), - waiter, - ...(model ? { model } : {}), - ...(effort ? { effort } : {}), + if (!goose || !("activeRunId" in goose)) return null; + return typeof goose.activeRunId === "string" + ? goose.activeRunId + : undefined; + }, + steer: async (goose, sessionId, runId, text) => { + await goose.request("_goose/unstable/session/steer", { + sessionId, + prompt: [{ type: "text", text }], + expectedRunId: runId, + }); + }, + }, + select: async (goose, session, want) => { + const configure = async (configId: string, value: string) => { + try { + await goose.request("session/set_config_option", { + sessionId: session.id, + configId, + value, }); - this.acknowledge(activity, waiter); - } - } finally { - release(); - this.steering.delete(waiter); - // Runs the next turn, or ends work whose turn ended during the steer. - void this.pump(); - } - } - - private discard(waiter: Waiter) { - this.waiters = this.waiters.filter((pending) => pending !== waiter); - this.pending = this.waiters.length; - if (this.owner === waiter) this.owner = this.waiters.at(-1); - for (const [scope, receipt] of this.active) - if (receipt.current && !receipt.current()) this.active.delete(scope); - waiter({ ok: true, undelivered: true }); - } - - private get handingOff() { - return [...this.steering].some( - (waiter) => - this.waiters.includes(waiter) && - !this.stopped?.waiters.includes(waiter), - ); - } - - private async pump() { - if (this.prompting || this.handingOff) return; - this.prompting = true; - try { - while (this.queue.length && !this.handingOff && !this.closed) { - // Stopped work has ended, and messages sent since run as new work. - if (this.stopped) this.endStopped(); - const { text, model, effort, current, refresh, waiter, delivered } = - this.queue.shift() ?? { - text: "", - }; - try { - await this.select(model, effort); - const prompt = refresh ? await refresh() : text; - // The owner stopped it while its provider switch was going. - if (this.stopped) { - this.cancelEnded(); - continue; - } - if (current && !current()) { - if (waiter) this.discard(waiter); - continue; - } - delivered?.(); - const { stopReason } = (await this.goose.request("session/prompt", { - sessionId: this.id, - prompt: [{ type: "text", text: prompt }], - })) as { stopReason?: string }; - if (stopReason === "cancelled") this.cancelEnded(); - if (stopReason === "refusal") - this.lastError = "goose refused the request"; - } catch (error) { - this.lastError = errorMessage(error); - } + } catch (error) { + throw new Error( + `goose could not switch its ${configId} to ${value}: ${errorMessage(error)}`, + ); } - } finally { - this.prompting = false; - this.runId = undefined; - this.ranAs(undefined); - } - if (!this.steering.size && !this.closed) this.done(); - this.wake(); - } - - /** Puts the session on the provider, model and thinking level chosen now, - * or `model` of that provider and level `effort`. A steer joins the running - * turn, so a change waits for the next. */ - private async select(model?: string, effort?: string) { - const chosen = this.choice?.(); - if (!chosen) return; - const want = { - ...chosen, - ...(model ? { model } : {}), - ...(effort ? { effort } : {}), }; - this.model = want.model; - this.level = want.effort ?? ""; if ( - want.provider !== this.chosen.provider || - (this.chosen.model && !want.model) + want.provider !== session.chosen.provider || + (session.chosen.model && !want.model) ) { // goose keeps the model when the provider stays the same. - if (want.provider && want.provider === this.chosen.provider) - await this.configure("model", await this.defaultModel(want.provider)); - else await this.configure("provider", want.provider || CONFIGURED); - this.chosen = { ...this.chosen, provider: want.provider, model: "" }; + if (want.provider && want.provider === session.chosen.provider) + await configure("model", await defaultModel(goose, want.provider)); + else await configure("provider", want.provider || CONFIGURED); + session.chosen = { + ...session.chosen, + provider: want.provider, + model: "", + }; } - if (want.model !== this.chosen.model) { - await this.configure("model", want.model); - this.chosen = { ...this.chosen, model: want.model }; + if (want.model !== session.chosen.model) { + await configure("model", want.model); + session.chosen = { ...session.chosen, model: want.model }; } // goose keeps the level across models and providers. It has no value // for its own default, so a session put on a level is reopened for it. - if (want.effort && want.effort !== this.chosen.effort) { - await this.configure("thinking_effort", want.effort); - this.chosen = { ...this.chosen, effort: want.effort }; - } - } - private async configure(configId: string, value: string) { - try { - await this.goose.request("session/set_config_option", { - sessionId: this.id, - configId, - value, - }); - } catch (error) { - throw new Error( - `goose could not switch its ${configId} to ${value}: ${errorMessage(error)}`, - ); + if (want.effort && want.effort !== session.chosen.effort) { + await configure("thinking_effort", want.effort); + session.chosen = { ...session.chosen, effort: want.effort }; } - } - private async defaultModel(provider: string) { - const { entries } = (await this.goose.request( - "_goose/unstable/providers/list", - { providerIds: [provider] }, - )) as { entries: { providerId: string; defaultModel?: string }[] }; - const model = entries.find( - (entry) => entry.providerId === provider, - )?.defaultModel; - if (!model) throw new Error(`goose has no default model for ${provider}`); - return model; - } - - private ranAs(runId: string | undefined) { - for (const resolve of this.runWaiters.splice(0)) resolve(runId); - } - private acknowledge(activity: ActivityReceipt, waiter: Waiter) { - if (this.waiters.includes(waiter)) this.owner = waiter; - // Watches and timers have no message activity: they must not hide a mention. - if (!activity.step) return; - const scope = activity.activityScope ?? "conversation"; - this.active.get(scope)?.superseded?.(); - this.active.set(scope, activity); - this.lastStep = ""; - activity.acknowledged?.(); - } - private done() { - // Work that wrote text and never answered in Buzz answered only goose: a - // model that forgot `send`, or goose saying it cannot reach its provider. - const text = this.acted || this.stopped ? "" : this.written.trim(); - const error = this.lastError; - this.settle({ - ...(error ? { ok: false, error } : { ok: true }), - ...(text ? { text } : {}), - ...(!error && this.stopped?.ended ? { stopped: true } : {}), - }); - } - /** The owner's cancel ended the turn, if it was stopping it. */ - private cancelEnded() { - if (this.stopped) this.stopped.ended = true; - } - /** Settles the work the owner stopped, leaving messages sent since. */ - private endStopped() { - const stopped = this.stopped; - if (!stopped) return; - this.stopped = undefined; - const error = this.lastError; - this.lastError = undefined; - this.acted = false; - this.written = ""; - for (const [scope, receipt] of this.active) - if (stopped.receipts.has(receipt)) this.active.delete(scope); - const owner = - this.owner && stopped.waiters.includes(this.owner) - ? this.owner - : stopped.waiters.at(-1); - if (this.owner === owner) this.owner = undefined; - this.waiters = this.waiters.filter( - (waiter) => !stopped.waiters.includes(waiter), - ); - this.pending = this.waiters.length; - for (const resolve of stopped.waiters) - resolve( - resolve === owner && error - ? { ok: false, error } - : { - ok: !error, - ...(!error && stopped.ended ? { stopped: true } : {}), - }, - ); - } - private settle(settled: Settled) { - this.pending = 0; - this.active.clear(); - this.lastError = undefined; - this.acted = false; - this.written = ""; - this.stopped = undefined; - const owner = this.owner ?? this.waiters.at(-1); - this.owner = undefined; - for (const resolve of this.waiters.splice(0)) - resolve( - resolve === owner - ? settled - : { ok: settled.ok, ...(settled.stopped ? { stopped: true } : {}) }, - ); - this.wake(); - } - private quiet() { - return ( - this.closed || - (!this.prompting && - !this.steering.size && - !this.pending && - !this.queue.length) - ); - } - private wake() { - if (this.quiet()) for (const resolve of this.idlers.splice(0)) resolve(); - } -} - -/** A `session/update` as the step it shows: a tool call by its title and - * target, thinking, or writing. */ -export function stepLabel( - update: Record, -): { tool: string; target?: string } | undefined { - switch (update.sessionUpdate) { - case "tool_call": { - const title = typeof update.title === "string" ? update.title : ""; - // goose titles a call `shell · sleep 10`, or `buzz: send`. - const tool = - title.split(" · ")[0]?.split(/__|: /).at(-1)?.trim() || "Tool"; - const input = (update.rawInput ?? {}) as Record; - const target = STEP_KEYS.map((key) => input[key]).find( - (value): value is string => typeof value === "string" && !!value, - ); - return target ? { tool, target: shortTarget(target) } : { tool }; - } - case "agent_thought_chunk": - return { tool: "Thinking" }; - case "agent_message_chunk": - return { tool: "Writing reply" }; - default: - return undefined; - } -} - -/** The option that allows the call, preferring one that keeps allowing. */ -function allow(params: Record) { - const options = Array.isArray(params.options) - ? (params.options as { optionId?: unknown; kind?: unknown }[]) - : []; - const option = - options.find((item) => item.kind === "allow_always") ?? - options.find((item) => item.kind === "allow_once"); - return option && typeof option.optionId === "string" - ? { outcome: "selected", optionId: option.optionId } - : { outcome: "cancelled" }; -} + }, +}; -function errorMessage(error: unknown) { - if (error instanceof Error) return error.message; - if (error && typeof error === "object" && "message" in error) { - const { message, data } = error as { message: unknown; data?: unknown }; - return typeof data === "string" && data - ? `${message}: ${data}` - : String(message); - } - return String(error); -} +/** Starts `goose-acp` and checks it can reach the Buzz tools. */ +export const GooseProcess = { + start: ( + spawn: Spawn, + launch: Readonly<{ + cwd: string; + env?: Readonly>; + serve?: ServeMcp; + }>, + ) => + AcpProcess.start(spawn, GOOSE, { + ...launch, + env: { ...databricks(), ...launch.env }, + }), +}; -function shortTarget(value: string) { - return value.startsWith("/") ? (value.split("/").at(-1) ?? value) : value; +async function defaultModel(goose: AcpProcess, provider: string) { + const { entries } = (await goose.request("_goose/unstable/providers/list", { + providerIds: [provider], + })) as { entries: { providerId: string; defaultModel?: string }[] }; + const model = entries.find( + (entry) => entry.providerId === provider, + )?.defaultModel; + if (!model) throw new Error(`goose has no default model for ${provider}`); + return model; } diff --git a/src/bundled/goose/runtime.test.ts b/src/bundled/goose/runtime.test.ts index 856a893c22..03b22dcd02 100644 --- a/src/bundled/goose/runtime.test.ts +++ b/src/bundled/goose/runtime.test.ts @@ -309,6 +309,28 @@ describe("GooseRuntime", () => { expect(process?.cancelled).toEqual([process?.turns[0]?.sessionId]); }); + it("posts nothing goose wrote without `send`", async () => { + // Text the agent writes outside `send` is narration, not a reply. + for (const reply of ["I've replied to Will.", "Plain answer"]) { + const { deliver, published, runtime } = setup({ reply }); + let activity: Promise = Promise.resolve(); + await deliver(message("1", "@Gander hi"), { + workingUntil: (done: Promise) => { + activity = done; + }, + }); + expect(await activity, reply).toMatchObject({ ok: true, text: reply }); + // The turn's report step ends by notifying, after any post it makes. + await new Promise((resolve) => { + const stop = runtime.subscribe(() => { + stop(); + resolve(); + }); + }); + expect(published, reply).toEqual([]); + } + }); + it("tells a subscriber when a new conversation's first turn starts working", async () => { const { deliver, goose, runtime } = setup({ hold: true }); // As a view does: each notification reads the snapshot again. @@ -337,7 +359,6 @@ describe("GooseRuntime", () => { const { deliver, goose, published } = setup({ scope: "channel", hold: true, - reply: "Forgot to send", }); await deliver(reply("3", one)); await flush(100); @@ -346,17 +367,14 @@ describe("GooseRuntime", () => { const process = goose(); expect(process?.turns).toHaveLength(1); expect(process?.turns[0]?.steered).toHaveLength(1); - process?.turns[0]?.finish(); + process?.exit(1, "crashed\n"); await flush(100); - expect(published).toEqual([ - { - kind: 9, - content: "Forgot to send", - tags: [ - ["h", channel], - ["e", two.id, "", "reply"], - ], - }, + expect(published.map((event) => event.tags)).toEqual([ + [ + ["h", channel], + ["e", two.id, "", "reply"], + ["p", alice], + ], ]); }); @@ -373,24 +391,6 @@ describe("GooseRuntime", () => { expect(memoryOpen).not.toHaveBeenCalled(); }); - it("posts goose's own text in the thread, or at a Solo chat's top level", async () => { - const plain = setup({ reply: "Plain answer" }); - await plain.deliver(message("1", "@Gander hi")); - await flush(100); - expect(plain.published.map((event) => event.tags)).toEqual([ - [ - ["h", channel], - ["e", "1".repeat(64), "", "reply"], - ], - ]); - const solo = setup({ reply: "Plain answer", solo: new Set([channel]) }); - await solo.deliver(message("1", "@Gander hi")); - await flush(100); - expect(solo.published.map((event) => event.tags)).toEqual([ - [["h", channel]], - ]); - }); - it("leaves a Solo chat's session for goose to name, and offers that name", async () => { const solo = setup({ hold: true, solo: new Set([channel]) }); const named = vi.fn(); diff --git a/src/bundled/goose/runtime.ts b/src/bundled/goose/runtime.ts index 9a0c8891a4..53997a0918 100644 --- a/src/bundled/goose/runtime.ts +++ b/src/bundled/goose/runtime.ts @@ -55,12 +55,7 @@ import { } from "../claude-code/prompt"; import { ownerCanvas, readProject } from "../../features/agent-harness/project"; import type { ServeMcp, ToolServer } from "./goose"; -import { - AgentSessions, - localSessions, - ServerPool, - SharedProcess, -} from "./sessions"; +import { AgentSessions, localSessions, SharedProcess } from "./sessions"; export type Config = Readonly<{ /** A model the provider offers; empty uses the provider's default, or the @@ -139,8 +134,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( { @@ -504,13 +497,10 @@ export class GooseRuntime { .then((result) => { if (result.undelivered) agent.restoreNotes?.(channel, notes, noteRoot); - // Of the messages one turn answered, only one has its error or text. + // Of the messages one turn answered, only one has its error. An + // answer goose wrote without `send` stays in local activity. if (result.error !== undefined) return this.report(agent, channel, event, result.error, topLevel); - // An answer goose wrote without `send` is posted for it, so the - // thread is not left waiting. - if (result.text) - return this.post(agent, channel, event, result.text, topLevel); }) .catch((error) => console.warn("goose turn failed", error)) .finally(() => this.notify()); @@ -765,7 +755,6 @@ export class GooseRuntime { resolveDirectory: (path) => this.desktop.resolveDirectory(path), tools, process: this.goose, - pool: this.pool, store: localSessions(this.storage, agent), onChange: () => this.notify(), // Read as each turn starts, so a change needs no restart. diff --git a/src/bundled/goose/sessions.test.ts b/src/bundled/goose/sessions.test.ts index cb0f6020df..87ffc1cb6c 100644 --- a/src/bundled/goose/sessions.test.ts +++ b/src/bundled/goose/sessions.test.ts @@ -188,9 +188,11 @@ describe("AgentSessions", () => { const { fake, sessions } = setup({}, true); const one = sessions.deliver("a", "one", 1); const other = sessions.deliver("b", "other", 2); - await flush(20); + // Both turns are running before the interrupt. + await vi.waitFor(() => expect(fake.geese[0]?.turns).toHaveLength(2)); const [first, second] = fake.geese[0]?.turns ?? []; expect(first?.text).toContain("one"); + expect(second?.text).toContain("other"); await sessions.interrupt("a"); await flush(); expect(fake.geese[0]?.cancelled).toEqual([first?.sessionId]); diff --git a/src/bundled/goose/sessions.ts b/src/bundled/goose/sessions.ts index bf5ad2e9ea..bb8d82cad1 100644 --- a/src/bundled/goose/sessions.ts +++ b/src/bundled/goose/sessions.ts @@ -1,100 +1,32 @@ -// One agent's goose sessions: one ACP session per conversation (a thread, a -// whole DM, or a channel), in the `goose-acp` process all goose agents share. -// A quiet 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, so all goose agents share one pool of them, and the least -// recently used idle session closes to make room. Sessions live as long as -// the app does; only when each conversation last used one is saved, so its -// sessions stay listed after they close. +// One agent's goose sessions, in the `goose-acp` process all goose agents +// share. The sessions themselves live in the shared ACP client. import { - GooseProcess, - type ActivityReceipt, - type Choice, - type GooseHost, - type GooseSession, - type ServeMcp, - type Settled, - type ToolServer, -} from "./goose"; -import type { SessionAction } from "../../features/agents/service"; -import { absoluteWorkspace } from "../codex/config"; - -/** A conversation's session closes after this long without a message. */ -export const IDLE_MS = 15 * 60_000; -/** The most MCP servers goose sessions hold at once, below the host's 64 for - * every plugin, so other plugins keep room. */ -export const POOL_LIMIT = 48; -const STOPPED: Settled = { ok: false, error: "goose was stopped" }; -const LATE: Settled = { - ok: false, - error: "goose did not start in time", - undelivered: true, -}; -/** Saved conversations per agent; the least recently used are forgotten. */ -const SAVED_LIMIT = 200; - -/** What is saved for one conversation: when its session last took in a - * message or ended a turn. */ -export type SavedConversation = Readonly<{ at: number }>; -export type SessionStore = { - set(key: string, saved: SavedConversation): void; -}; - -/** A session's hold on one MCP server in the pool. */ -type Slot = { - /** Whether it can be closed to make room: no turn runs in it. */ - idle(): boolean; - /** Closes the session, and resolves once its server has stopped. */ - evict(): Promise; -}; - -/** The MCP servers every goose agent's sessions hold, least recently used - * first. One runtime shares one. */ -export class ServerPool { - private readonly slots = new Set(); - constructor(private readonly limit = POOL_LIMIT) {} - - get size() { - return this.slots.size; - } - /** Holds a server for `slot`, first closing idle sessions to keep below the - * limit. Busy ones are never closed, so it can go over. Resolves once the - * closed sessions' servers have stopped: the host counts them until then. */ - take(slot: Slot) { - const evicted: Promise[] = []; - while (this.slots.size >= this.limit) { - const stopped = this.evict(); - if (!stopped) break; - evicted.push(stopped); - } - this.slots.add(slot); - return Promise.all(evicted); - } - /** Marks `slot` as used now. */ - touch(slot: Slot) { - if (this.slots.delete(slot)) this.slots.add(slot); - } - release(slot: Slot) { - this.slots.delete(slot); - } - /** Closes the least recently used idle session, and resolves once its - * server has stopped; undefined when every session is busy. */ - evict() { - for (const slot of this.slots) - if (slot.idle()) { - this.slots.delete(slot); - return slot.evict(); - } - return undefined; - } -} + AgentSessions as AcpAgentSessions, + type SessionsOptions as AcpSessionsOptions, +} from "../../features/agent-harness/acp/agent-sessions"; +import type { AcpProcess } from "../../features/agent-harness/acp/process"; +import { GooseProcess, type GooseHost, type ServeMcp } from "./goose"; +import type { + SavedConversation, + SessionStore, +} from "../../features/agent-harness/acp/agent-sessions"; + +export { + abortable, + absolute, + IDLE_MS, + POOL_LIMIT, + ServerPool, + type Launch, + type SavedConversation, + type SessionStore, +} from "../../features/agent-harness/acp/agent-sessions"; /** The one `goose-acp` every goose agent's sessions open in, started in the * home folder when a session first needs it, and again after it exits. One * runtime shares one. */ export class SharedProcess { - private ready: Promise | undefined; + private ready: Promise | undefined; constructor( private readonly host: GooseHost, /** Starts a session's MCP server for its agent's tools. */ @@ -103,7 +35,7 @@ export class SharedProcess { /** The running process. One that failed to start is started again on the * next call. */ - get(): Promise { + get(): Promise { if (this.ready) return this.ready; const ready = (async () => GooseProcess.start(this.host.spawn, { @@ -125,499 +57,18 @@ export class SharedProcess { } } -/** What a session is opened with. */ -export type Launch = Readonly<{ - /** The agent's working directory, absolute or `~/…`. */ - workspace: string; - systemPrompt: string; -}>; - -export type SessionsOptions = Readonly<{ - /** The host's, for the workspace's absolute path. */ - resolveDirectory: GooseHost["resolveDirectory"]; - /** Answers the agent's Buzz tools. */ - tools: ToolServer; - /** The process all goose agents' sessions open in. */ - process: SharedProcess; - /** The servers all goose agents' sessions hold. */ - pool: ServerPool; - /** The agent's current settings, for a new session. */ - launch(): Promise; - /** The agent's provider, model and thinking level now, which each new - * turn runs on. */ - choice(): Choice; - /** Changes when a setting a session was opened with has changed. */ - fingerprint(): string; - /** Where each conversation's last use is saved. */ - store?: SessionStore; - /** Called when a conversation's session opens, is sent a message, or - * closes. */ - onChange?(): void; -}>; - -type Live = { - session: GooseSession; - process: GooseProcess; - fingerprint: string; - slot: Slot; - /** When it last took in a message or ended a turn. */ - at: number; - timer?: ReturnType; -}; - -export class AgentSessions { - /** The latest process its sessions opened in, whose exit drops them. */ - private process: GooseProcess | undefined; - private readonly live = new Map(); - /** One-off turns running or starting, by label. */ - private readonly once = new Map(); - private onceCount = 0; - /** Conversations whose session is opening, and the turn it opened for. */ - private readonly opening = new Map< - string, - Promise<{ settled?: Promise }> - >(); - /** created_at of the newest event each conversation's session was shown. */ - private readonly shown = new Map(); - /** Conversations that have had a session, whose next new one may have been - * prompted for the old one. */ - private readonly had = new Set(); - private disposed = false; - - constructor(private readonly options: SessionsOptions) {} - - /** What this conversation's session has been shown. */ - seen(key: string) { - return this.shown.get(key) ?? 0; - } - active(key: string) { - return this.opening.has(key) || !!this.live.get(key)?.session.working; - } - /** Conversations with a session, whether each is working, and when each - * last took in a message or ended a turn. */ - snapshot(): { - key: string; - busy: boolean; - sessionId?: string; - at?: number; - }[] { - return [ - ...[...this.live].map(([key, live]) => ({ - key, - busy: live.session.busy, - sessionId: live.session.id, - at: live.at, - })), - ...[...this.once.keys()].map((key) => ({ key, busy: true })), - ]; - } - - /** Sends `text` into the conversation `key`, and resolves when the session - * has answered it. `steer` replaces `text` when the session is mid-turn; - * `fresh` builds it again when the session it was built for was closed; - * `title` names a session opened for it, except that the conversation's - * first one is left for goose to name, if it names sessions, when `named` - * takes that name; - * `model`, one of the agent's - * provider's, and `effort`, one of its thinking levels, are what its turn - * runs on. Once `signal` aborts, it stops - * waiting for a session to open, which is then kept for the next message, - * and sends nothing. */ - async deliver( - key: string, - text: string, - seen: number, - { - steer = text, - current, - fresh, - title, - named, - signal, - model, - effort, - ...activity - }: Readonly<{ - steer?: string; - current?: () => boolean; - fresh?: () => Promise; - title?: string; - named?: (title: string) => void; - signal?: AbortSignal; - model?: string; - effort?: string; - }> & - ActivityReceipt = {}, - ): Promise { - for (;;) { - if (current && !current()) return { ok: true, undelivered: true }; - if (this.disposed) return { ...STOPPED, undelivered: true }; - // A message that arrives while its conversation's session opens uses it. - const opening = this.opening.get(key); - if (opening) { - // The session stays shared; only this message gives up on it. - try { - await abortable( - opening.catch(() => undefined), - signal, - ); - } catch { - return LATE; - } - continue; - } - const live = this.live.get(key); - if (!live) break; - if ( - !live.session.open || - (!live.session.busy && (this.stale(live) || this.leveled(live, effort))) - ) { - this.drop(key, live); - break; - } - return this.answer( - key, - live, - seen, - live.session.busy ? steer : text, - { ...activity, ...(current ? { current } : {}) }, - model, - effort, - ); - } - // Reserved before anything is awaited, so a message that comes meanwhile - // waits for this session instead of opening another. - const starting = (async () => { - const first = !this.had.has(key); - // Perhaps built for a session closed since, which had been shown the rest. - if (!first && fresh) text = await fresh(); - // Naming costs a model call: goose names only the conversation's start. - const naming = first ? named : undefined; - const live = await this.open( - key, - title, - (live) => this.disposed || this.adopt(key, live), - naming, - ); - if (this.disposed) { - void this.close(live); - throw new Error("goose was stopped"); - } - if (signal?.aborted) return {}; - if (current && !current()) - return { - settled: Promise.resolve({ ok: true, undelivered: true } as Settled), - }; - const refresh = activity.refresh; - return { - settled: this.answer( - key, - live, - seen, - text, - { - ...activity, - ...(current ? { current } : {}), - ...(refresh - ? { - refresh: () => refresh(!first), - } - : {}), - }, - model, - effort, - ), - }; - })(); - this.opening.set(key, starting); - const opened = () => { - if (this.opening.get(key) === starting) this.opening.delete(key); - }; - starting.then(opened, opened); - // Every failure here comes before a turn takes the message in. - let started: Awaited; - try { - started = await abortable(starting, signal); - } catch (error) { - if (this.disposed) return { ...STOPPED, undelivered: true }; - if (signal?.aborted) return LATE; - console.warn("goose could not start", error); - return { - ok: false, - error: `goose could not start: ${message(error)}`, - undelivered: true, - }; - } - return started.settled ?? LATE; - } - - /** Stops the turn running in conversation `key`, if any, once a session - * opening for it has taken its message. */ - async interrupt(key: string) { - await this.opening.get(key)?.catch(() => undefined); - this.live.get(key)?.session.interrupt(); - } - /** Includes the model call after a final send has answered the turn. */ - async idle(key: string) { - await this.opening.get(key)?.catch(() => undefined); - await this.live.get(key)?.session.idle(); - } - /** Waits for the conversation's session to finish its work, then compacts - * it or closes it, so the message opens a new one. Without an open session - * there is nothing to do. */ - async reset(key: string, action: SessionAction): Promise { - const live = this.live.get(key); - if (!live?.session.open) return { ok: true }; - await live.session.idle(); - if (this.disposed) return STOPPED; - if (this.live.get(key) !== live || !live.session.open) return { ok: true }; - if (action === "new") { - this.drop(key, live); - return { ok: true }; - } - const result = await this.answer( - key, - live, - this.seen(key), - "/compact", - {}, - undefined, - ); - // goose ends the turn either way, and says which in its text. - if (!result.ok || result.text === "Compaction complete") return result; - return { - ok: false, - error: result.text || "goose could not compact the session", - }; - } - - /** Runs one turn in a new session, then closes it. `label` names it in the - * snapshot. It gives up, and stops the turn, once `signal` aborts. - * `prompt` is asked for the turn's text once the session is open, so a turn - * that should no longer run by then returns undefined and sends nothing. */ - async runOnce( - label: string, - prompt: () => string | undefined | Promise, - { - signal, - title, - current, - }: Readonly<{ - signal?: AbortSignal; - current?: () => boolean; - title?: string; - }> = {}, - ): Promise { - if (this.disposed) return STOPPED; - const key = `${label} #${++this.onceCount}`; - this.once.set(key, undefined); - this.options.onChange?.(); - const opening = this.open( - key, - title, - (live) => this.once.has(key) && this.once.set(key, live), - ); - let live: Live | undefined; - try { - live = await abortable(opening, signal); - if (this.disposed) return STOPPED; - const text = await abortable(Promise.resolve(prompt()), signal); - if (text === undefined || this.disposed) return { ok: true }; - return await abortable( - live.session.send(text, { ...(current ? { current } : {}) }), - signal, - ); - } catch (error) { - if (signal?.aborted) return { ok: false, error: message(signal.reason) }; - console.warn("goose could not start a one-off session", error); - return { ok: false, error: `goose could not start: ${message(error)}` }; - } finally { - this.once.delete(key); - // One that opens after the turn was given up is closed once it has. - if (live) await this.close(live); - else void opening.then((late) => this.close(late)).catch(() => undefined); - this.options.onChange?.(); - } - } - - /** Closes idle sessions opened with old settings; busy ones close when they - * next fall idle. */ - reconfigure() { - for (const [key, live] of this.live) - if (!live.session.busy && this.stale(live)) this.drop(key, live); - } - - /** Closes its sessions, ending their turns; the process keeps serving other - * agents. */ - dispose() { - this.disposed = true; - for (const [key, live] of this.live) this.drop(key, live); - for (const live of this.once.values()) if (live) void this.close(live); - } - - /** Opened with old settings, which only a new session runs on. */ - private stale(live: Live) { - return live.fingerprint !== this.options.fingerprint(); - } - /** On a thinking level, when a turn with neither a pick nor a saved level - * wants goose's own. goose names no value for its own, so only a new - * session runs on it; checked as that turn comes, so a session put on a - * picked level keeps its conversation until then. */ - private leveled(live: Live, effort?: string) { - return Boolean( - live.session.effort && !effort && !this.options.choice().effort, - ); - } - - /** Opens a session for `key` in the agent's workspace, which `use` records - * as its user's. */ - private async open( - key: string, - title: string | undefined, - use: (live: Live) => unknown, - named?: (title: string) => void, - ): Promise { - const { pool, resolveDirectory } = this.options; - // Read before `launch` reads the settings, so it never describes newer ones. - const fingerprint = this.options.fingerprint(); - const launch = await this.options.launch(); - // Removed meanwhile: a goose started now would outlive it. - if (this.disposed) throw new Error("goose was stopped"); - const [process, cwd] = await Promise.all([ - this.options.process.get(), - absolute(resolveDirectory, launch.workspace), - ]); - if (process !== this.process) { - this.process = process; - void process.exited.then(() => { - for (const [key, live] of this.live) - if (live.process === process) this.drop(key, live); - }); - } - // Busy until it is adopted as a conversation's. - const slot: Slot = { idle: () => false, evict: async () => undefined }; - const room = pool.take(slot); - const start = () => - process.open({ - cwd, - systemPrompt: launch.systemPrompt, - ...(title ? { title } : {}), - ...(named ? { named } : {}), - tools: this.options.tools, - choice: this.options.choice, - }); - let session: GooseSession; - try { - await room; - session = await start().catch(async (error: unknown) => { - // Other plugins hold servers too: make room once, and try again. - const room = /too many mcp servers/i.test(message(error)) - ? pool.evict() - : undefined; - if (!room) throw error; - await room; - return start(); - }); - } catch (error) { - pool.release(slot); - throw error; - } - const released = () => pool.release(slot); - void (session.server?.closed ?? Promise.resolve()).then(released, released); - session.conversation = key; - const live = { session, process, fingerprint, slot, at: Date.now() }; - if (fingerprint !== this.options.fingerprint() && !this.disposed) { - // Settings saved while it opened: it would work with the old ones. - void this.close(live); - return this.open(key, title, use, named); - } - use(live); - return live; - } +export type SessionsOptions = Omit; - private adopt(key: string, live: Live) { - this.had.add(key); - this.live.set(key, live); - // Not idle while the message it opened for is still on its way to it. - live.slot.idle = () => - !live.session.busy && - this.live.get(key) === live && - !this.opening.has(key); - live.slot.evict = async () => { - this.drop(key, live); - await live.session.server?.closed; - }; - // Not saved until a message is handed to it: its own may have given up. - this.countdown(key, live); - this.options.onChange?.(); - } - private async answer( - key: string, - live: Live, - seen: number, - text: string, - activity: ActivityReceipt, - model: string | undefined, - effort?: string, - ) { - const sent = live.session.send( - text, - { - ...activity, - delivered: () => { - if (this.live.get(key) === live) this.touch(key, live, seen); - activity.delivered?.(); - }, - }, - model, - effort, - ); - // A final Buzz send can answer before goose finishes its model work. - void live.session.idle().then(() => { - if (this.live.get(key) === live) this.options.onChange?.(); - }); - // Sending made it busy; a snapshot read when it opened is not. - this.options.onChange?.(); - const settled = await sent; - if (this.live.get(key) === live) { - this.touch(key, live); - if (!live.session.busy && this.stale(live)) this.drop(key, live); - } - return settled; - } - /** Notes that the session took in a message or ended a turn, and what it - * has been shown, and restarts its idle countdown. */ - private touch(key: string, live: Live, seen?: number) { - live.at = Date.now(); - this.options.store?.set(key, { at: live.at }); - this.countdown(key, live); - if (seen !== undefined) - this.shown.set(key, Math.max(seen, this.shown.get(key) ?? 0)); - } - /** Closes the session after IDLE_MS, or waits again while it is busy: a - * long turn is not new use. */ - private countdown(key: string, live: Live) { - this.options.pool.touch(live.slot); - clearTimeout(live.timer); - live.timer = setTimeout(() => { - if (live.session.busy) this.countdown(key, live); - else if (this.live.get(key) === live) this.drop(key, live); - }, IDLE_MS); - } - private drop(key: string, live: Live) { - clearTimeout(live.timer); - if (this.live.get(key) === live) { - this.live.delete(key); - this.shown.delete(key); - this.options.onChange?.(); - } - void this.close(live); - } - private close(live: Live) { - return live.process.close(live.session); +/** An agent's sessions, in the process all goose agents share. */ +export class AgentSessions extends AcpAgentSessions { + constructor(options: SessionsOptions) { + super({ ...options, label: "goose" }); } } +/** Saved conversations per agent; the least recently used are forgotten. */ +const SAVED_LIMIT = 200; + /** Conversations saved in local storage, by agent and then conversation. */ export function localSessions( storage: Storage, @@ -653,32 +104,3 @@ export function localSessions( }, }; } - -/** `workspace` as an absolute path, which ACP sessions need. One already - * absolute stays as written, as the sessions started in it recorded it. */ -export async function absolute( - resolveDirectory: GooseHost["resolveDirectory"], - workspace: string, -) { - return absoluteWorkspace(workspace) ? workspace : resolveDirectory(workspace); -} - -/** `work`, or a rejection with `signal`'s reason once it aborts. */ -export function abortable( - work: Promise, - signal?: AbortSignal, -): Promise { - if (!signal) return work; - return new Promise((resolve, reject) => { - const abort = () => reject(signal.reason); - if (signal.aborted) abort(); - signal.addEventListener("abort", abort, { once: true }); - work - .then(resolve, reject) - .finally(() => signal.removeEventListener("abort", abort)); - }); -} - -function message(error: unknown) { - return error instanceof Error ? error.message : String(error); -} diff --git a/src/features/agent-harness/acp/agent-sessions.ts b/src/features/agent-harness/acp/agent-sessions.ts new file mode 100644 index 0000000000..ac221cf31c --- /dev/null +++ b/src/features/agent-harness/acp/agent-sessions.ts @@ -0,0 +1,633 @@ +// One agent's sessions in an ACP harness: one ACP session per conversation (a +// thread, a whole DM, or a channel), in the process its source gives. A quiet +// 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 one pool every ACP plugin's agents share, and the +// least recently used idle session closes to make room, whichever plugin's. +// Sessions live as long as the app does; only when each conversation last used +// one is saved, so its sessions stay listed after they close. +import { absoluteWorkspace } from "../../../bundled/codex/config"; +import type { SessionAction } from "../../agents/service"; +import type { AcpProcess } from "./process"; +import type { AcpSession } from "./session"; +import { codeOf } from "./steps"; +import type { + ActivityReceipt, + Choice, + ResolveDirectory, + Settled, + ToolServer, +} from "./types"; + +/** A conversation's session closes after this long without a message. */ +export const IDLE_MS = 15 * 60_000; +/** 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; + +/** What is saved for one conversation: when its session last took in a + * message or ended a turn. */ +export type SavedConversation = Readonly<{ at: number }>; +export type SessionStore = { + set(key: string, saved: SavedConversation): void; +}; + +/** A session's hold on one MCP server in the pool. */ +export type Slot = { + /** Whether it can be closed to make room: no turn runs in it. */ + idle(): boolean; + /** Closes the session, and resolves once its server has stopped. */ + evict(): Promise; +}; + +/** The MCP servers agents' sessions hold, least recently used first. */ +export class ServerPool { + private readonly slots = new Set(); + constructor(private readonly limit = POOL_LIMIT) {} + + get size() { + return this.slots.size; + } + /** Holds a server for `slot`, first closing idle sessions to keep below the + * limit. Busy ones are never closed, so it can go over. Resolves once the + * closed sessions' servers have stopped: the host counts them until then. */ + take(slot: Slot) { + const evicted: Promise[] = []; + while (this.slots.size >= this.limit) { + const stopped = this.evict(); + if (!stopped) break; + evicted.push(stopped); + } + this.slots.add(slot); + return Promise.all(evicted); + } + /** Marks `slot` as used now. */ + touch(slot: Slot) { + if (this.slots.delete(slot)) this.slots.add(slot); + } + release(slot: Slot) { + this.slots.delete(slot); + } + /** Closes the least recently used idle session, and resolves once its + * server has stopped; undefined when every session is busy. */ + evict() { + for (const slot of this.slots) + if (slot.idle()) { + this.slots.delete(slot); + return slot.evict(); + } + return undefined; + } +} + +/** 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 `~/…`. */ + workspace: string; + systemPrompt: string; +}>; + +export type SessionsOptions = Readonly<{ + /** Names the harness in errors. */ + label: string; + /** The host's, for the workspace's absolute path. */ + resolveDirectory: ResolveDirectory; + /** Answers the agent's Buzz tools. */ + tools: ToolServer; + /** The process its sessions open in, started when needed. */ + process: Readonly<{ get(): Promise }>; + /** 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, model and thinking level now, which each new + * turn runs on. */ + choice(): Choice; + /** Changes when a setting a session was opened with has changed. */ + fingerprint(): string; + /** Where each conversation's last use is saved. */ + store?: SessionStore; + /** Called when a conversation's session opens, is sent a message, or + * closes. */ + onChange?(): void; +}>; + +type Live = { + session: AcpSession; + process: AcpProcess; + fingerprint: string; + slot: Slot; + /** When it last took in a message or ended a turn. */ + at: number; + timer?: ReturnType; +}; + +export class AgentSessions { + /** The latest process its sessions opened in, whose exit drops them. */ + private process: AcpProcess | undefined; + private readonly live = new Map(); + /** One-off turns running or starting, by label. */ + private readonly once = new Map(); + private onceCount = 0; + /** Conversations whose session is opening, and the turn it opened for. */ + private readonly opening = new Map< + string, + Promise<{ settled?: Promise }> + >(); + /** created_at of the newest event each conversation's session was shown. */ + private readonly shown = new Map(); + /** Conversations that have had a session, whose next new one may have been + * prompted for the old one. */ + private readonly had = new Set(); + private disposed = false; + + private readonly stopped: Settled; + private readonly late: Settled; + + constructor(private readonly options: SessionsOptions) { + this.stopped = { ok: false, error: `${options.label} was stopped` }; + this.late = { + ok: false, + error: `${options.label} did not start in time`, + undelivered: true, + }; + } + + /** What this conversation's session has been shown. */ + seen(key: string) { + return this.shown.get(key) ?? 0; + } + active(key: string) { + return this.opening.has(key) || !!this.live.get(key)?.session.working; + } + /** Conversations with a session, whether each is working, and when each + * last took in a message or ended a turn. */ + snapshot(): { + key: string; + busy: boolean; + sessionId?: string; + at?: number; + }[] { + return [ + ...[...this.live].map(([key, live]) => ({ + key, + busy: live.session.busy, + sessionId: live.session.id, + at: live.at, + })), + ...[...this.once.keys()].map((key) => ({ key, busy: true })), + ]; + } + + /** Sends `text` into the conversation `key`, and resolves when the session + * has answered it. `steer` replaces `text` when the session is mid-turn; + * `fresh` builds it again when the session it was built for was closed; + * `title` names a session opened for it, except that the conversation's + * first one is left for the harness to name, if it names sessions, when `named` + * takes that name; + * `model`, one of the agent's + * provider's, and `effort`, one of its thinking levels, are what its turn + * runs on. Once `signal` aborts, it stops + * waiting for a session to open, which is then kept for the next message, + * and sends nothing. */ + async deliver( + key: string, + text: string, + seen: number, + { + steer = text, + current, + fresh, + title, + named, + signal, + model, + effort, + ...activity + }: Readonly<{ + steer?: string; + current?: () => boolean; + fresh?: () => Promise; + title?: string; + named?: (title: string) => void; + signal?: AbortSignal; + model?: string; + effort?: string; + }> & + ActivityReceipt = {}, + ): Promise { + for (;;) { + if (current && !current()) return { ok: true, undelivered: true }; + if (this.disposed) return { ...this.stopped, undelivered: true }; + // A message that arrives while its conversation's session opens uses it. + const opening = this.opening.get(key); + if (opening) { + // The session stays shared; only this message gives up on it. + try { + await abortable( + opening.catch(() => undefined), + signal, + ); + } catch { + return this.late; + } + continue; + } + const live = this.live.get(key); + if (!live) break; + if ( + !live.session.open || + (!live.session.busy && (this.stale(live) || this.leveled(live, effort))) + ) { + this.drop(key, live); + break; + } + return this.answer( + key, + live, + seen, + live.session.busy ? steer : text, + { ...activity, ...(current ? { current } : {}) }, + model, + effort, + ); + } + // Reserved before anything is awaited, so a message that comes meanwhile + // waits for this session instead of opening another. + const starting = (async () => { + const first = !this.had.has(key); + // Perhaps built for a session closed since, which had been shown the rest. + if (!first && fresh) text = await fresh(); + // Naming costs a model call: the harness names only the conversation's start. + const naming = first ? named : undefined; + const live = await this.open( + key, + title, + (live) => this.disposed || this.adopt(key, live), + naming, + ); + if (this.disposed) { + void this.close(live); + throw new Error(`${this.options.label} was stopped`); + } + if (signal?.aborted) return {}; + if (current && !current()) + return { + settled: Promise.resolve({ ok: true, undelivered: true } as Settled), + }; + const refresh = activity.refresh; + return { + settled: this.answer( + key, + live, + seen, + text, + { + ...activity, + ...(current ? { current } : {}), + ...(refresh + ? { + refresh: () => refresh(!first), + } + : {}), + }, + model, + effort, + ), + }; + })(); + this.opening.set(key, starting); + const opened = () => { + if (this.opening.get(key) === starting) this.opening.delete(key); + }; + starting.then(opened, opened); + // Every failure here comes before a turn takes the message in. + let started: Awaited; + try { + started = await abortable(starting, signal); + } catch (error) { + if (this.disposed) return { ...this.stopped, undelivered: true }; + if (signal?.aborted) return this.late; + console.warn(`${this.options.label} could not start`, error); + const code = codeOf(error); + return { + ok: false, + error: `${this.options.label} could not start: ${message(error)}`, + ...(code === undefined ? {} : { code }), + undelivered: true, + }; + } + return started.settled ?? this.late; + } + + /** Stops the turn running in conversation `key`, if any, once a session + * opening for it has taken its message. */ + async interrupt(key: string) { + await this.opening.get(key)?.catch(() => undefined); + this.live.get(key)?.session.interrupt(); + } + /** Includes the model call after a final send has answered the turn. */ + async idle(key: string) { + await this.opening.get(key)?.catch(() => undefined); + await this.live.get(key)?.session.idle(); + } + /** Waits for the conversation's session to finish its work, then compacts + * it or closes it, so the message opens a new one. Without an open session + * there is nothing to do. */ + async reset(key: string, action: SessionAction): Promise { + const live = this.live.get(key); + if (!live?.session.open) return { ok: true }; + await live.session.idle(); + if (this.disposed) return this.stopped; + if (this.live.get(key) !== live || !live.session.open) return { ok: true }; + if (action === "new") { + this.drop(key, live); + return { ok: true }; + } + const result = await this.answer( + key, + live, + this.seen(key), + "/compact", + {}, + undefined, + ); + // goose ends the turn either way, and says which in its text. + if (!result.ok || result.text === "Compaction complete") return result; + return { + ok: false, + error: + result.text || `${this.options.label} could not compact the session`, + }; + } + + /** Runs one turn in a new session, then closes it. `label` names it in the + * snapshot. It gives up, and stops the turn, once `signal` aborts. + * `prompt` is asked for the turn's text once the session is open, so a turn + * that should no longer run by then returns undefined and sends nothing. */ + async runOnce( + label: string, + prompt: () => string | undefined | Promise, + { + signal, + title, + current, + }: Readonly<{ + signal?: AbortSignal; + current?: () => boolean; + title?: string; + }> = {}, + ): Promise { + if (this.disposed) return this.stopped; + const key = `${label} #${++this.onceCount}`; + this.once.set(key, undefined); + this.options.onChange?.(); + const opening = this.open( + key, + title, + (live) => this.once.has(key) && this.once.set(key, live), + ); + let live: Live | undefined; + try { + live = await abortable(opening, signal); + if (this.disposed) return this.stopped; + const text = await abortable(Promise.resolve(prompt()), signal); + if (text === undefined || this.disposed) return { ok: true }; + return await abortable( + live.session.send(text, { ...(current ? { current } : {}) }), + signal, + ); + } catch (error) { + if (signal?.aborted) return { ok: false, error: message(signal.reason) }; + console.warn( + `${this.options.label} could not start a one-off session`, + error, + ); + return { + ok: false, + error: `${this.options.label} could not start: ${message(error)}`, + }; + } finally { + this.once.delete(key); + // One that opens after the turn was given up is closed once it has. + if (live) await this.close(live); + else void opening.then((late) => this.close(late)).catch(() => undefined); + this.options.onChange?.(); + } + } + + /** Closes idle sessions opened with old settings; busy ones close when they + * next fall idle. */ + reconfigure() { + for (const [key, live] of this.live) + if (!live.session.busy && this.stale(live)) this.drop(key, live); + } + + /** Closes its sessions, ending their turns; the process keeps serving other + * agents. */ + dispose() { + this.disposed = true; + for (const [key, live] of this.live) this.drop(key, live); + for (const live of this.once.values()) if (live) void this.close(live); + } + + private get pool() { + return this.options.pool ?? SHARED_POOL; + } + + /** Opened with old settings, which only a new session runs on. */ + private stale(live: Live) { + return live.fingerprint !== this.options.fingerprint(); + } + /** On a thinking level, when a turn with neither a pick nor a saved level + * wants the harness's own. goose names no value for its own, so only a new + * session runs on it; checked as that turn comes, so a session put on a + * picked level keeps its conversation until then. */ + private leveled(live: Live, effort?: string) { + return Boolean( + live.session.effort && !effort && !this.options.choice().effort, + ); + } + + /** Opens a session for `key` in the agent's workspace, which `use` records + * as its user's. */ + private async open( + key: string, + title: string | undefined, + use: (live: Live) => unknown, + named?: (title: string) => void, + ): Promise { + const { resolveDirectory } = 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(); + // Removed meanwhile: a process started now would outlive it. + if (this.disposed) throw new Error(`${this.options.label} was stopped`); + const [process, cwd] = await Promise.all([ + this.options.process.get(), + absolute(resolveDirectory, launch.workspace), + ]); + if (process !== this.process) { + this.process = process; + void process.exited.then(() => { + for (const [key, live] of this.live) + if (live.process === process) this.drop(key, live); + }); + } + // Busy until it is adopted as a conversation's. + const slot: Slot = { idle: () => false, evict: async () => undefined }; + const room = pool.take(slot); + const start = () => + process.open({ + cwd, + systemPrompt: launch.systemPrompt, + ...(title ? { title } : {}), + ...(named ? { named } : {}), + tools: this.options.tools, + choice: this.options.choice, + }); + let session: AcpSession; + try { + await room; + session = await start().catch(async (error: unknown) => { + // Other plugins hold servers too: make room once, and try again. + const room = /too many mcp servers/i.test(message(error)) + ? pool.evict() + : undefined; + if (!room) throw error; + await room; + return start(); + }); + } catch (error) { + pool.release(slot); + throw error; + } + const released = () => pool.release(slot); + void (session.server?.closed ?? Promise.resolve()).then(released, released); + session.conversation = key; + const live = { session, process, fingerprint, slot, at: Date.now() }; + if (fingerprint !== this.options.fingerprint() && !this.disposed) { + // Settings saved while it opened: it would work with the old ones. + void this.close(live); + return this.open(key, title, use, named); + } + use(live); + return live; + } + + private adopt(key: string, live: Live) { + this.had.add(key); + this.live.set(key, live); + // Not idle while the message it opened for is still on its way to it. + live.slot.idle = () => + !live.session.busy && + this.live.get(key) === live && + !this.opening.has(key); + live.slot.evict = async () => { + this.drop(key, live); + await live.session.server?.closed; + }; + // Not saved until a message is handed to it: its own may have given up. + this.countdown(key, live); + this.options.onChange?.(); + } + private async answer( + key: string, + live: Live, + seen: number, + text: string, + activity: ActivityReceipt, + model: string | undefined, + effort?: string, + ) { + const sent = live.session.send( + text, + { + ...activity, + delivered: () => { + if (this.live.get(key) === live) this.touch(key, live, seen); + activity.delivered?.(); + }, + }, + model, + effort, + ); + // A final Buzz send can answer before the harness finishes its model work. + void live.session.idle().then(() => { + if (this.live.get(key) === live) this.options.onChange?.(); + }); + // Sending made it busy; a snapshot read when it opened is not. + this.options.onChange?.(); + const settled = await sent; + if (this.live.get(key) === live) { + this.touch(key, live); + if (!live.session.busy && this.stale(live)) this.drop(key, live); + } + return settled; + } + /** Notes that the session took in a message or ended a turn, and what it + * has been shown, and restarts its idle countdown. */ + private touch(key: string, live: Live, seen?: number) { + live.at = Date.now(); + this.options.store?.set(key, { at: live.at }); + this.countdown(key, live); + if (seen !== undefined) + this.shown.set(key, Math.max(seen, this.shown.get(key) ?? 0)); + } + /** Closes the session after IDLE_MS, or waits again while it is busy: a + * long turn is not new use. */ + private countdown(key: string, live: Live) { + this.pool.touch(live.slot); + clearTimeout(live.timer); + live.timer = setTimeout(() => { + if (live.session.busy) this.countdown(key, live); + else if (this.live.get(key) === live) this.drop(key, live); + }, IDLE_MS); + } + private drop(key: string, live: Live) { + clearTimeout(live.timer); + if (this.live.get(key) === live) { + this.live.delete(key); + this.shown.delete(key); + this.options.onChange?.(); + } + void this.close(live); + } + private close(live: Live) { + return live.process.close(live.session); + } +} + +/** `workspace` as an absolute path, which ACP sessions need. One already + * absolute stays as written, as the sessions started in it recorded it. */ +export async function absolute( + resolveDirectory: ResolveDirectory, + workspace: string, +) { + return absoluteWorkspace(workspace) ? workspace : resolveDirectory(workspace); +} + +/** `work`, or a rejection with `signal`'s reason once it aborts. */ +export function abortable( + work: Promise, + signal?: AbortSignal, +): Promise { + if (!signal) return work; + return new Promise((resolve, reject) => { + const abort = () => reject(signal.reason); + if (signal.aborted) abort(); + signal.addEventListener("abort", abort, { once: true }); + work + .then(resolve, reject) + .finally(() => signal.removeEventListener("abort", abort)); + }); +} + +function message(error: unknown) { + return error instanceof Error ? error.message : String(error); +} 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..86e8beeb7c --- /dev/null +++ b/src/features/agent-harness/acp/client.test.ts @@ -0,0 +1,618 @@ +// 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 { GOOSE } from "../../../bundled/goose/goose"; +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; + /** Kill asks it to stop, and it never exits. */ + holdExit?: boolean; + /** Rejects this many prompts first, as a harness signed out does. */ + rejectPrompts?: number; + /** Every prompt is answered with this error. */ + promptError?: { code: number; message: string }; + /** Why a turn that ran ends. */ + stopReason?: string; + }> = {}, +) { + const started: { + options: HostProcessOptions; + killed: boolean; + kills: number; + }[] = []; + const received: Message[] = []; + const turns: { + sessionId: string; + text: string; + steered: string[]; + finish(): void; + }[] = []; + let servers = 0; + let rejectPrompts = options.rejectPrompts ?? 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, kills: 0 }; + started.push(record); + const send = (message: Record) => + queueMicrotask(() => { + if (alive) + process.onStdout?.( + `${JSON.stringify({ jsonrpc: "2.0", ...message })}\n`, + ); + }); + const reply = (id: Message["id"], result: unknown) => send({ id, result }); + /** Each session's running turn. */ + const running = new Map(); + const run = (sessionId: string, runId: string | null) => + send({ + method: "session/update", + params: { sessionId, update: { sessionUpdate: "run", runId } }, + }); + 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" && rejectPrompts > 0) { + rejectPrompts--; + send({ id, error: { code: -32000, message: "Sign in first" } }); + } else if (method === "session/prompt" && options.promptError) + send({ id, error: options.promptError }); + else if (method === "session/prompt") { + const sessionId = String(params.sessionId); + const text = (params.prompt as { text: string }[])[0]?.text ?? ""; + const runId = `run-${turns.length + 1}`; + const turn = { + sessionId, + text, + steered: [] as string[], + finish: () => { + running.delete(sessionId); + run(sessionId, null); + reply(id, { stopReason: options.stopReason ?? "end_turn" }); + }, + }; + turns.push(turn); + running.set(sessionId, turn); + run(sessionId, runId); + if (!options.hold) turn.finish(); + } else if (method === "steer") { + const turn = running.get(String(params.sessionId)); + if (turn && `run-${turns.indexOf(turn) + 1}` === params.runId) { + turn.steered.push(String(params.text)); + reply(id, {}); + } else + send({ id, error: { code: -32602, message: "No active run" } }); + } + } + }, + end: async () => undefined, + kill: async () => { + record.killed = true; + record.kills++; + if (!options.holdExit) 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); + } + }); + + const diagnosing: AcpHooks = { ...HOOKS, diagnoseStartup: 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, {}, diagnosing)).rejects.toThrow( + "Install the harness CLI.", + ); + } + }); + + it("reports another early exit as itself", async () => { + const harness = fakeHarness({ exitCode: 1 }); + await expect(start(harness, {}, diagnosing)).rejects.toThrow( + "env: harness: No such file or directory", + ); + }); + + it("reports goose's early exit by its code alone", async () => { + for (const exitCode of [1, 127]) { + const harness = fakeHarness({ exitCode }); + await expect(start(harness, {}, GOOSE)).rejects.toThrow( + new Error(`goose exited (${exitCode})`), + ); + } + }); + + 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("rejects at once when cancelled while it starts, even if it never exits", async () => { + const harness = fakeHarness({ silent: ["initialize"], holdExit: true }); + 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("leaves no request deadline behind when cancelled while it starts", async () => { + vi.useFakeTimers(); + const harness = fakeHarness({ silent: ["initialize"], holdExit: true }); + const controller = new AbortController(); + const starting = start(harness, { signal: controller.signal }); + await vi.advanceTimersByTimeAsync(0); + controller.abort(new Error("Check was cancelled")); + await expect(starting).rejects.toThrow("Check was cancelled"); + // Only the background stop's own wait is left. + await vi.advanceTimersByTimeAsync(5_000); + expect(vi.getTimerCount()).toBe(0); + await vi.advanceTimersByTimeAsync(REQUEST_TIMEOUT_MS); + expect(harness.started[0]?.kills).toBe(1); + }); + + it("rejects at once when cancelled while the host is still starting it, and stops the late harness", async () => { + const harness = fakeHarness(); + let release!: () => void; + const gate = new Promise((resolve) => (release = resolve)); + const spawn: typeof harness.spawn = async (...args) => { + await gate; + return harness.spawn(...args); + }; + const controller = new AbortController(); + const starting = AcpProcess.start(spawn, HOOKS, { + cwd: "/work", + serve: harness.serve, + signal: controller.signal, + }); + const outcome = starting.then( + () => "started", + (error: Error) => error.message, + ); + await flush(); + controller.abort(new Error("Check was cancelled")); + await flush(); + expect(await Promise.race([outcome, Promise.resolve("pending")])).toBe( + "Check was cancelled", + ); + expect(harness.started).toEqual([]); + release(); + await flush(); + 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); + }); + + it("gives up waiting at the time limit for a harness that never exits", async () => { + vi.useFakeTimers(); + const harness = fakeHarness({ holdExit: true }); + const acp = await start(harness); + let stopped = false; + const stopping = acp.stop(1_000).then(() => { + stopped = true; + }); + await vi.advanceTimersByTimeAsync(999); + expect(stopped).toBe(false); + await vi.advanceTimersByTimeAsync(1); + await stopping; + expect(harness.started[0]?.killed).toBe(true); + expect(harness.sent("session/new")).toEqual([]); + }); +}); + +describe("a session with a steering hook", () => { + it("steers a message sent while the next turn picks its model into that turn, not the finished one", async () => { + let selecting: (() => void) | undefined; + const hooks: AcpHooks = { + ...HOOKS, + steering: { + runId: (update) => + update.sessionUpdate !== "run" + ? null + : typeof update.runId === "string" + ? update.runId + : undefined, + steer: async (acp, sessionId, runId, text) => { + await acp.request("steer", { sessionId, runId, text }); + }, + }, + // The second turn's model pick waits until released. + select: (_acp, session, want) => + want.model === "slow" + ? new Promise((resolve) => { + selecting = () => { + session.chosen = want; + resolve(); + }; + }) + : Promise.resolve(), + }; + const harness = fakeHarness({ hold: true }); + const acp = await start(harness, {}, hooks); + const session = await acp.open({ + cwd: "/work", + systemPrompt: "", + choice: () => ({ provider: "", model: "" }), + }); + const first = session.send("first"); + await flush(); + // Another model: queued as the next turn rather than steered. + const second = session.send("second", {}, "slow"); + await flush(); + harness.turns[0]?.finish(); + await flush(); + expect(selecting).toBeDefined(); + const third = session.send("third"); + await flush(); + selecting?.(); + await flush(); + expect(harness.turns.map((turn) => turn.text)).toEqual(["first", "second"]); + expect(harness.turns[1]?.steered).toEqual(["third"]); + harness.turns[1]?.finish(); + await expect(Promise.all([first, second, third])).resolves.toHaveLength(3); + }); + it("never steers a message stopped while it waits for the run's id", async () => { + const hooks: AcpHooks = { + ...HOOKS, + steering: { + // The run's id comes only when the test announces it. + runId: (update) => + update.sessionUpdate === "announce" ? String(update.runId) : null, + steer: async (acp, sessionId, runId, text) => { + await acp.request("steer", { sessionId, runId, text }); + }, + }, + }; + const harness = fakeHarness({ hold: true }); + const acp = await start(harness, {}, hooks); + const session = await acp.open({ cwd: "/work", systemPrompt: "" }); + const first = session.send("first"); + await flush(); + const second = session.send("second"); + await flush(); + session.interrupt(); + session.update({ sessionUpdate: "announce", runId: "run-1" }); + await flush(); + expect(harness.sent("steer")).toEqual([]); + harness.turns[0]?.finish(); + await expect(Promise.all([first, second])).resolves.toHaveLength(2); + }); +}); + +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("keeps the instructions for the next prompt when the first is rejected", async () => { + const harness = fakeHarness({ rejectPrompts: 1 }); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "Be Buzz." }); + await expect(session.send("first")).resolves.toMatchObject({ + ok: false, + error: "Sign in first", + }); + await expect(session.send("second")).resolves.toMatchObject({ ok: true }); + expect(harness.turns.map((turn) => turn.text)).toEqual([ + "Be Buzz.\n\nsecond", + ]); + }); + + 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("keeps the harness's error code on a failed turn", async () => { + const harness = fakeHarness({ + promptError: { code: -32000, message: "Authentication required" }, + }); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "" }); + await expect(session.send("first")).resolves.toEqual({ + ok: false, + error: "Authentication required", + code: -32000, + }); + }); + + it("drops an earlier prompt's error code when a later one is refused", async () => { + const harness = fakeHarness({ rejectPrompts: 1, stopReason: "refusal" }); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "" }); + void session.send("first"); + // The latest message owns the burst's error. + await expect(session.send("second")).resolves.toEqual({ + ok: false, + error: "Harness refused the request", + }); + }); + + 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", + resolveDirectory: async (path) => path, + 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", + resolveDirectory: goose.resolveDirectory, + tools, + process: new SharedProcess(goose, goose.serve), + pool, + launch, + choice, + fingerprint: () => "", + }); + const acp = await start(harness); + const genericSessions = new AgentSessions({ + label: "Harness", + resolveDirectory: async (path) => path, + 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 new file mode 100644 index 0000000000..239257f315 --- /dev/null +++ b/src/features/agent-harness/acp/process.ts @@ -0,0 +1,487 @@ +// One ACP harness process, driven over line-delimited JSON-RPC. Each +// conversation is one ACP session in it. Each session gets its own loopback +// HTTP MCP server for its agent's Buzz tools, which the app serves and +// answers: the harness is given its address and token in `session/new`, never +// a Buzz key. What a harness adds to plain ACP comes from its `AcpHooks`. +import type { HostMcpServer, HostProcess } from "../../host/service"; +import { AcpSession } from "./session"; +import { allow, errorMessage } from "./steps"; +import type { + AcpHooks, + Choice, + McpServer, + ServeMcp, + Spawn, + ToolServer, +} from "./types"; + +/** The Buzz tools that answer where people see it. A read, or a send that + * failed, still leaves the harness's text to post. */ +const ANSWERS = new Set(["send", "edit", "react"]); +const STOPPED = + "Buzz stopped this session's tools: another program used their token"; + +/** How long a harness has to answer a request that starts it or opens or + * closes a session. One that does not is stopped. A turn's prompt is never + * timed. */ +export const REQUEST_TIMEOUT_MS = 60_000; + +/** The HTTP MCP server every session declares; its tools are named + * `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; + +/** 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; + id?: number | string | null; + method?: string; + params?: Record; + result?: unknown; + error?: { code?: number; message: string; data?: unknown }; +}; +type Pending = { + resolve: (result: unknown) => void; + reject: (error: Error) => void; +}; + +export class AcpProcess { + readonly exited: Promise; + private readonly sessions = new Map(); + private readonly calls = new Map(); + private requests = 0; + private buffer = ""; + private stderr = ""; + private alive = true; + private code: number | null | undefined; + private capabilities: Readonly> = {}; + /** Whether the harness names untitled sessions, read once. */ + private naming: Promise | undefined; + + private constructor( + private readonly process: HostProcess, + 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"})`, + ); + for (const call of this.calls.values()) call.reject(error); + this.calls.clear(); + for (const session of this.sessions.values()) { + session.fail(error.message); + void session.server?.close(); + } + this.sessions.clear(); + return code; + }); + } + + /** 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 { signal } = launch; + signal?.throwIfAborted(); + let acp: AcpProcess | undefined; + const early: string[] = []; + const earlyErrors: string[] = []; + const spawning = spawn(hooks.program, { + cwd: launch.cwd, + ...(launch.args ? { args: launch.args } : {}), + env: { ...hooks.env, ...launch.env, ...PROTECTED_ENV }, + onStdout: (data) => (acp ? acp.read(data) : early.push(data)), + onStderr: (data) => { + if (acp) acp.readError(data); + else if (hooks.diagnoseStartup) earlyErrors.push(data); + }, + }).catch((error: unknown) => { + throw /os error 2|no such file|not found/i.test(errorMessage(error)) + ? new Error(hooks.missing) + : error; + }); + // A cancel while the host starts it rejects at once; the harness that + // starts after it is stopped. + const process = await new Promise((resolve, reject) => { + const cancel = () => { + reject(signal?.reason); + void spawning.then((late) => late.kill()).catch(() => undefined); + }; + signal?.addEventListener("abort", cancel, { once: true }); + void spawning + .then(resolve, reject) + .finally(() => signal?.removeEventListener("abort", cancel)); + }); + acp = new AcpProcess(process, hooks, launch.serve); + for (const data of earlyErrors) acp.readError(data); + for (const data of early) acp.read(data); + // A cancel rejects the pending `initialize` at once, clearing its + // deadline, whether or not the harness ever exits. + const abort = () => { + for (const call of acp.calls.values()) call.reject(signal?.reason); + acp.calls.clear(); + }; + signal?.addEventListener("abort", abort, { once: true }); + try { + signal?.throwIfAborted(); + const init = (await acp.request( + "initialize", + { + protocolVersion: 1, + clientCapabilities: {}, + clientInfo: { name: "buzz", version: "1" }, + }, + REQUEST_TIMEOUT_MS, + )) as { + agentCapabilities?: Record & { + mcpCapabilities?: { http?: boolean }; + }; + }; + signal?.throwIfAborted(); + acp.capabilities = init.agentCapabilities ?? {}; + if (init.agentCapabilities?.mcpCapabilities?.http !== true) + throw new Error( + `This ${hooks.label} cannot reach Buzz tools (no HTTP MCP support)`, + ); + } catch (error) { + if (signal?.aborted) { + void acp.stop(STOP_TIMEOUT_MS); + throw signal.reason; + } + await acp.stop(STOP_TIMEOUT_MS); + // A program `env` cannot find or run exits before it answers. + if (hooks.diagnoseStartup && MISSING_CODES.has(acp.code ?? 0)) + throw new Error(hooks.missing); + throw error; + } finally { + signal?.removeEventListener("abort", abort); + } + return acp; + } + + get running() { + return this.alive; + } + get busy() { + return [...this.sessions.values()].some((session) => session.busy); + } + /** The `agentCapabilities` the harness answered `initialize` with. */ + 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. With `named`, the harness names it instead when + * it names sessions, and `named` takes that name. `tools` answers its Buzz + * tools; without them it has no tools at all. Each turn runs on the + * provider and model `choice` gives as it starts. */ + async open( + launch: Readonly<{ + cwd: string; + systemPrompt: string; + title?: string; + named?: (title: string) => void; + tools?: ToolServer; + choice?: () => Choice; + }>, + ): Promise { + const { hooks } = this; + const named = + launch.named && (await this.names()) ? launch.named : undefined; + const title = named ? undefined : launch.title; + 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; + const { tools } = launch; + if (tools) { + if (!this.serve) + throw new Error(`This ${hooks.label} serves no Buzz tools`); + server = await this.serve((message) => this.mcp(tools, session, message)); + } + const buzz: McpServer | undefined = server && { + type: "http", + name: TOOL_SERVER, + url: server.url, + headers: server.headers, + }; + // The host stops a server whose token a second client used, so the + // session's tools may be someone else's. A session closed here has + // already left `sessions`. + let stopped = false; + void server?.closed.then(() => { + stopped = true; + if (session && this.sessions.get(session.id) === session) { + session.fail(STOPPED); + void this.close(session); + } + }); + let sessionId: string; + try { + ({ sessionId } = (await this.request( + "session/new", + { + cwd: launch.cwd, + mcpServers: buzz ? [buzz] : [], + ...(standard && { + systemPrompt: { + mode: "append", + content: [{ type: "text", text: systemPrompt }], + }, + }), + ...(hooks.newSessionMeta && { + _meta: hooks.newSessionMeta(buzz, { + ...(title ? { title } : {}), + provider, + }), + }), + }, + REQUEST_TIMEOUT_MS, + )) as { sessionId: string }); + } catch (error) { + void server?.close(); + throw error; + } + session = new AcpSession( + this, + sessionId, + server, + launch.choice, + { provider, model: "" }, + prompting && !standard ? systemPrompt : "", + ); + session.named = named; + this.sessions.set(sessionId, session); + if (stopped) { + void this.close(session); + throw new Error(STOPPED); + } + try { + if (systemPrompt && hooks.systemPrompt) + await hooks.systemPrompt(this, sessionId, systemPrompt); + } catch (error) { + void this.close(session); + throw error; + } + return session; + } + + private names() { + const { names } = this.hooks; + this.naming ??= names ? names(this) : Promise.resolve(false); + return this.naming; + } + + /** 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 }, + REQUEST_TIMEOUT_MS, + ).catch((error) => + console.warn(`${this.hooks.label} did not close a session`, error), + ); + } + + kill() { + 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) { + const { label } = this.hooks; + if (!this.alive) + return Promise.reject(new Error(`${label} is not running`)); + const id = ++this.requests; + const answer = new Promise((resolve, reject) => { + if (timeout === undefined) return this.calls.set(id, { resolve, reject }); + const timer = setTimeout(() => { + this.calls.delete(id); + reject( + new Error( + `${label} did not answer ${method} in ${Math.round(timeout / 1000)}s`, + ), + ); + void this.kill(); + }, timeout); + this.calls.set(id, { + resolve: (result) => { + clearTimeout(timer); + resolve(result); + }, + reject: (error) => { + clearTimeout(timer); + reject(error); + }, + }); + }); + this.write({ jsonrpc: "2.0", id, method, params }).catch((error) => { + this.calls.get(id)?.reject(new Error(String(error))); + this.calls.delete(id); + }); + return answer; + } + notify(method: string, params: Record) { + void this.write({ jsonrpc: "2.0", method, params }).catch(() => undefined); + } + + private write(message: Wire) { + return this.process.write(`${JSON.stringify(message)}\n`); + } + private readError(data: string) { + this.stderr = (this.stderr + data).slice(-STDERR_LIMIT); + } + private read(data: string) { + this.buffer += data; + let newline = this.buffer.indexOf("\n"); + while (newline >= 0) { + const line = this.buffer.slice(0, newline).trim(); + this.buffer = this.buffer.slice(newline + 1); + newline = this.buffer.indexOf("\n"); + if (!line) continue; + let message: Wire; + try { + message = JSON.parse(line); + } catch { + continue; + } + this.handle(message); + } + } + private handle(message: Wire) { + const { id, method } = message; + if (method === undefined) { + if (typeof id !== "number") return; + const call = this.calls.get(id); + this.calls.delete(id); + if (message.error) + call?.reject( + Object.assign(new Error(errorMessage(message.error)), { + code: message.error.code, + }), + ); + else call?.resolve(message.result); + return; + } + if (id === undefined || id === null) { + if (method === "session/update") this.update(message.params ?? {}); + return; + } + void this.answer(id, method, message.params ?? {}); + } + private update(params: Record) { + const session = this.sessions.get(String(params.sessionId)); + const update = params.update as Record | undefined; + if (session && update) session.update(update); + } + /** Answers what the harness asks its client: since tools run without + * asking, any permission prompt with allow. */ + private async answer( + id: number | string, + method: string, + params: Record, + ) { + let reply: Wire; + if (method === "session/request_permission") + reply = { jsonrpc: "2.0", id, result: { outcome: allow(params) } }; + else + reply = { + jsonrpc: "2.0", + id, + error: { code: -32601, message: `Buzz does not handle ${method}` }, + }; + return this.write(reply).catch(() => undefined); + } + /** One MCP message for a session's Buzz tools, answered with its JSON-RPC + * response; a notification gets none. A final `send` that succeeded ends + * its turn. The host answers a throw with an internal error. */ + private async mcp( + tools: ToolServer, + session: AcpSession | undefined, + message: unknown, + ): Promise { + const { id, method = "", params } = (message ?? {}) as Wire; + if (id === undefined || id === null) return undefined; + const response = (await tools(session?.conversation, { + id, + method, + params, + })) as Wire; + const call = (params ?? {}) as { + name?: unknown; + arguments?: { final?: unknown }; + }; + const worked = + method === "tools/call" && + !response.error && + !(response.result as { isError?: unknown } | undefined)?.isError; + if (worked && ANSWERS.has(String(call.name))) session?.answered(); + if (worked && call.name === "send" && call.arguments?.final === true) + session?.finished(); + return response; + } +} diff --git a/src/features/agent-harness/acp/session.ts b/src/features/agent-harness/acp/session.ts new file mode 100644 index 0000000000..49cdc37e22 --- /dev/null +++ b/src/features/agent-harness/acp/session.ts @@ -0,0 +1,459 @@ +import { steerPrompt } from "../../../bundled/claude-code/prompt"; +import type { HostMcpServer } from "../../host/service"; +import type { AcpProcess } from "./process"; +import { codeOf, errorMessage, stepLabel } from "./steps"; +import type { ActivityReceipt, Choice, Settled } from "./types"; + +/** The most of a turn's text kept, to post when it answers nowhere in Buzz. */ +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, 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; + /** Takes the first name the harness generates for it; a titled session gets + * none. */ + named: ((title: string) => void) | undefined; + /** The harness's id for the running turn, which a steer must name. */ + private runId: string | undefined; + /** Steers waiting for the running turn's id, which the harness sends just + * after the turn starts. */ + private runWaiters: ((runId: string | undefined) => void)[] = []; + /** A `session/prompt` is in flight. */ + private prompting = false; + private readonly steering = new Set(); + private steered = Promise.resolve(); + /** Turns to run, each with the model and level it was asked for, if any. */ + private readonly queue: { + text: string; + model?: string; + effort?: string; + current?: () => boolean; + refresh?: ActivityReceipt["refresh"]; + waiter?: Waiter; + delivered?: () => void; + }[] = []; + /** The model the running turn runs on; empty is its provider's default. */ + private model = ""; + /** The thinking level the running turn runs on; empty is the harness's own. */ + private level = ""; + /** Messages sent and not yet answered. */ + private pending = 0; + private waiters: Waiter[] = []; + /** The waiter that posts the burst's error or text: the latest message + * acknowledged, else the latest sent. */ + private owner: Waiter | undefined; + private readonly active = new Map(); + private lastStep = ""; + /** The burst's latest error, with the JSON-RPC code of that same error. */ + private lastFailure: { error: string; code?: number } | undefined; + /** Whether this work answered in Buzz, and what the harness wrote meanwhile: a + * provider failure comes back as text, and text is never posted. */ + private acted = false; + private written = ""; + private closed = false; + /** The work the owner stopped, while the harness ends it: the messages it + * had taken, and their activity. What it wrote meanwhile is not posted, and + * messages sent since wait for a turn of their own. `ended` is set once + * the cancel ended the turn, not when the turn finished on its own. */ + private stopped: + | { waiters: Waiter[]; receipts: Set; ended?: true } + | undefined; + private idlers: (() => void)[] = []; + + constructor( + private readonly acp: AcpProcess, + readonly id: string, + /** Serves this session's Buzz tools; closed with it. */ + readonly server?: HostMcpServer, + private readonly choice?: () => Choice, + /** The provider, model and thinking level 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() { + return this.pending > 0; + } + get working() { + return !this.quiet(); + } + get open() { + return !this.closed; + } + /** The thinking level it was put on; empty while it runs on the harness's own. */ + get effort() { + return this.chosen.effort ?? ""; + } + /** Resolves once nothing runs, waits to run or waits for its answer: + * unlike `busy`, it waits out the model call the harness makes after a final + * `send` too. */ + idle(): Promise { + if (this.quiet()) return Promise.resolve(); + return new Promise((resolve) => this.idlers.push(resolve)); + } + + /** Sends one user message, whose turn runs on `model` of the provider and + * thinking level `effort` when given. Resolves when it, and anything sent + * after it, has been answered, or a final `send` ended the work. */ + send( + text: string, + activity: ActivityReceipt = {}, + model?: string, + effort?: string, + ): Promise { + if (this.closed) + return Promise.resolve({ ok: false, error: "The session was closed" }); + this.pending++; + let waiter!: Waiter; + const settled = new Promise((resolve) => { + waiter = resolve; + this.waiters.push(resolve); + }); + // A steer cannot change the model or level, and must not pass a queued + // turn. + if ( + this.prompting && + !this.stopped && + this.acp.hooks.steering && + !this.queue.length && + (!model || model === this.model) && + (!effort || effort === this.level) + ) + void this.steer(text, activity, waiter, model, effort); + else { + this.queue.push({ + text, + ...(activity.delivered ? { delivered: activity.delivered } : {}), + ...(activity.current ? { current: activity.current } : {}), + ...(activity.refresh ? { refresh: activity.refresh } : {}), + waiter, + ...(model ? { model } : {}), + ...(effort ? { effort } : {}), + }); + this.acknowledge(activity, waiter); + void this.pump(); + } + return settled; + } + + /** A final `send` succeeded: the work counts as done at once, unless + * another message is still waiting for its answer. The harness may still end + * its turn with one more model call, whose text nobody posts. */ + finished() { + if (this.pending === 1 && !this.queue.length && !this.steering.size) + this.settle({ ok: true }); + } + + /** Stops the work: the running turn, as ACP's `session/cancel` does, and + * the messages waiting for a turn after it, which the harness cannot stop + * apart. A turn not yet sent is never sent. The work then ends as the + * harness ends it; a message sent before it has runs as a new turn after. */ + interrupt() { + if (!this.working || this.closed) return; + this.queue.length = 0; + this.steered = Promise.resolve(); + this.stopped = { + waiters: [...this.waiters], + receipts: new Set(this.active.values()), + }; + this.acp.notify("session/cancel", { sessionId: this.id }); + } + + /** A Buzz tool that answers, such as `send`, succeeded. */ + answered() { + this.acted = true; + } + + /** Called by the process with each `session/update` for this session. */ + update(update: Record) { + const kind = update.sessionUpdate; + if (kind === "agent_message_chunk" && !this.acted) { + const content = update.content as { text?: unknown } | undefined; + if (typeof content?.text === "string") + this.written = (this.written + content.text).slice(0, WRITTEN_LIMIT); + } + if (kind === "session_info_update") { + const title = typeof update.title === "string" ? update.title.trim() : ""; + const named = this.named; + if (title && named) { + this.named = undefined; + named(title); + } + } + // `undefined` from the hook means no run is active: keep it. + const { steering } = this.acp.hooks; + const runId = steering ? steering.runId(update) : null; + if (runId !== null) { + this.runId = runId; + if (runId) this.ranAs(runId); + } + if (kind === "session_info_update" || runId !== null) return; + const step = stepLabel(update); + if (!step) return; + const label = `${step.tool}\u0000${step.target ?? ""}`; + if (label === this.lastStep) return; + this.lastStep = label; + for (const receipt of this.active.values()) + receipt.step?.(step.tool, step.target); + } + + /** Ends the session's work with `error`; nothing more is sent to it. */ + fail(error: string) { + this.closed = true; + this.queue.length = 0; + this.settle({ ok: false, error }); + } + + private async steer( + text: string, + activity: ActivityReceipt, + waiter: Waiter, + model?: string, + effort?: string, + ) { + this.steering.add(waiter); + const prior = this.steered; + let release!: () => void; + this.steered = new Promise((resolve) => { + release = resolve; + }); + // Work the owner stopped, or that ended meanwhile, takes no more. + const live = () => + this.waiters.includes(waiter) && !this.stopped?.waiters.includes(waiter); + try { + await prior; + if (!live()) return; + const runId = + this.runId ?? + (this.prompting + ? await new Promise((resolve) => + this.runWaiters.push(resolve), + ) + : undefined); + if (activity.refresh) text = steerPrompt(await activity.refresh()); + if (!live()) return; + if (activity.current && !activity.current()) { + this.discard(waiter); + return; + } + const { steering } = this.acp.hooks; + if (runId && steering) { + try { + await steering.steer(this.acp, this.id, runId, text); + if (live()) { + activity.delivered?.(); + this.acknowledge(activity, waiter); + } + return; + } catch { + // The turn ended, or another took its place: run it as the next. + } + } + if (live()) { + const next = this.queue.findIndex( + (turn) => + turn.waiter && + this.waiters.indexOf(turn.waiter) > this.waiters.indexOf(waiter), + ); + this.queue.splice(next < 0 ? this.queue.length : next, 0, { + text, + ...(activity.delivered ? { delivered: activity.delivered } : {}), + ...(activity.current ? { current: activity.current } : {}), + ...(activity.refresh ? { refresh: activity.refresh } : {}), + waiter, + ...(model ? { model } : {}), + ...(effort ? { effort } : {}), + }); + this.acknowledge(activity, waiter); + } + } finally { + release(); + this.steering.delete(waiter); + // Runs the next turn, or ends work whose turn ended during the steer. + void this.pump(); + } + } + + private discard(waiter: Waiter) { + this.waiters = this.waiters.filter((pending) => pending !== waiter); + this.pending = this.waiters.length; + if (this.owner === waiter) this.owner = this.waiters.at(-1); + for (const [scope, receipt] of this.active) + if (receipt.current && !receipt.current()) this.active.delete(scope); + waiter({ ok: true, undelivered: true }); + } + + private get handingOff() { + return [...this.steering].some( + (waiter) => + this.waiters.includes(waiter) && + !this.stopped?.waiters.includes(waiter), + ); + } + + private async pump() { + if (this.prompting || this.handingOff) return; + this.prompting = true; + try { + while (this.queue.length && !this.handingOff && !this.closed) { + // Stopped work has ended, and messages sent since run as new work. + if (this.stopped) this.endStopped(); + const { text, model, effort, current, refresh, waiter, delivered } = + this.queue.shift() ?? { + text: "", + }; + try { + await this.select(model, effort); + const prompt = refresh ? await refresh() : text; + // The owner stopped it while its provider switch was going. + if (this.stopped) { + this.cancelEnded(); + continue; + } + if (current && !current()) { + if (waiter) this.discard(waiter); + continue; + } + delivered?.(); + const lead = this.preamble; + const { stopReason } = (await this.acp.request("session/prompt", { + sessionId: this.id, + prompt: [ + { type: "text", text: lead ? `${lead}\n\n${prompt}` : prompt }, + ], + })) as { stopReason?: string }; + // A rejected prompt never reached the model: the next one leads. + this.preamble = ""; + if (stopReason === "cancelled") this.cancelEnded(); + if (stopReason === "refusal") + this.lastFailure = { + error: `${this.acp.hooks.label} refused the request`, + }; + } catch (error) { + const code = codeOf(error); + this.lastFailure = { + error: errorMessage(error), + ...(code === undefined ? {} : { code }), + }; + } + } + } finally { + this.prompting = false; + this.runId = undefined; + this.ranAs(undefined); + } + if (!this.steering.size && !this.closed) this.done(); + this.wake(); + } + + /** Puts the session on the provider, model and thinking level chosen now, + * or `model` of that provider and level `effort`. A steer joins the running + * turn, so a change waits for the next. */ + private async select(model?: string, effort?: string) { + const chosen = this.choice?.(); + if (!chosen) return; + const want = { + ...chosen, + ...(model ? { model } : {}), + ...(effort ? { effort } : {}), + }; + this.model = want.model; + this.level = want.effort ?? ""; + await this.acp.hooks.select?.(this.acp, this, want); + } + + private ranAs(runId: string | undefined) { + for (const resolve of this.runWaiters.splice(0)) resolve(runId); + } + private acknowledge(activity: ActivityReceipt, waiter: Waiter) { + if (this.waiters.includes(waiter)) this.owner = waiter; + // Watches and timers have no message activity: they must not hide a mention. + if (!activity.step) return; + const scope = activity.activityScope ?? "conversation"; + this.active.get(scope)?.superseded?.(); + this.active.set(scope, activity); + this.lastStep = ""; + activity.acknowledged?.(); + } + private done() { + // Work that wrote text and never answered in Buzz answered only the + // harness: a model that forgot `send`, or one that cannot reach its provider. + const text = this.acted || this.stopped ? "" : this.written.trim(); + const failure = this.lastFailure; + this.settle({ + ...(failure?.error ? { ok: false, ...failure } : { ok: true }), + ...(text ? { text } : {}), + ...(!failure?.error && this.stopped?.ended ? { stopped: true } : {}), + }); + } + /** The owner's cancel ended the turn, if it was stopping it. */ + private cancelEnded() { + if (this.stopped) this.stopped.ended = true; + } + /** Settles the work the owner stopped, leaving messages sent since. */ + private endStopped() { + const stopped = this.stopped; + if (!stopped) return; + this.stopped = undefined; + const failure = this.lastFailure; + this.lastFailure = undefined; + this.acted = false; + this.written = ""; + for (const [scope, receipt] of this.active) + if (stopped.receipts.has(receipt)) this.active.delete(scope); + const owner = + this.owner && stopped.waiters.includes(this.owner) + ? this.owner + : stopped.waiters.at(-1); + if (this.owner === owner) this.owner = undefined; + this.waiters = this.waiters.filter( + (waiter) => !stopped.waiters.includes(waiter), + ); + this.pending = this.waiters.length; + for (const resolve of stopped.waiters) + resolve( + resolve === owner && failure?.error + ? { ok: false, ...failure } + : { + ok: !failure?.error, + ...(!failure?.error && stopped.ended ? { stopped: true } : {}), + }, + ); + } + private settle(settled: Settled) { + this.pending = 0; + this.active.clear(); + this.lastFailure = undefined; + this.acted = false; + this.written = ""; + this.stopped = undefined; + const owner = this.owner ?? this.waiters.at(-1); + this.owner = undefined; + for (const resolve of this.waiters.splice(0)) + resolve( + resolve === owner + ? settled + : { ok: settled.ok, ...(settled.stopped ? { stopped: true } : {}) }, + ); + this.wake(); + } + private quiet() { + return ( + this.closed || + (!this.prompting && + !this.steering.size && + !this.pending && + !this.queue.length) + ); + } + private wake() { + if (this.quiet()) for (const resolve of this.idlers.splice(0)) resolve(); + } +} diff --git a/src/features/agent-harness/acp/steps.ts b/src/features/agent-harness/acp/steps.ts new file mode 100644 index 0000000000..592237ca7b --- /dev/null +++ b/src/features/agent-harness/acp/steps.ts @@ -0,0 +1,71 @@ +// What a harness's session updates show, and how its requests are answered. + +const STEP_KEYS = [ + "description", + "path", + "file_path", + "pattern", + "url", + "query", + "command", +]; + +/** A `session/update` as the step it shows: a tool call by its title and + * target, thinking, or writing. */ +export function stepLabel( + update: Record, +): { tool: string; target?: string } | undefined { + switch (update.sessionUpdate) { + case "tool_call": { + const title = typeof update.title === "string" ? update.title : ""; + // goose titles a call `shell · sleep 10`, or `buzz: send`. + const tool = + title.split(" · ")[0]?.split(/__|: /).at(-1)?.trim() || "Tool"; + const input = (update.rawInput ?? {}) as Record; + const target = STEP_KEYS.map((key) => input[key]).find( + (value): value is string => typeof value === "string" && !!value, + ); + return target ? { tool, target: shortTarget(target) } : { tool }; + } + case "agent_thought_chunk": + return { tool: "Thinking" }; + case "agent_message_chunk": + return { tool: "Writing reply" }; + default: + return undefined; + } +} + +/** The option that allows the call, preferring one that keeps allowing. */ +export function allow(params: Record) { + const options = Array.isArray(params.options) + ? (params.options as { optionId?: unknown; kind?: unknown }[]) + : []; + const option = + options.find((item) => item.kind === "allow_always") ?? + options.find((item) => item.kind === "allow_once"); + return option && typeof option.optionId === "string" + ? { outcome: "selected", optionId: option.optionId } + : { outcome: "cancelled" }; +} + +/** The JSON-RPC code of a harness's error answer, when `error` is one. */ +export function codeOf(error: unknown) { + const code = (error as { code?: unknown } | null)?.code; + return typeof code === "number" ? code : undefined; +} + +export function errorMessage(error: unknown) { + if (error instanceof Error) return error.message; + if (error && typeof error === "object" && "message" in error) { + const { message, data } = error as { message: unknown; data?: unknown }; + return typeof data === "string" && data + ? `${message}: ${data}` + : String(message); + } + return String(error); +} + +function shortTarget(value: string) { + return value.startsWith("/") ? (value.split("/").at(-1) ?? value) : value; +} diff --git a/src/features/agent-harness/acp/types.ts b/src/features/agent-harness/acp/types.ts new file mode 100644 index 0000000000..a9ce9d6a9b --- /dev/null +++ b/src/features/agent-harness/acp/types.ts @@ -0,0 +1,126 @@ +// What the shared ACP client exchanges with the harness plugins built on it. +import type { + HostMcpServer, + HostProcess, + HostProcessOptions, + McpHandler, +} from "../../host/service"; +import type { AcpProcess } from "./process"; + +export type Spawn = ( + id: string, + options?: HostProcessOptions, +) => Promise; +export type ServeMcp = (handler: McpHandler) => Promise; +/** The host's absolute path for a directory, resolving `~/…`. */ +export type ResolveDirectory = (path: string) => Promise; + +/** How a burst of work ended: every message sent so far has been answered, + * or the session or its process is gone. One message of the burst owns its + * `error` and `text`, to post them once; the others get only `ok`. */ +export type Settled = Readonly<{ + ok: boolean; + error?: string; + /** The JSON-RPC code of the harness's error answer behind `error`. */ + code?: number; + /** What the harness wrote, when it answered nowhere in Buzz. Nobody saw it. */ + text?: string; + /** The message gave up before any turn took it in. */ + undelivered?: true; + /** The owner's interrupt ended the turn. */ + stopped?: true; +}>; + +export type ActivityReceipt = Readonly<{ + current?: () => boolean; + /** Rebuilds an Attention prompt after startup or queue waits. */ + refresh?: (fresh?: boolean) => Promise; + /** The prompt was handed to the harness, after cancellation and provider + * checks. */ + delivered?: () => void; + /** Steps show on the latest acknowledged message of each scope. */ + activityScope?: string; + step?: (tool: string, target?: string) => void; + acknowledged?: () => void; + superseded?: () => void; +}>; + +/** Answers one MCP message from a session serving `conversation` with its + * JSON-RPC response. */ +export type ToolServer = ( + conversation: string | undefined, + message: Readonly<{ id: string | number; method: string; params?: unknown }>, +) => Promise; + +/** The provider, model and thinking level a session's turns run on; empty is + * what the harness's own config picks. */ +export type Choice = Readonly<{ + provider: string; + model: string; + effort?: string; +}>; + +/** The HTTP MCP server a session declares in `session/new`. */ +export type McpServer = Readonly<{ + type: "http"; + name: string; + url: string; + headers: HostMcpServer["headers"]; +}>; + +/** 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, or cannot run. */ + missing: string; + /** Keeps stderr written before the host reports the spawn, and reads exit + * 126 or 127 before `initialize` answers as `missing`: for a program the + * owner launches through `env`. */ + diagnoseStartup?: boolean; + /** Its environment, under the launch's own. The Buzz identity is always + * removed, whatever either says. */ + env: Readonly>; + /** Whether the harness names an untitled session itself, read once per + * process. Without it, a session opened to be named keeps its title. */ + names?(process: AcpProcess): Promise; + /** `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. + * 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. 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. */ + runId(update: Record): string | undefined | null; + steer( + process: AcpProcess, + sessionId: string, + runId: string, + text: string, + ): Promise; + }>; + /** Puts a session on `want` before a turn, keeping `session.chosen` to + * 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"; +}>;