From ffc67c9b14bc3eaa0c2339692c0d1a99fcf8cef8 Mon Sep 17 00:00:00 2001 From: Duncan Date: Tue, 27 Oct 2026 11:11:03 -0400 Subject: [PATCH 1/2] refactor(agent-harness): extract shared ACP client and server pool Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- src/bundled/goose/goose.ts | 1100 +++-------------- src/bundled/goose/runtime.test.ts | 58 +- src/bundled/goose/runtime.ts | 17 +- src/bundled/goose/sessions.test.ts | 4 +- src/bundled/goose/sessions.ts | 642 +--------- .../agent-harness/acp/agent-sessions.ts | 633 ++++++++++ src/features/agent-harness/acp/client.test.ts | 592 +++++++++ src/features/agent-harness/acp/process.ts | 487 ++++++++ src/features/agent-harness/acp/session.ts | 459 +++++++ src/features/agent-harness/acp/steps.ts | 71 ++ src/features/agent-harness/acp/types.ts | 126 ++ 11 files changed, 2576 insertions(+), 1613 deletions(-) create mode 100644 src/features/agent-harness/acp/agent-sessions.ts create mode 100644 src/features/agent-harness/acp/client.test.ts create mode 100644 src/features/agent-harness/acp/process.ts create mode 100644 src/features/agent-harness/acp/session.ts create mode 100644 src/features/agent-harness/acp/steps.ts create mode 100644 src/features/agent-harness/acp/types.ts 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..943d0ae66c --- /dev/null +++ b/src/features/agent-harness/acp/client.test.ts @@ -0,0 +1,592 @@ +// 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); + }); +}); + +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"; +}>; From 3c6f0c57ab4a3c6623bc033be67c587a70aaa2b7 Mon Sep 17 00:00:00 2001 From: Duncan Date: Tue, 27 Oct 2026 11:15:50 -0400 Subject: [PATCH 2/2] test(acp): never steer a message stopped while it waits for the run id Covers the case from review: an interrupt that lands while a steer waits for the running turn's id must not send the steer once the id arrives. The live() check after that wait (carried over from main's goose session) is what holds it. Co-authored-by: Will Pfleger Signed-off-by: Will Pfleger --- src/features/agent-harness/acp/client.test.ts | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) diff --git a/src/features/agent-harness/acp/client.test.ts b/src/features/agent-harness/acp/client.test.ts index 943d0ae66c..86e8beeb7c 100644 --- a/src/features/agent-harness/acp/client.test.ts +++ b/src/features/agent-harness/acp/client.test.ts @@ -412,6 +412,32 @@ describe("a session with a steering hook", () => { 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", () => {