From e3b225c14b32c4415ef7a13c40a4d8b660862718 Mon Sep 17 00:00:00 2001 From: Salman Mohammed Date: Tue, 13 Oct 2026 17:59:40 -0400 Subject: [PATCH 01/16] feat(agents): prototype a Goose agent type over ACP A bundled `buzz.goose` plugin whose agents run as one `goose-acp` process each, with one ACP session per conversation. The Buzz tools reach goose over MCP-over-ACP (`mcp/message` on the same pipe), so the process holds no Buzz key. Uses the Claude plugin's Buzz prompt, appended with `_goose/unstable/session/system-prompt/set`; names sessions with `_meta.sessionTitle`; steers mid-turn messages with `_goose/unstable/session/steer`. Provider and model come from the owner's goose config unless the agent names its own. Co-Authored-By: Claude Opus 5.5 Signed-off-by: Salman Mohammed --- crates/plugin-manager/src/lib.rs | 3 + src/bundled/goose/goose-testing.ts | 251 +++++++++++ src/bundled/goose/goose.live.test.ts | 185 ++++++++ src/bundled/goose/goose.test.ts | 304 +++++++++++++ src/bundled/goose/goose.ts | 573 +++++++++++++++++++++++++ src/bundled/goose/index.ts | 72 ++++ src/bundled/goose/manifest.json | 12 + src/bundled/goose/runtime.test.ts | 17 + src/bundled/goose/runtime.ts | 612 +++++++++++++++++++++++++++ src/bundled/goose/sessions.test.ts | 190 +++++++++ src/bundled/goose/sessions.ts | 355 ++++++++++++++++ src/bundled/goose/tabs.tsx | 146 +++++++ src/bundled/index.ts | 7 + 13 files changed, 2727 insertions(+) create mode 100644 src/bundled/goose/goose-testing.ts create mode 100644 src/bundled/goose/goose.live.test.ts create mode 100644 src/bundled/goose/goose.test.ts create mode 100644 src/bundled/goose/goose.ts create mode 100644 src/bundled/goose/index.ts create mode 100644 src/bundled/goose/manifest.json create mode 100644 src/bundled/goose/runtime.test.ts create mode 100644 src/bundled/goose/runtime.ts create mode 100644 src/bundled/goose/sessions.test.ts create mode 100644 src/bundled/goose/sessions.ts create mode 100644 src/bundled/goose/tabs.tsx diff --git a/crates/plugin-manager/src/lib.rs b/crates/plugin-manager/src/lib.rs index dafb6969b8..bad45a0e8b 100644 --- a/crates/plugin-manager/src/lib.rs +++ b/crates/plugin-manager/src/lib.rs @@ -221,6 +221,8 @@ pub fn bundled_manifests() -> Vec { .expect("claude manifest"), serde_json::from_str(include_str!("../../../src/bundled/codex/manifest.json")) .expect("codex manifest"), + serde_json::from_str(include_str!("../../../src/bundled/goose/manifest.json")) + .expect("goose manifest"), serde_json::from_str(include_str!("../../../src/bundled/workflows/manifest.json")) .expect("workflows manifest"), serde_json::from_str(include_str!("../../../src/bundled/feedback/manifest.json")) @@ -262,6 +264,7 @@ fn enabled_by_default(id: &str) -> bool { | "buzz.agents" | "buzz.claude-code" | "buzz.codex" + | "buzz.goose" | "buzz.workflows" | "buzz.sessions" | "block.hosted-communities" diff --git a/src/bundled/goose/goose-testing.ts b/src/bundled/goose/goose-testing.ts new file mode 100644 index 0000000000..239d4f089c --- /dev/null +++ b/src/bundled/goose/goose-testing.ts @@ -0,0 +1,251 @@ +// A stand-in for `goose-acp`'s ACP over stdio, for tests. +import type { + HostProcess, + HostProcessOptions, +} from "../../features/host/service"; + +type Message = { + jsonrpc: "2.0"; + id?: number | string; + method?: string; + params?: Record; + result?: unknown; + error?: { code: number; message: string }; +}; + +export type FakeTurn = { + sessionId: string; + runId: string; + text: string; + /** Texts steered into it. */ + steered: string[]; + finish(stopReason?: string): void; +}; + +export type FakeGoose = { + id: string; + options: HostProcessOptions; + /** Every message written to its stdin. */ + received: Message[]; + /** Each `session/new`'s params. */ + opened: Record[]; + /** Each system prompt set, by session. */ + prompts: Map; + /** Turns, in the order they began. */ + turns: FakeTurn[]; + closed: string[]; + cancelled: string[]; + /** When set, a turn stays open until it is finished. */ + hold: boolean; + killed: boolean; + /** Asks the client a request and resolves with its response. */ + ask(method: string, params: Record): Promise; + /** An MCP message for the Buzz tools from `sessionId`. */ + mcp( + sessionId: string, + method: string, + params?: Record, + ): Promise; + update(sessionId: string, update: Record): void; + exit(code: number | null, stderr?: string): void; +}; + +/** A spawn whose `goose` processes answer like `goose-acp`, and whose + * `workspace` process prints `home` for a `~/…` directory. */ +export function fakeSpawn( + options: Readonly<{ acp?: boolean; home?: string; hold?: boolean }> = {}, +) { + const processes: FakeGoose[] = []; + const spawn = async ( + id: string, + process: HostProcessOptions = {}, + ): Promise => { + let resolveExit!: (code: number | null) => void; + const exited = new Promise((resolve) => { + resolveExit = resolve; + }); + if (id === "workspace") { + const cwd = process.cwd ?? ""; + queueMicrotask(() => { + process.onStdout?.( + `${cwd.replace(/^~/, options.home ?? "/home/test")}\n`, + ); + resolveExit(0); + }); + return { + write: async () => undefined, + end: async () => undefined, + kill: async () => undefined, + exited, + }; + } + let alive = true; + let sessions = 0; + let requests = 0; + const answers = new Map void>(); + const send = (message: Message) => + queueMicrotask(() => { + if (alive) process.onStdout?.(`${JSON.stringify(message)}\n`); + }); + const reply = (id: number | string | undefined, result: unknown) => + send({ jsonrpc: "2.0", ...(id !== undefined && { id }), result }); + const running = new Map(); + const fake: FakeGoose = { + id, + options: process, + received: [], + opened: [], + prompts: new Map(), + turns: [], + closed: [], + cancelled: [], + hold: options.hold ?? false, + killed: false, + ask: (method, params) => { + const requestId = `goose-${++requests}`; + const answer = new Promise((resolve) => + answers.set(requestId, resolve), + ); + send({ jsonrpc: "2.0", id: requestId, method, params }); + return answer; + }, + mcp: (sessionId, method, params = {}) => + fake.ask("mcp/message", { + serverId: "buzz", + requestId: `mcp-${requests + 1}`, + method, + params, + _meta: { sessionId }, + }), + update: (sessionId, update) => + send({ + jsonrpc: "2.0", + method: "session/update", + params: { sessionId, update }, + }), + exit: (code, stderr = "") => { + if (!alive) return; + if (stderr) process.onStderr?.(stderr); + alive = false; + resolveExit(code); + }, + }; + processes.push(fake); + const handle = (message: Message) => { + fake.received.push(message); + const { id, method, params = {} } = message; + if (method === undefined) { + if (id !== undefined) answers.get(id)?.(message); + answers.delete(id ?? ""); + return; + } + const sessionId = String(params.sessionId ?? ""); + switch (method) { + case "initialize": + return reply(id, { + protocolVersion: 1, + agentCapabilities: { + mcpCapabilities: { acp: options.acp ?? true }, + }, + }); + case "session/new": { + fake.opened.push(params); + return reply(id, { sessionId: `session-${++sessions}` }); + } + case "_goose/unstable/session/system-prompt/set": + fake.prompts.set(sessionId, String(params.text)); + return reply(id, {}); + case "session/prompt": { + const runId = `run-${fake.turns.length + 1}`; + const text = (params.prompt as { text: string }[]) + .map((block) => block.text) + .join(""); + const turn: FakeTurn = { + sessionId, + runId, + text, + steered: [], + finish: (stopReason = "end_turn") => { + if (running.get(sessionId) !== turn) return; + running.delete(sessionId); + fake.update(sessionId, { + sessionUpdate: "session_info_update", + _meta: { goose: { activeRunId: null } }, + }); + reply(id, { stopReason }); + }, + }; + fake.turns.push(turn); + running.set(sessionId, turn); + fake.update(sessionId, { + sessionUpdate: "session_info_update", + _meta: { goose: { activeRunId: runId } }, + }); + if (!fake.hold) queueMicrotask(() => turn.finish()); + return; + } + case "_goose/unstable/session/steer": { + const turn = running.get(sessionId); + if (!turn || turn.runId !== params.expectedRunId) + return send({ + jsonrpc: "2.0", + ...(id !== undefined && { id }), + error: { code: -32602, message: "No active run" }, + }); + turn.steered.push( + (params.prompt as { text: string }[]) + .map((block) => block.text) + .join(""), + ); + return reply(id, { runId: turn.runId, messageId: "m" }); + } + case "session/cancel": + fake.cancelled.push(sessionId); + running.get(sessionId)?.finish("cancelled"); + return; + case "session/close": + fake.closed.push(sessionId); + return reply(id, {}); + default: + return send({ + jsonrpc: "2.0", + ...(id !== undefined && { id }), + error: { code: -32601, message: `Unknown ${method}` }, + }); + } + }; + let buffer = ""; + return { + write: async (data) => { + if (!alive) throw new Error("goose is not running"); + buffer += data; + let newline = buffer.indexOf("\n"); + while (newline >= 0) { + const line = buffer.slice(0, newline); + buffer = buffer.slice(newline + 1); + newline = buffer.indexOf("\n"); + handle(JSON.parse(line)); + } + }, + end: async () => undefined, + kill: async () => { + fake.killed = true; + fake.exit(null); + }, + exited, + }; + }; + return { + spawn, + processes, + /** The `goose` processes started, without `workspace` lookups. */ + get geese() { + return processes.filter((process) => process.id === "goose"); + }, + }; +} + +/** Lets queued microtasks and the timers they start run. */ +export const flush = async (rounds = 10) => { + for (let index = 0; index < rounds; index++) await Promise.resolve(); +}; diff --git a/src/bundled/goose/goose.live.test.ts b/src/bundled/goose/goose.live.test.ts new file mode 100644 index 0000000000..409b1bc836 --- /dev/null +++ b/src/bundled/goose/goose.live.test.ts @@ -0,0 +1,185 @@ +// @vitest-environment node +// Drives a real `goose-acp` with the owner's goose config: run with +// GOOSE_ACP_BIN=/path/to/goose-acp. The Buzz tools are listed as the app lists +// them; their calls are recorded, not sent to a relay. +import { spawn as spawnChild } from "node:child_process"; +import { mkdtempSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import { respond } from "../../buzz-mcp/rpc"; +import type { HostProcessOptions } from "../../features/host/service"; +import { systemPrompt } from "../claude-code/prompt"; +import { GooseProcess, type Spawn, type ToolServer } from "./goose"; + +const BIN = process.env.GOOSE_ACP_BIN; + +const spawn: Spawn = async (_id, options: HostProcessOptions = {}) => { + const env: Record = { ...process.env }; + for (const [key, value] of Object.entries(options.env ?? {})) + if (value === null) delete env[key]; + else env[key] = value; + const child = spawnChild(BIN ?? "goose-acp", [], { + cwd: options.cwd, + env, + stdio: ["pipe", "pipe", "pipe"], + }); + child.stdout.setEncoding("utf8").on("data", (data) => { + if (process.env.GOOSE_ACP_DEBUG) process.stderr.write(`<< ${data}`); + options.onStdout?.(data); + }); + child.stderr.setEncoding("utf8").on("data", (data) => { + if (process.env.GOOSE_ACP_DEBUG) process.stderr.write(data); + options.onStderr?.(data); + }); + const exited = new Promise((resolve) => + child.on("exit", (code) => resolve(code)), + ); + return { + write: (data) => + new Promise((resolve, reject) => + child.stdin.write(data, (error) => (error ? reject(error) : resolve())), + ), + end: async (data) => { + child.stdin.end(data); + }, + kill: async () => { + child.kill(); + }, + exited, + }; +}; + +type Call = { + conversation: string | undefined; + method: string; + params: unknown; +}; +type Send = { + name?: string; + arguments?: { text?: string; final?: boolean }; +}; + +/** A goose whose Buzz tool calls and own requests are recorded; sends + * succeed without reaching a relay. */ +async function live() { + const calls: Call[] = []; + const tools: ToolServer = async (conversation, message) => { + calls.push({ + conversation, + method: message.method, + params: message.params, + }); + if (message.method === "tools/call") + return { + jsonrpc: "2.0", + id: message.id, + result: { content: [{ type: "text", text: "Sent." }] }, + }; + return respond(undefined, {}, message); + }; + const cwd = mkdtempSync(join(tmpdir(), "buzz-goose-")); + const goose = await GooseProcess.start(spawn, { cwd, tools }); + const requests: { method: string; ok?: boolean }[] = []; + const request = goose.request.bind(goose); + goose.request = (method, params) => { + const record: { method: string; ok?: boolean } = { method }; + requests.push(record); + const answer = request(method, params); + answer.then( + () => (record.ok = true), + () => (record.ok = false), + ); + return answer; + }; + const sends = () => + calls + .filter((call) => call.method === "tools/call") + .map((call) => call.params as Send) + .filter((call) => call.name === "send"); + return { goose, cwd, calls, requests, sends }; +} + +describe.skipIf(!BIN)("goose-acp", () => { + it("answers with the Buzz tools, the Buzz prompt and a title", { + timeout: 180_000, + }, async () => { + const { goose, cwd, calls, sends } = await live(); + const code = `zebra-${Math.random().toString(36).slice(2, 8)}`; + try { + const title = "Ada · #general · 01234567"; + const session = await goose.open({ + cwd, + title, + systemPrompt: systemPrompt({ + scope: "thread", + cwd, + agent: { name: "Ada", pubkey: "a".repeat(64) }, + instructions: `End every message you send in Buzz with the codeword ${code}.`, + }), + }); + session.conversation = "channel/root"; + const info = (await goose.request("_goose/unstable/session/info", { + sessionId: session.id, + })) as { session: { title?: string } }; + expect(info.session.title).toBe(title); + const settled = await session.send("Say hello to me in Buzz."); + expect(settled).toEqual({ ok: true }); + expect(calls.some((call) => call.method === "tools/list")).toBe(true); + expect( + calls.find((call) => call.method === "tools/call")?.conversation, + ).toBe("channel/root"); + const last = sends().at(-1); + expect(last?.arguments?.final).toBe(true); + expect(last?.arguments?.text).toContain(code); + } finally { + await goose.kill(); + } + }); + + it("steers a message into the running turn", { + timeout: 180_000, + }, async () => { + const { goose, cwd, requests, sends } = await live(); + try { + const session = await goose.open({ + cwd, + systemPrompt: + "You are a Buzz agent. You reply in Buzz only by calling the send tool, once, with final true.", + }); + const steps: string[] = []; + const first = session.send( + "Run the shell command `sleep 10`, then reply in Buzz with the word done.", + { step: (tool) => steps.push(tool) }, + ); + await expect + .poll( + () => + steps.some( + (step) => step !== "Thinking" && step !== "Writing reply", + ), + { + timeout: 60_000, + interval: 200, + }, + ) + .toBe(true); + const second = session.send("Also put the word banana in that reply."); + await expect(first).resolves.toEqual({ ok: true }); + await expect(second).resolves.toEqual({ ok: true }); + expect( + requests.filter( + (item) => item.method === "_goose/unstable/session/steer", + ), + ).toEqual([{ method: "_goose/unstable/session/steer", ok: true }]); + expect( + requests.filter((item) => item.method === "session/prompt"), + ).toHaveLength(1); + expect(sends().at(-1)?.arguments?.text?.toLowerCase()).toContain( + "banana", + ); + } finally { + await goose.kill(); + } + }); +}); diff --git a/src/bundled/goose/goose.test.ts b/src/bundled/goose/goose.test.ts new file mode 100644 index 0000000000..c13a7ff3a2 --- /dev/null +++ b/src/bundled/goose/goose.test.ts @@ -0,0 +1,304 @@ +import { describe, expect, it, vi } from "vitest"; +import { GooseProcess, stepLabel, type ToolServer } from "./goose"; +import { fakeSpawn, flush } from "./goose-testing"; + +const ok: ToolServer = async (_conversation, message) => ({ + jsonrpc: "2.0", + id: message.id, + result: { content: [{ type: "text", text: "done" }] }, +}); + +const start = async (tools: ToolServer = ok, acp = true) => { + const fake = fakeSpawn({ acp }); + const goose = await GooseProcess.start(fake.spawn, { + cwd: "/work", + env: { GOOSE_MODEL: "m" }, + tools, + }); + const [process] = fake.geese; + if (!process) throw new Error("goose did not start"); + return { goose, process, fake }; +}; + +describe("GooseProcess", () => { + it("starts goose-acp without the app's Buzz identity", async () => { + const { process } = await start(); + expect(process.options.cwd).toBe("/work"); + expect(process.options.env).toMatchObject({ + GOOSE_MODE: "auto", + GOOSE_DISABLE_SESSION_NAMING: "true", + GOOSE_MODEL: "m", + BUZZ_PRIVATE_KEY: null, + BUZZ_AUTH_TAG: null, + BUZZ_RELAY_URL: null, + }); + }); + + it("refuses a goose that cannot serve the Buzz tools", async () => { + const fake = fakeSpawn({ acp: false }); + await expect( + GooseProcess.start(fake.spawn, { cwd: "/work", tools: ok }), + ).rejects.toThrow(/MCP-over-ACP/); + expect(fake.geese[0]?.killed).toBe(true); + }); + + it("opens a session with the Buzz tools and its system prompt", async () => { + const { goose, process } = await start(); + const session = await goose.open({ cwd: "/work", systemPrompt: "Be Buzz" }); + const buzz = { type: "acp", name: "buzz", serverId: "buzz" }; + expect(process.opened).toEqual([ + { + cwd: "/work", + mcpServers: [buzz], + _meta: { + enabledExtensions: [ + { type: "platform", name: "developer" }, + { type: "mcp", server: buzz }, + ], + }, + }, + ]); + expect(process.prompts.get(session.id)).toBe("Be Buzz"); + }); + + it("settles a message when its turn ends", async () => { + const { goose, process } = await start(); + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + await expect(session.send("hello")).resolves.toEqual({ ok: true }); + expect(process.turns.map((turn) => turn.text)).toEqual(["hello"]); + expect(session.busy).toBe(false); + }); + + it("answers the Buzz tools' MCP messages for the session's conversation", async () => { + const tools = vi.fn(ok); + const { goose, process } = await start(tools); + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + session.conversation = "channel/root"; + const response = await process.mcp(session.id, "tools/list"); + expect(tools).toHaveBeenCalledWith("channel/root", { + id: expect.any(String), + method: "tools/list", + params: {}, + }); + expect(response.result).toEqual({ + result: { content: [{ type: "text", text: "done" }] }, + }); + }); + + it("returns a tool server's failure as the MCP error", async () => { + const { goose, process } = await start(async () => { + throw new Error("relay down"); + }); + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const response = await process.mcp(session.id, "tools/call", { + name: "send", + arguments: {}, + }); + expect(response.result).toEqual({ + error: { code: -32603, message: "relay down" }, + }); + }); + + it("counts the work done when a final send succeeds", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const settled = session.send("hello"); + await flush(); + await process.mcp(session.id, "tools/call", { + name: "send", + arguments: { content: "hi", final: true }, + }); + await expect(settled).resolves.toEqual({ ok: true }); + expect(session.busy).toBe(false); + process.turns[0]?.finish(); + }); + + it("steers a message that arrives mid-turn into it", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const acknowledged = vi.fn(); + const first = session.send("one"); + await flush(); + const second = session.send("two", { step: () => {}, acknowledged }); + await flush(); + expect(process.turns[0]?.steered).toEqual(["two"]); + expect(acknowledged).toHaveBeenCalled(); + process.turns[0]?.finish(); + await expect(first).resolves.toEqual({ ok: true }); + await expect(second).resolves.toEqual({ ok: true }); + expect(process.turns).toHaveLength(1); + }); + + it("steers a message sent before goose names the turn", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const first = session.send("one"); + const second = session.send("two"); + await flush(); + expect(process.turns.map((turn) => [turn.text, turn.steered])).toEqual([ + ["one", ["two"]], + ]); + process.turns[0]?.finish(); + await expect(first).resolves.toEqual({ ok: true }); + await expect(second).resolves.toEqual({ ok: true }); + }); + + it("names a session with its title", async () => { + const { goose, process } = await start(); + await goose.open({ + cwd: "/work", + systemPrompt: "", + title: "Ada · #general", + }); + expect(process.opened[0]?._meta).toMatchObject({ + sessionTitle: "Ada · #general", + }); + }); + + it("runs a message the turn could not take as the next turn", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const first = session.send("one"); + await flush(); + // The run ended on goose's side, but its answer has not arrived yet. + process.update(session.id, { + sessionUpdate: "session_info_update", + _meta: { goose: { activeRunId: "run-gone" } }, + }); + await flush(); + const second = session.send("two"); + await flush(); + process.turns[0]?.finish(); + await flush(); + expect(process.turns.map((turn) => turn.text)).toEqual(["one", "two"]); + process.turns[1]?.finish(); + await expect(first).resolves.toEqual({ ok: true }); + await expect(second).resolves.toEqual({ ok: true }); + }); + + it("shows the latest message's steps", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const step = vi.fn(); + void session.send("one", { step }); + await flush(); + process.update(session.id, { + sessionUpdate: "tool_call", + toolCallId: "t", + title: "shell · ls", + rawInput: { command: "ls" }, + }); + process.update(session.id, { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "hi" }, + }); + await flush(); + expect(step.mock.calls).toEqual([ + ["shell", "ls"], + ["Writing reply", undefined], + ]); + process.turns[0]?.finish(); + }); + + it("fails work that only wrote text, with what goose wrote", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const settled = session.send("hello"); + await flush(); + process.update(session.id, { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Network error: no provider" }, + }); + await flush(); + process.turns[0]?.finish(); + await expect(settled).resolves.toEqual({ + ok: false, + error: "Network error: no provider", + }); + }); + + it("ends work that acted in Buzz without its closing text", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const settled = session.send("hello"); + await flush(); + await process.mcp(session.id, "tools/call", { + name: "send", + arguments: { content: "hi" }, + }); + process.update(session.id, { + sessionUpdate: "agent_message_chunk", + content: { type: "text", text: "Sent it." }, + }); + await flush(); + process.turns[0]?.finish(); + await expect(settled).resolves.toEqual({ ok: true }); + }); + + it("allows what goose asks permission for", async () => { + const { process } = await start(); + const response = await process.ask("session/request_permission", { + sessionId: "s", + options: [ + { optionId: "no", kind: "reject_once" }, + { optionId: "once", kind: "allow_once" }, + { optionId: "always", kind: "allow_always" }, + ], + }); + expect(response.result).toEqual({ + outcome: { outcome: "selected", optionId: "always" }, + }); + }); + + it("fails a working session with goose's last words when it exits", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const settled = session.send("hello"); + await flush(); + process.exit(1, "starting\nNo provider configured\n"); + await expect(settled).resolves.toEqual({ + ok: false, + error: "No provider configured", + }); + expect(goose.running).toBe(false); + }); + + it("cancels a session's turn when it closes", async () => { + const { goose, process } = await start(); + process.hold = true; + const session = await goose.open({ cwd: "/work", systemPrompt: "" }); + const settled = session.send("hello"); + await flush(); + await goose.close(session); + expect(process.cancelled).toEqual([session.id]); + expect(process.closed).toEqual([session.id]); + await expect(settled).resolves.toMatchObject({ ok: false }); + }); +}); + +describe("stepLabel", () => { + it("names a tool call by its tool and target, as goose titles it", () => { + expect( + stepLabel({ + sessionUpdate: "tool_call", + title: "shell · sleep 10", + rawInput: { command: "sleep 10" }, + }), + ).toEqual({ tool: "shell", target: "sleep 10" }); + expect( + stepLabel({ + sessionUpdate: "tool_call", + title: "buzz: send", + rawInput: { text: "hi", path: "/a/b/notes.md" }, + }), + ).toEqual({ tool: "send", target: "notes.md" }); + }); +}); diff --git a/src/bundled/goose/goose.ts b/src/bundled/goose/goose.ts new file mode 100644 index 0000000000..771df52140 --- /dev/null +++ b/src/bundled/goose/goose.ts @@ -0,0 +1,573 @@ +// One `goose-acp` process for one agent, driven over ACP's line-delimited +// JSON-RPC. Each conversation the agent is in is one ACP session in it. The +// Buzz tools are served over the same pipe with MCP-over-ACP: a session +// declares an `acp` MCP server, and goose sends that server's MCP messages +// back as `mcp/message` requests, so the process holds no Buzz key. +import type { + HostProcess, + HostProcessOptions, +} from "../../features/host/service"; + +export type Spawn = ( + id: string, + options?: HostProcessOptions, +) => Promise; + +/** How a burst of work ended: every message sent so far has been answered, + * or the session or its process is gone. */ +export type Settled = Readonly<{ ok: boolean; error?: string }>; + +export type ActivityReceipt = Readonly<{ + /** 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; method: string; params?: unknown }>, +) => Promise; + +/** The `acp` MCP server every session declares; goose names its tools + * `buzz__`. */ +export const TOOL_SERVER = "buzz"; +const ACP_SERVER = { type: "acp", name: TOOL_SERVER, serverId: TOOL_SERVER }; +/** The whole tool set: goose's own developer tools and the Buzz tools. Sent + * as the session's selection, so extensions in the owner's goose config and + * plugins never load into a Buzz agent. */ +const EXTENSIONS = [ + { type: "platform", name: "developer" }, + { type: "mcp", server: ACP_SERVER }, +]; + +/** What every `goose-acp` runs with: tools run without asking, sessions are + * not named (one model call per turn, not two), and no Buzz identity reaches + * it from the app's environment. */ +export const ENV: Readonly> = { + GOOSE_MODE: "auto", + GOOSE_DISABLE_SESSION_NAMING: "true", + BUZZ_PRIVATE_KEY: null, + BUZZ_AUTH_TAG: null, + BUZZ_RELAY_URL: null, +}; + +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 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; + + private constructor( + private readonly process: HostProcess, + private readonly tools: ToolServer, + ) { + 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); + this.sessions.clear(); + return code; + }); + } + + /** Starts `goose-acp` and checks it can serve the Buzz tools. */ + static async start( + spawn: Spawn, + launch: Readonly<{ + cwd: string; + env?: Readonly>; + tools: ToolServer; + }>, + ) { + let goose: GooseProcess | undefined; + const early: string[] = []; + const process = await spawn("goose", { + cwd: launch.cwd, + env: { ...ENV, ...launch.env }, + onStdout: (data) => (goose ? goose.read(data) : early.push(data)), + onStderr: (data) => goose?.readError(data), + }); + goose = new GooseProcess(process, launch.tools); + for (const data of early) goose.read(data); + try { + const init = (await goose.request("initialize", { + protocolVersion: 1, + clientCapabilities: {}, + clientInfo: { name: "buzz", version: "1" }, + })) as { + agentCapabilities?: { mcpCapabilities?: { acp?: boolean } }; + }; + if (!init.agentCapabilities?.mcpCapabilities?.acp) + throw new Error( + "This goose cannot serve Buzz tools (no MCP-over-ACP 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. */ + async open( + launch: Readonly<{ cwd: string; systemPrompt: string; title?: string }>, + ): Promise { + const { sessionId } = (await this.request("session/new", { + cwd: launch.cwd, + mcpServers: [ACP_SERVER], + _meta: { + enabledExtensions: EXTENSIONS, + ...(launch.title ? { sessionTitle: launch.title } : {}), + }, + })) as { sessionId: string }; + const session = new GooseSession(this, sessionId); + this.sessions.set(sessionId, session); + try { + // Appended under goose's own prompt, which carries each extension's + // instructions; `set` would replace it, and render ours as a template. + await this.request("_goose/unstable/session/system-prompt/set", { + sessionId, + mode: "append", + key: "buzz", + text: launch.systemPrompt, + }); + } catch (error) { + void this.close(session); + throw error; + } + return session; + } + + /** Ends a session. Its turn, if any, is cancelled first. */ + async close(session: GooseSession) { + this.sessions.delete(session.id); + session.fail("The session was closed"); + if (!this.alive) return; + this.notify("session/cancel", { sessionId: session.id }); + await this.request("session/close", { sessionId: session.id }).catch( + (error) => console.warn("goose did not close a session", error), + ); + } + + kill() { + return this.process.kill(); + } + + /** Sends a request and resolves with its result. */ + request(method: string, params: Record) { + if (!this.alive) return Promise.reject(new Error("goose is not running")); + const id = ++this.requests; + const answer = new Promise((resolve, reject) => + this.calls.set(id, { resolve, reject }), + ); + 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: the Buzz tools' MCP messages and, + * 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 === "mcp/message") reply = await this.mcp(id, params); + else 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 the Buzz tools. Its outcome is the response's + * `result` or `error`; a final `send` that succeeded ends its turn. */ + private async mcp( + id: number | string, + params: Record, + ): Promise { + const sessionId = String( + (params._meta as { sessionId?: unknown } | undefined)?.sessionId ?? "", + ); + const method = String(params.method ?? ""); + const inner = { + id: String(params.requestId ?? id), + method, + params: params.params, + }; + if (params.serverId !== TOOL_SERVER) + return { + jsonrpc: "2.0", + id, + error: { + code: -32602, + message: `No MCP server ${String(params.serverId)}`, + }, + }; + let response: Wire; + try { + response = (await this.tools( + this.sessions.get(sessionId)?.conversation, + inner, + )) as Wire; + } catch (error) { + response = { error: { code: -32603, message: errorMessage(error) } }; + } + const session = this.sessions.get(sessionId); + if (method === "tools/call") session?.used(); + if (response.error) + return { jsonrpc: "2.0", id, result: { error: response.error } }; + const call = (inner.params ?? {}) as { + name?: unknown; + arguments?: { final?: unknown }; + }; + if ( + method === "tools/call" && + call.name === "send" && + call.arguments?.final === true && + !(response.result as { isError?: unknown } | undefined)?.isError + ) + session?.finished(); + return { jsonrpc: "2.0", id, result: { result: response.result } }; + } +} + +/** One conversation's ACP session. Messages sent while a turn runs are + * steered into it; one that arrives as the turn ends runs as the next turn. */ +export class GooseSession { + /** The conversation it serves. */ + conversation: string | 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 steering = 0; + private readonly queue: string[] = []; + /** Messages sent and not yet answered. */ + private pending = 0; + private waiters: ((settled: Settled) => void)[] = []; + private readonly active = new Map(); + private lastStep = ""; + private lastError: string | undefined; + /** Whether this work called a Buzz tool, 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; + + constructor( + private readonly goose: GooseProcess, + readonly id: string, + ) {} + + get busy() { + return this.pending > 0; + } + get open() { + return !this.closed; + } + + /** Sends one user message. Resolves when it, and anything sent after it, + * has been answered, or a final `send` ended the work. */ + send(text: string, activity: ActivityReceipt = {}): Promise { + if (this.closed) + return Promise.resolve({ ok: false, error: "The session was closed" }); + this.pending++; + const settled = new Promise((resolve) => + this.waiters.push(resolve), + ); + if (this.prompting) void this.steer(text, activity); + else { + this.queue.push(text); + this.acknowledge(activity); + 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) + this.settle({ ok: true }); + } + + /** goose called a Buzz tool. */ + used() { + 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, 2_000); + } + if (kind === "session_info_update") { + 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) { + this.steering++; + try { + const runId = + this.runId ?? + (this.prompting + ? await new Promise((resolve) => + this.runWaiters.push(resolve), + ) + : undefined); + if (runId) { + try { + await this.goose.request("_goose/unstable/session/steer", { + sessionId: this.id, + prompt: [{ type: "text", text }], + expectedRunId: runId, + }); + this.acknowledge(activity); + return; + } catch { + // The turn ended, or another took its place: run it as the next. + } + } + this.queue.push(text); + this.acknowledge(activity); + } finally { + this.steering--; + } + void this.pump(); + } + + private async pump() { + if (this.prompting) return; + this.prompting = true; + try { + while (this.queue.length && !this.closed) { + const text = this.queue.shift() ?? ""; + try { + const { stopReason } = (await this.goose.request("session/prompt", { + sessionId: this.id, + prompt: [{ type: "text", text }], + })) as { stopReason?: string }; + if (stopReason === "refusal") + this.lastError = "goose refused the request"; + } catch (error) { + this.lastError = errorMessage(error); + } + } + } finally { + this.prompting = false; + this.runId = undefined; + this.ranAs(undefined); + } + if (!this.steering && !this.closed) this.done(); + } + + private ranAs(runId: string | undefined) { + for (const resolve of this.runWaiters.splice(0)) resolve(runId); + } + private acknowledge(activity: ActivityReceipt) { + // 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 acted in Buzz left the conversation + // waiting: its text says why, as when goose cannot reach its provider. + const error = + this.lastError ?? + (!this.acted && this.written.trim() ? this.written.trim() : undefined); + this.settle(error ? { ok: false, error } : { ok: true }); + } + private settle(settled: Settled) { + this.pending = 0; + this.active.clear(); + this.lastError = undefined; + this.acted = false; + this.written = ""; + for (const resolve of this.waiters.splice(0)) resolve(settled); + } +} + +/** 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); +} + +function shortTarget(value: string) { + return value.startsWith("/") ? (value.split("/").at(-1) ?? value) : value; +} diff --git a/src/bundled/goose/index.ts b/src/bundled/goose/index.ts new file mode 100644 index 0000000000..72478c9629 --- /dev/null +++ b/src/bundled/goose/index.ts @@ -0,0 +1,72 @@ +// goose agents: an agent type whose agents are goose on this computer, one +// `goose-acp` process per agent. Each conversation the agent is in is one ACP +// session in it; goose answers with the in-process Buzz tools, served to it +// over the same ACP connection. +import { isTauri } from "@tauri-apps/api/core"; +import type { PluginModule } from "../../plugins/api"; +import { GooseLogoIcon } from "../../shared/design-system/icons"; +import { type Config, config, DEFAULT_CONFIG, GooseRuntime } from "./runtime"; +import { Create, SettingsTab } from "./tabs"; + +export const TYPE_ID = "goose"; +/** A turn's run returns once goose has it, so this bounds only the hand-over. */ +const HANDOVER_MS = 2 * 60_000; + +export const inject = ["agents", "host"]; +export const apply: PluginModule["apply"] = (ctx) => { + // Only the desktop app can start `goose-acp`, and file uploads run `base64`, + // which stock Windows lacks; no unusable type elsewhere. + if (!isTauri() || !/Mac|Linux/i.test(navigator.platform)) return; + const { agents, host } = ctx; + const typeKey = `${ctx.pluginOwner?.id ?? "buzz.goose"}/${TYPE_ID}`; + const runtime = new GooseRuntime(host, agents.relay); + const sync = () => { + const snapshot = agents.snapshot(); + if (snapshot.status === "loading") return; + runtime.sync(snapshot.running.filter((agent) => agent.type === typeKey)); + }; + if (host.spawn) { + const unsubscribe = agents.subscribe(sync); + sync(); + ctx.effect(() => () => { + unsubscribe(); + runtime.dispose(); + }); + } + + agents.register({ + id: TYPE_ID, + title: "Goose", + icon: GooseLogoIcon, + description: + "goose on this computer, with the provider in your goose config. Each thread it is mentioned in is one continuing session.", + defaults: () => ({ config: DEFAULT_CONFIG }), + Create, + adopt: (carry, saved) => ({ + ...saved, + instructions: carry.instructions ?? saved.instructions, + workspace: carry.workspace ?? saved.workspace, + }), + carry: (saved) => { + const { instructions, workspace } = config(saved); + return { + ...(instructions.trim() && { instructions }), + ...(workspace !== DEFAULT_CONFIG.workspace && { workspace }), + }; + }, + summary: (agent) => { + const { model, provider } = config(agent.config); + return [provider, model].filter(Boolean).join(" · ") || "Goose"; + }, + tabs: [ + { + id: "settings", + title: "Settings", + role: "settings", + component: SettingsTab, + }, + ], + run: (delivery) => runtime.run(delivery), + timeoutMs: HANDOVER_MS, + }); +}; diff --git a/src/bundled/goose/manifest.json b/src/bundled/goose/manifest.json new file mode 100644 index 0000000000..7e2a58aacf --- /dev/null +++ b/src/bundled/goose/manifest.json @@ -0,0 +1,12 @@ +{ + "id": "buzz.goose", + "name": "Goose", + "apiVersion": 1, + "host": { + "processes": [ + { "id": "goose", "program": "goose-acp" }, + { "id": "read", "program": "base64", "args": ["-i"] }, + { "id": "workspace", "program": "pwd" } + ] + } +} diff --git a/src/bundled/goose/runtime.test.ts b/src/bundled/goose/runtime.test.ts new file mode 100644 index 0000000000..8e86916fb7 --- /dev/null +++ b/src/bundled/goose/runtime.test.ts @@ -0,0 +1,17 @@ +import { describe, expect, it } from "vitest"; +import { sessionTitle } from "./runtime"; + +describe("sessionTitle", () => { + it("names the agent, the channel and a thread's root", () => { + expect(sessionTitle("Ada", "#general", "0123456789abcdef")).toBe( + "Ada · #general · 01234567", + ); + expect(sessionTitle(undefined, "DM")).toBe("Buzz agent · DM"); + }); + + it("keeps the root when a long name is cut", () => { + const title = sessionTitle("A".repeat(100), "#general", "0123456789"); + expect(title).toHaveLength(80); + expect(title.endsWith(" · 01234567")).toBe(true); + }); +}); diff --git a/src/bundled/goose/runtime.ts b/src/bundled/goose/runtime.ts new file mode 100644 index 0000000000..3f386b2939 --- /dev/null +++ b/src/bundled/goose/runtime.ts @@ -0,0 +1,612 @@ +// Turns Agents deliveries into goose turns. Each agent of this type has its +// own `goose-acp` process; a delivery becomes a prompt in the harness format, +// sent to the ACP session for its conversation; a watch or timer wake runs as +// a one-off turn in a new session, in its Interest's lane. The run returns once +// the turn is handed over, so one agent works in several conversations at once. +// goose acts in Buzz with the in-process Buzz tools over MCP-over-ACP, signed +// by the agent's native-held key; the runtime only reports a turn that fails. +import type { Context, Memory } from "../../buzz-mcp/client"; +import { respond } from "../../buzz-mcp/rpc"; +import { appClient } from "../../buzz-mcp/app-client"; +import { + fromOwner, + type Agent, + type AgentHandle, + type Delivery, +} from "../../features/agents/service"; +import type { Host } from "../../features/host/service"; +import type { EventData } from "../../features/relay/events"; +import type { RelayData } from "../../features/relay/service"; +import { threadReference } from "../../features/relay/thread-reference"; +import { Lanes } from "../claude-code/lanes"; +import { + ATTENTION_REVIEW, + CONTEXT_LIMIT, + type Scope, + steerPrompt, + systemPrompt, + timerPrompt, + turnPrompt, + watchPrompt, +} from "../claude-code/prompt"; +import type { Settled, Spawn, ToolServer } from "./goose"; +import { AgentSessions } from "./sessions"; + +export type Config = Readonly<{ + /** A model the provider offers; empty uses the goose config's. */ + model: string; + /** A goose provider; empty uses the goose config's. */ + provider: string; + instructions: string; + /** The agent's working directory, absolute or `~/…`. */ + workspace: string; + /** Whether each thread is its own session, or the whole channel is one. */ + scope: Scope; +}>; +export const DEFAULT_CONFIG: Config = Object.freeze({ + model: "", + provider: "", + instructions: "", + workspace: "~/.buzz", + scope: "thread", +}); +export const config = (value: unknown): Config => ({ + ...DEFAULT_CONFIG, + ...(value && typeof value === "object" ? (value as Partial) : {}), +}); + +/** How long a read of the agent's core memory is reused. */ +const MEMORY_TTL_MS = 5 * 60_000; +const MEMORY_TIMEOUT_MS = 5_000; +const NAMES_TIMEOUT_MS = 2_000; +const CHAT = [9, 40002]; +const TITLE_LIMIT = 80; + +/** A session's title in goose, as harness agents name theirs: + * `Agent · #channel`, and the thread's root when each thread is a session. */ +export function sessionTitle( + agent: string | undefined, + place: string, + root?: string, +) { + const suffix = root ? ` · ${root.slice(0, 8)}` : ""; + const title = `${agent?.trim() || "Buzz agent"} · ${place}` + .replace(/\s+/g, " ") + .slice(0, TITLE_LIMIT - suffix.length) + .trim(); + return `${title}${suffix}`; +} + +type Entry = { + sessions: AgentSessions; + config: Config; + /** The team sections of the latest run; a change reopens idle sessions. */ + team: string; + /** The agent's name, for its prompt; a rename reopens idle sessions. */ + name?: string; + /** Whether the owner has attention on: new sessions get its tools, its + * guidance and the end-of-turn review. */ + attention: boolean; + /** Watch and timer turns, one lane per Interest. */ + lanes: Lanes; + memory?: { value: string | null | undefined; at: number }; + /** The agent's handle from its latest delivery, for the tools to act with. */ + handle?: AgentHandle; + /** Where each conversation's tools default to: its latest turn's thread. */ + contexts: Map; +}; + +/** The environment that picks the agent's model and provider over the goose + * config's. */ +const env = ({ model, provider }: Config) => ({ + ...(model.trim() ? { GOOSE_MODEL: model.trim() } : {}), + ...(provider.trim() ? { GOOSE_PROVIDER: provider.trim() } : {}), +}); + +export class GooseRuntime { + private readonly agents = new Map(); + private readonly reading = new Map>(); + private readonly listeners = new Set<() => void>(); + private readonly views = new Map< + string, + ReturnType + >(); + + constructor( + private readonly host: Host, + /** The owner's connection to each agent's own community. */ + private readonly relay: (pubkey: string) => RelayData | undefined, + ) {} + + private get spawn(): Spawn { + const spawn = this.host.spawn?.bind(this.host); + if (!spawn) throw new Error("goose agents run only in the desktop app"); + return spawn; + } + + /** Brings the runtime in line with this type's agents: changed settings + * reopen idle sessions, and removed agents stop. */ + sync(agents: readonly Agent[]) { + const current = new Set(agents.map((agent) => agent.pubkey)); + for (const [pubkey, entry] of this.agents) + if (!current.has(pubkey)) { + entry.lanes.stop(); + entry.sessions.dispose(); + this.agents.delete(pubkey); + } + for (const agent of agents) { + const next = config(agent.config); + const entry = this.agents.get(agent.pubkey); + if (!entry) + this.entry(agent.pubkey, next, agent.attentionEnabled).name = + agent.name; + else if ( + JSON.stringify(entry.config) !== JSON.stringify(next) || + entry.name !== agent.name || + entry.attention !== agent.attentionEnabled + ) { + if (!agent.attentionEnabled) entry.lanes.clear(); + entry.config = next; + entry.name = agent.name; + entry.attention = agent.attentionEnabled; + entry.sessions.reconfigure(); + } + } + this.notify(); + } + + /** Conversations each agent has a session for, for its status view. */ + sessions(pubkey: string) { + let view = this.views.get(pubkey); + if (!view) { + view = this.agents.get(pubkey)?.sessions.snapshot() ?? []; + this.views.set(pubkey, view); + } + return view; + } + subscribe(listener: () => void) { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + private notify() { + this.views.clear(); + for (const listener of this.listeners) listener(); + } + + dispose() { + for (const entry of this.agents.values()) { + entry.lanes.stop(); + entry.sessions.dispose(); + } + this.agents.clear(); + } + + async run({ + trigger, + agent, + config: raw, + channelId, + team, + workingUntil, + ignored, + signal, + step, + acknowledged, + superseded, + current, + }: Delivery) { + const settings = config(raw); + // goose runs commands as you; mentions need owner or delegated authority. + // A watch is the agent's own attention: its event is observed data, + // framed as such, from anyone. + if (trigger.type === "mention") { + const snapshot = this.relay(agent.pubkey)?.snapshot(); + if ( + !(await fromOwner(trigger.event, agent.owner, { + agent: agent.pubkey, + authority: + snapshot?.status === "ready" + ? snapshot.session.workflowAuthority + : undefined, + profiles: + snapshot?.status === "ready" + ? snapshot.session.profiles + : undefined, + })) + ) { + ignored?.(); + return; + } + signal.throwIfAborted(); + } + const entry = this.entry(agent.pubkey, settings, agent.attention.enabled()); + entry.handle = agent; + entry.team = team ?? ""; + void this.memory(agent.pubkey, entry); + if (trigger.type !== "mention") { + workingUntil(this.wake(entry, agent, trigger, current)); + return; + } + const { event } = trigger; + const channel = + channelId ?? event.tags.find((tag) => tag[0] === "h")?.[1] ?? undefined; + if (!channel) return; + // `fresh` is for a session that has been shown nothing yet. + const prompt = async (fresh: boolean) => { + const turn = await this.turn(entry, agent.pubkey, event, channel, fresh); + const { channel: where } = turn; + return { + key: turn.key, + place: where.name + ? `#${where.name}` + : where.dm + ? "DM" + : where.id.slice(0, 8), + text: this.review( + entry, + turnPrompt({ ...turn, agent: agent.pubkey, label: "@mention" }), + ), + }; + }; + const { key, text, place } = await prompt(false); + entry.contexts.set( + key, + key === channel + ? { channel } + : { channel, root: threadReference(event)?.rootId ?? event.id }, + ); + const done = entry.sessions.deliver(key, text, event.created_at, { + steer: steerPrompt(text), + fresh: async () => (await prompt(true)).text, + title: sessionTitle(entry.name, place, key.split("/")[1]), + activityScope: threadReference(event)?.rootId ?? event.id, + ...(step ? { step } : {}), + ...(acknowledged ? { acknowledged } : {}), + ...(superseded ? { superseded } : {}), + }); + workingUntil(done); + this.notify(); + void done + .then((result) => { + if (!result.ok) + return this.report(agent, channel, event, result.error ?? ""); + }) + .catch((error) => console.warn("goose turn failed", error)) + .finally(() => this.notify()); + } + + /** A watch or timer wake: a one-off turn in its Interest's lane. It joins + * no conversation. The host says whether it is still `current`, asked when + * its lane starts it and again once its session is open. */ + private wake( + entry: Entry, + agent: AgentHandle, + trigger: Exclude, + current: () => boolean, + ): Promise { + const { slug } = trigger; + const id = slug.replace(/^watch\//, ""); + const lane = (trigger.type === "watch" ? trigger.watch : trigger.timer) + .interest_id; + let settle!: (result: Settled) => void; + const settled = new Promise((resolve) => { + settle = resolve; + }); + const dropped = (why: string) => { + console.warn(`goose dropped a ${trigger.type} turn: ${why}`); + settle({ ok: true }); + }; + const queued = entry.lanes.add( + lane, + async () => { + try { + if (!current()) return settle({ ok: true }); + const label = `${trigger.type}/${id}`; + const done = entry.sessions.runOnce( + label, + () => this.wakePrompt(agent, trigger, id, current), + { title: sessionTitle(entry.name, label) }, + ); + this.notify(); + const result = await done; + this.notify(); + if (!result.ok) + console.warn( + `goose ${trigger.type} turn failed: ${id}`, + result.error, + ); + settle(result); + } finally { + settle({ ok: false, error: "The turn failed" }); + } + }, + () => dropped("attention was turned off or the agent stopped"), + ); + if (queued !== "queued") + dropped( + queued === "full" + ? `too many are waiting (Interest ${lane})` + : "the agent stopped", + ); + return settled; + } + + /** The prompt for a wake as its watch now is, or undefined when it should + * no longer run. */ + private async wakePrompt( + agent: AgentHandle, + trigger: Exclude, + id: string, + stillCurrent: () => boolean, + ) { + const { attention } = agent; + if (!stillCurrent()) return undefined; + const current = (await attention.show(trigger.slug)).object; + const type = trigger.type === "watch" ? "event" : "timer"; + if (!current || current.type !== type) return undefined; + const interest = (await attention.show(`interest/${current.interest_id}`)) + .object; + const instructions = + interest?.type === "interest" + ? { instructions: interest.instructions } + : {}; + const text = + trigger.type === "watch" + ? watchPrompt({ + id, + interest: current.interest_id, + ...instructions, + event: trigger.event, + ...(trigger.classifier ? { classifier: trigger.classifier } : {}), + }) + : timerPrompt({ + id, + interest: current.interest_id, + prompt: current.type === "timer" ? current.prompt : "", + ...instructions, + spent: trigger.spent, + }); + return stillCurrent() ? `${text}\n\n${ATTENTION_REVIEW}` : undefined; + } + + /** `text` with the end-of-turn attention review, when attention is on. */ + private review(entry: Entry, text: string) { + return entry.attention ? `${text}\n\n${ATTENTION_REVIEW}` : text; + } + + /** Says in the thread that a turn failed, so nobody waits on it. */ + private report( + agent: AgentHandle, + channel: string, + event: EventData, + error: string, + ) { + return agent + .publish({ + kind: 9, + content: `⚠️ I couldn't finish that: ${error.slice(0, 500)}`, + tags: [ + ["h", channel], + ["e", threadReference(event)?.rootId ?? event.id, "", "reply"], + ["p", agent.owner], + ], + }) + .catch((failure) => + console.warn("goose could not report a failed turn", failure), + ); + } + + /** Where the event's turn goes, and what its prompt needs. */ + private async turn( + entry: Entry, + self: string, + event: EventData, + channelId: string, + fresh = false, + ) { + const snapshot = this.relay(self)?.snapshot(); + const session = snapshot?.status === "ready" ? snapshot.session : undefined; + const summary = + session?.channels.list().channels.find((item) => item.id === channelId) ?? + session?.channels.get?.(channelId); + const dm = summary?.channelType === "dm"; + const thread = threadReference(event); + const rootId = thread?.rootId ?? event.id; + const scope = entry.config.scope; + const key = + dm || scope === "channel" ? channelId : `${channelId}/${rootId}`; + const seen = fresh ? 0 : entry.sessions.seen(key); + let earlier: readonly EventData[] = []; + if (session && (thread || dm)) + try { + const events = await session.read( + thread + ? [ + { ids: [rootId], limit: 1 }, + { kinds: CHAT, "#h": [channelId], "#e": [rootId], limit: 200 }, + ] + : [{ kinds: CHAT, "#h": [channelId], limit: CONTEXT_LIMIT + 1 }], + { signal: AbortSignal.timeout(NAMES_TIMEOUT_MS * 2) }, + ); + earlier = events + .filter( + (item) => + CHAT.includes(item.kind) && + item.id !== event.id && + item.created_at <= event.created_at, + ) + .sort((a, b) => a.created_at - b.created_at); + } catch (error) { + console.warn("goose could not read the conversation", error); + } + // A session already has its own replies; a new one is shown them too. + const unseen = earlier.filter( + (item) => item.created_at > seen && (!seen || item.pubkey !== self), + ); + const pubkeys = [ + ...new Set([ + event.pubkey, + ...unseen.map((item) => item.pubkey), + ...event.tags + .filter((tag) => tag[0] === "p") + .map((tag) => tag[1] ?? ""), + ]), + ].filter(Boolean); + if (session) + await Promise.race([ + session.profiles.ensure(pubkeys, "foreground").catch(() => undefined), + new Promise((resolve) => setTimeout(resolve, NAMES_TIMEOUT_MS)), + ]); + const members = summary?.members ?? []; + return { + key, + event, + scope, + channel: { + id: channelId, + ...(summary?.name ? { name: summary.name } : {}), + dm, + }, + ...(thread ? { thread } : {}), + context: unseen, + total: earlier.length + 1, + name: (pubkey: string) => + session?.names.resolve(pubkey, undefined, members) ?? + session?.profiles.snapshot().get(pubkey)?.name ?? + undefined, + }; + } + + private entry(pubkey: string, initial: Config, attention: boolean): Entry { + const existing = this.agents.get(pubkey); + if (existing) return existing; + const entry: Entry = { + config: initial, + team: "", + attention, + lanes: new Lanes(), + sessions: undefined as unknown as AgentSessions, + contexts: new Map(), + }; + // One process serves every session, so the tools read the settings, the + // handle and the conversation's place as they are when called. + const tools: ToolServer = async (conversation, message) => { + const handle = entry.handle; + const client = handle && { + ...appClient(handle, { + spawn: this.spawn, + cwd: entry.config.workspace, + memories: () => this.memories(pubkey), + }), + remember: async (slug: string, body: string, after: number) => { + await handle.remember(slug, body, after); + if (slug === "core") delete entry.memory; + }, + }; + const context = (conversation && entry.contexts.get(conversation)) || {}; + return respond(client, context, message, { + offered: entry.attention, + ...(handle ? { api: handle.attention } : {}), + }); + }; + entry.sessions = new AgentSessions({ + spawn: (id, options) => this.spawn(id, options), + tools, + onChange: () => this.notify(), + processFingerprint: () => + JSON.stringify([ + entry.config.workspace, + entry.config.model, + entry.config.provider, + ]), + fingerprint: () => + JSON.stringify([ + entry.config.workspace, + entry.config.instructions, + entry.config.scope, + entry.team, + entry.name, + entry.attention, + ]), + launch: async () => { + const { workspace, instructions, scope } = entry.config; + const memory = await this.memory(pubkey, entry); + return { + workspace, + env: env(entry.config), + systemPrompt: systemPrompt({ + scope, + cwd: workspace, + ...(entry.name ? { agent: { name: entry.name, pubkey } } : {}), + instructions, + team: entry.team, + attention: entry.attention, + ...(memory === undefined ? {} : { memory }), + }), + }; + }, + }); + this.agents.set(pubkey, entry); + return entry; + } + + /** The agent's `core` memory, as a new session's prompt carries it: its + * text, `null` when it has none, or undefined when it could not be read. */ + private async memory(pubkey: string, entry: Entry) { + if (entry.memory && Date.now() - entry.memory.at < MEMORY_TTL_MS) + return entry.memory.value; + let value: string | null | undefined; + try { + const memories = await this.memories(pubkey); + value = memories.find((memory) => memory.slug === "core")?.body ?? null; + } catch (error) { + console.warn("goose could not read the agent's memory", error); + } + entry.memory = { value, at: Date.now() }; + return value; + } + + /** The agent's memory, read back through the owner's session: it is + * encrypted to the owner, so no agent key is needed to read it. */ + private memories(pubkey: string): Promise { + let reading = this.reading.get(pubkey); + if (!reading) { + reading = this.readMemories(pubkey).finally(() => + this.reading.delete(pubkey), + ); + this.reading.set(pubkey, reading); + } + return reading; + } + private async readMemories(pubkey: string): Promise { + const snapshot = this.relay(pubkey)?.snapshot(); + if (snapshot?.status !== "ready") throw new Error("Buzz is not connected"); + const view = snapshot.session.agentMemories.open(pubkey); + try { + const done = new Promise((resolve) => { + const settled = () => + !["idle", "loading"].includes(view.snapshot().status) && resolve(); + view.subscribe(settled); + settled(); + }); + void view.refresh(); + await Promise.race([ + done, + new Promise((_, reject) => + setTimeout( + () => reject(new Error("Memory read timed out")), + MEMORY_TIMEOUT_MS, + ), + ), + ]); + const { status, listing } = view.snapshot(); + if (status !== "ready" || !listing) + throw new Error(`Memory unavailable (${status})`); + return listing.entries; + } finally { + view.dispose(); + } + } +} diff --git a/src/bundled/goose/sessions.test.ts b/src/bundled/goose/sessions.test.ts new file mode 100644 index 0000000000..49de14c194 --- /dev/null +++ b/src/bundled/goose/sessions.test.ts @@ -0,0 +1,190 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { ToolServer } from "./goose"; +import { fakeSpawn, flush } from "./goose-testing"; +import { AgentSessions, IDLE_MS, type Launch } from "./sessions"; + +const tools: ToolServer = async (_conversation, message) => ({ + jsonrpc: "2.0", + id: message.id, + result: {}, +}); + +const setup = (initial: Partial = {}, hold = false) => { + const fake = fakeSpawn({ home: "/home/test", hold }); + let settings: Launch = { + workspace: "~/.buzz", + env: {}, + systemPrompt: "prompt", + ...initial, + }; + const sessions = new AgentSessions({ + spawn: fake.spawn, + tools, + launch: async () => settings, + processFingerprint: () => + JSON.stringify([settings.workspace, settings.env]), + fingerprint: () => + JSON.stringify([settings.workspace, settings.systemPrompt]), + }); + return { + fake, + sessions, + change: (next: Partial, reconfigure = true) => { + settings = { ...settings, ...next }; + if (reconfigure) sessions.reconfigure(); + }, + }; +}; + +afterEach(() => { + vi.useRealTimers(); +}); + +describe("AgentSessions", () => { + it("serves each conversation from its own session in one process", async () => { + const { fake, sessions } = setup(); + await sessions.deliver("a", "one", 1); + await sessions.deliver("b", "two", 2); + await sessions.deliver("a", "three", 3); + expect(fake.geese).toHaveLength(1); + const [goose] = fake.geese; + expect(goose?.opened).toHaveLength(2); + expect(goose?.turns.map((turn) => [turn.sessionId, turn.text])).toEqual([ + ["session-1", "one"], + ["session-2", "two"], + ["session-1", "three"], + ]); + expect(sessions.seen("a")).toBe(3); + }); + + it("opens sessions in the workspace as an absolute path", async () => { + const { fake, sessions } = setup(); + await sessions.deliver("a", "one", 1); + expect(fake.geese[0]?.options.cwd).toBe("/home/test/.buzz"); + expect(fake.geese[0]?.opened[0]?.cwd).toBe("/home/test/.buzz"); + }); + + it("opens one session for messages that arrive together", async () => { + const { fake, sessions } = setup(); + await Promise.all([ + sessions.deliver("a", "one", 1), + sessions.deliver("a", "two", 2), + ]); + expect(fake.geese[0]?.opened).toHaveLength(1); + }); + + it("closes idle sessions when their settings change", async () => { + const { fake, sessions, change } = setup(); + await sessions.deliver("a", "one", 1); + change({ systemPrompt: "newer" }); + await flush(30); + expect(fake.geese[0]?.closed).toEqual(["session-1"]); + expect(sessions.seen("a")).toBe(0); + }); + + it("shows a reopened conversation everything again", async () => { + const { fake, sessions, change } = setup(); + const fresh = vi.fn(async () => "everything so far"); + await sessions.deliver("a", "one", 1, { fresh }); + expect(fresh).not.toHaveBeenCalled(); + change({ systemPrompt: "newer" }, false); + await sessions.deliver("a", "two", 2, { fresh }); + const [goose] = fake.geese; + expect(fake.geese).toHaveLength(1); + expect(goose?.closed).toEqual(["session-1"]); + expect(goose?.prompts.get("session-2")).toBe("newer"); + expect(goose?.turns.at(-1)?.text).toBe("everything so far"); + }); + + it("restarts the process for a new model once nothing runs in it", async () => { + const { fake, sessions, change } = setup(); + await sessions.deliver("a", "one", 1); + const [first] = fake.geese; + if (first) first.hold = true; + const working = sessions.deliver("b", "two", 2); + await flush(30); + change({ env: { GOOSE_MODEL: "other" } }); + await flush(30); + // `a` was idle, so it closed; `b` keeps the old process while it works. + expect(first?.closed).toEqual(["session-1"]); + expect(first?.killed).toBe(false); + await sessions.deliver("c", "three", 3); + expect(fake.geese).toHaveLength(2); + expect(fake.geese[1]?.options.env).toMatchObject({ GOOSE_MODEL: "other" }); + first?.turns.at(-1)?.finish(); + await working; + await flush(30); + expect(first?.killed).toBe(true); + }); + + it("closes a quiet conversation's session", async () => { + vi.useFakeTimers(); + const { fake, sessions } = setup(); + await sessions.deliver("a", "one", 1); + await vi.advanceTimersByTimeAsync(IDLE_MS); + expect(fake.geese[0]?.closed).toEqual(["session-1"]); + expect(sessions.snapshot()).toEqual([]); + expect(sessions.seen("a")).toBe(0); + }); + + it("starts goose again after it exits", async () => { + const { fake, sessions } = setup(); + await sessions.deliver("a", "one", 1); + fake.geese[0]?.exit(1, "crashed"); + await flush(30); + await expect(sessions.deliver("a", "two", 2)).resolves.toEqual({ + ok: true, + }); + expect(fake.geese).toHaveLength(2); + }); + + it("reports why goose could not start", async () => { + const fake = fakeSpawn({ acp: false }); + const sessions = new AgentSessions({ + spawn: fake.spawn, + tools, + launch: async () => ({ workspace: "/w", env: {}, systemPrompt: "" }), + processFingerprint: () => "", + fingerprint: () => "", + }); + await expect(sessions.deliver("a", "one", 1)).resolves.toMatchObject({ + ok: false, + error: expect.stringMatching(/MCP-over-ACP/), + }); + }); + + it("runs a one-off turn in a session it then closes", async () => { + const { fake, sessions } = setup(); + await expect(sessions.runOnce("watch/x", () => "look")).resolves.toEqual({ + ok: true, + }); + const [goose] = fake.geese; + expect(goose?.turns.map((turn) => turn.text)).toEqual(["look"]); + expect(goose?.closed).toEqual(["session-1"]); + expect(goose?.killed).toBe(false); + expect(sessions.snapshot()).toEqual([]); + }); + + it("stops a one-off turn at its deadline", async () => { + vi.useFakeTimers(); + const { fake, sessions } = setup({}, true); + const running = sessions.runOnce("timer/x", () => "wait", { + deadline: 1_000, + }); + await vi.advanceTimersByTimeAsync(1_000); + const [goose] = fake.geese; + await expect(running).resolves.toMatchObject({ ok: false }); + expect(goose?.cancelled).toEqual(["session-1"]); + }); + + it("stops its processes when disposed", async () => { + const { fake, sessions } = setup(); + await sessions.deliver("a", "one", 1); + sessions.dispose(); + await flush(30); + expect(fake.geese[0]?.killed).toBe(true); + await expect(sessions.deliver("a", "two", 2)).resolves.toMatchObject({ + ok: false, + }); + }); +}); diff --git a/src/bundled/goose/sessions.ts b/src/bundled/goose/sessions.ts new file mode 100644 index 0000000000..929c5188e1 --- /dev/null +++ b/src/bundled/goose/sessions.ts @@ -0,0 +1,355 @@ +// One agent's goose sessions: one `goose-acp` process for the agent, and in +// it one ACP session per conversation (a thread, a whole DM, or a channel). +// 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. Nothing is saved: sessions live as +// long as the app does. +import { + GooseProcess, + type ActivityReceipt, + type GooseSession, + type Settled, + type Spawn, + type ToolServer, +} from "./goose"; + +/** A conversation's session closes after this long without a message. */ +export const IDLE_MS = 15 * 60_000; +/** How long a one-off turn can run before it is given up; Janet's turn + * deadline. */ +export const ONCE_DEADLINE_MS = 5 * 60_000; + +/** What a session is opened with. */ +export type Launch = Readonly<{ + /** The agent's working directory, absolute or `~/…`. */ + workspace: string; + /** Added to the process's environment: its model and provider. */ + env: Readonly>; + systemPrompt: string; +}>; + +export type SessionsOptions = Readonly<{ + spawn: Spawn; + tools: ToolServer; + /** The agent's current settings, for a new session. */ + launch(): Promise; + /** Changes when a setting the process was started with has changed. */ + processFingerprint(): string; + /** Changes when a setting a session was opened with has changed. */ + fingerprint(): string; + /** Called when a conversation's session opens or closes. */ + onChange?(): void; +}>; + +type Process = { goose: GooseProcess; cwd: string; fingerprint: string }; +type Live = { + session: GooseSession; + process: Process; + fingerprint: string; + timer?: ReturnType; +}; + +export class AgentSessions { + /** The process new sessions open in. One started with older settings keeps + * running while a session in it works. */ + private current: { fingerprint: string; ready: Promise } | undefined; + private readonly live = new Map(); + /** One-off turns running or starting, by label. */ + private readonly once = new Map(); + private onceCount = 0; + private readonly opening = new Map>(); + /** 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; + } + /** Conversations with a session, and whether each is working. */ + snapshot() { + return [ + ...[...this.live].map(([key, live]) => ({ + key, + busy: live.session.busy, + })), + ...[...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. */ + async deliver( + key: string, + text: string, + seen: number, + { + steer = text, + fresh, + title, + ...activity + }: Readonly<{ + steer?: string; + fresh?: () => Promise; + title?: string; + }> & + ActivityReceipt = {}, + ): Promise { + if (this.disposed) return { ok: false, error: "goose was stopped" }; + // A message that arrives while its conversation's session opens uses it. + const opening = this.opening.get(key); + if (opening) await opening.catch(() => undefined); + let live = this.live.get(key); + if ( + live && + (!live.session.open || (!live.session.busy && this.stale(live))) + ) { + this.drop(key, live); + live = undefined; + } + if (live) { + this.touch(key, live, seen); + return this.answer(key, live, live.session.busy ? steer : text, activity); + } + // Perhaps built for a session closed since, which had been shown the rest. + if (this.had.has(key) && fresh) text = await fresh(); + const starting = this.open(key, title); + this.opening.set(key, starting); + try { + live = await starting; + } catch (error) { + console.warn("goose could not start", error); + return { ok: false, error: `goose could not start: ${message(error)}` }; + } finally { + if (this.opening.get(key) === starting) this.opening.delete(key); + } + if (this.disposed) { + void live.process.goose.close(live.session); + return { ok: false, error: "goose was stopped" }; + } + this.adopt(key, live, seen); + return this.answer(key, live, text, activity); + } + + /** Runs one turn in a new session, then closes it. `label` names it in the + * snapshot. It gives up, and stops the turn, after `deadline` ms. `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, + { + deadline = ONCE_DEADLINE_MS, + title, + }: Readonly<{ + deadline?: number; + title?: string; + }> = {}, + ): Promise { + if (this.disposed) return { ok: false, error: "goose was stopped" }; + const key = `${label} #${++this.onceCount}`; + this.once.set(key, undefined); + this.options.onChange?.(); + let live: Live | undefined; + let timer: ReturnType | undefined; + try { + live = await this.open(key, title); + if (this.disposed) return { ok: false, error: "goose was stopped" }; + this.once.set(key, live); + const text = await prompt(); + if (text === undefined || this.disposed) return { ok: true }; + return await Promise.race([ + live.session.send(text), + new Promise((resolve) => { + timer = setTimeout( + () => + resolve({ + ok: false, + error: `The turn ran past its ${Math.round(deadline / 1000)}s deadline and was stopped`, + }), + deadline, + ); + }), + ]); + } catch (error) { + console.warn("goose could not start a one-off session", error); + return { ok: false, error: `goose could not start: ${message(error)}` }; + } finally { + clearTimeout(timer); + this.once.delete(key); + if (live) { + await live.process.goose.close(live.session); + this.retire(live.process); + } + this.options.onChange?.(); + } + } + + /** Closes idle sessions opened with old settings; busy ones close when they + * next fall idle. A process started with old settings stops once nothing + * runs in it. */ + reconfigure() { + for (const [key, live] of this.live) + if (!live.session.busy && this.stale(live)) this.drop(key, live); + void this.current?.ready + .then((process) => this.retire(process)) + .catch(() => undefined); + } + + dispose() { + this.disposed = true; + const processes = new Set(); + for (const [key, live] of this.live) { + processes.add(live.process); + this.drop(key, live); + } + for (const live of this.once.values()) + if (live) processes.add(live.process); + for (const process of processes) void process.goose.kill(); + void this.current?.ready + .then((process) => process.goose.kill()) + .catch(() => undefined); + this.current = undefined; + } + + private stale(live: Live) { + return ( + live.fingerprint !== this.options.fingerprint() || + live.process.fingerprint !== this.options.processFingerprint() + ); + } + + private async open(key: string, title?: string): Promise { + // Read before `launch` reads the settings, so it never describes newer ones. + const fingerprint = this.options.fingerprint(); + const launch = await this.options.launch(); + const process = await this.process(launch); + const session = await process.goose.open({ + cwd: process.cwd, + systemPrompt: launch.systemPrompt, + ...(title ? { title } : {}), + }); + session.conversation = key; + if (fingerprint === this.options.fingerprint() || this.disposed) + return { session, process, fingerprint }; + // Settings saved while it opened: it would work with the old ones. + void process.goose.close(session); + return this.open(key, title); + } + + /** The process for the current settings, started when there is none. */ + private process(launch: Launch): Promise { + const fingerprint = this.options.processFingerprint(); + if (this.current?.fingerprint === fingerprint) return this.current.ready; + const old = this.current; + const current = { + fingerprint, + ready: (async (): Promise => { + const cwd = await absolute(this.options.spawn, launch.workspace); + const goose = await GooseProcess.start(this.options.spawn, { + cwd, + env: launch.env, + tools: this.options.tools, + }); + const process = { goose, cwd, fingerprint }; + void goose.exited.then(() => { + if (this.current === current) this.current = undefined; + for (const [key, live] of this.live) + if (live.process === process) this.drop(key, live); + }); + return process; + })(), + }; + // A process that failed to start is started again on the next message. + current.ready.catch(() => { + if (this.current === current) this.current = undefined; + }); + this.current = current; + void old?.ready + .then((process) => this.retire(process)) + .catch(() => undefined); + return current.ready; + } + + /** Stops `process` when it has old settings and nothing runs in it. */ + private retire(process: Process) { + if ( + !this.disposed && + process.fingerprint === this.options.processFingerprint() + ) + return; + const users = [...this.live.values(), ...this.once.values()]; + if (users.some((live) => live?.process === process)) return; + if (this.current?.fingerprint === process.fingerprint) + this.current = undefined; + void process.goose.kill(); + } + + private adopt(key: string, live: Live, seen: number) { + this.had.add(key); + this.live.set(key, live); + this.touch(key, live, seen); + this.options.onChange?.(); + } + private async answer( + key: string, + live: Live, + text: string, + activity: ActivityReceipt, + ) { + const settled = await live.session.send(text, activity); + if (this.live.get(key) === live) { + this.touch(key, live); + if (!live.session.busy && this.stale(live)) this.drop(key, live); + } + return settled; + } + /** Notes what the session has been shown and restarts its idle countdown. */ + private touch(key: string, live: Live, seen?: number) { + clearTimeout(live.timer); + live.timer = setTimeout(() => { + if (live.session.busy) this.touch(key, live); + else if (this.live.get(key) === live) this.drop(key, live); + }, IDLE_MS); + if (seen !== undefined) + this.shown.set(key, Math.max(seen, this.shown.get(key) ?? 0)); + } + 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 live.process.goose.close(live.session).then(() => { + this.retire(live.process); + }); + } +} + +/** `workspace` as an absolute path, which ACP sessions need: the host + * resolves `~/…` for the process it starts there. */ +async function absolute(spawn: Spawn, workspace: string) { + if (workspace.startsWith("/")) return workspace; + let directory = ""; + const process = await spawn("workspace", { + cwd: workspace, + onStdout: (data) => { + directory += data; + }, + }); + if ((await process.exited) !== 0 || !directory.trim()) + throw new Error(`Could not find the workspace ${workspace}`); + return directory.trim(); +} + +function message(error: unknown) { + return error instanceof Error ? error.message : String(error); +} diff --git a/src/bundled/goose/tabs.tsx b/src/bundled/goose/tabs.tsx new file mode 100644 index 0000000000..b94c996715 --- /dev/null +++ b/src/bundled/goose/tabs.tsx @@ -0,0 +1,146 @@ +// goose's New agent choices and agent settings. The provider and its +// credentials come from the owner's goose config; an agent can only name a +// different model or provider from it. +import { useState } from "react"; +import { AgentInstructions } from "../../features/agents/AgentInstructions"; +import type { + AgentCreateProps, + AgentViewProps, +} from "../../features/agents/service"; +import { Button } from "../../shared/design-system/ui/Button"; +import { Field } from "../../shared/design-system/ui/Field"; +import { Input } from "../../shared/design-system/ui/Input"; +import { Select } from "../../shared/design-system/ui/Select"; +import { type Config, config as readConfig, DEFAULT_CONFIG } from "./runtime"; + +const message = (error: unknown) => + error instanceof Error ? error.message : String(error); + +const SCOPES = [ + { + label: "", + options: [ + { value: "thread", label: "One session per thread" }, + { value: "channel", label: "One session per channel" }, + ], + }, +]; + +/** The instructions a new agent starts with. */ +export function Create({ + config, + onChange, + disabled, +}: AgentCreateProps) { + const current = readConfig(config); + return ( + onChange({ ...current, instructions })} + /> + ); +} + +export function SettingsTab({ + agent, + save, + settingsHost, +}: AgentViewProps) { + const saved = readConfig(agent.config); + const [draft, setDraft] = useState(saved); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(""); + const configChanged = JSON.stringify(draft) !== JSON.stringify(saved); + const changed = configChanged || Boolean(settingsHost?.pendingSave); + const submit = async () => { + setSaving(true); + setError(""); + try { + await save({ + ...draft, + model: draft.model.trim(), + provider: draft.provider.trim(), + workspace: draft.workspace.trim() || DEFAULT_CONFIG.workspace, + }); + } catch (reason) { + setError(message(reason)); + } finally { + setSaving(false); + } + }; + return ( +
{ + event.preventDefault(); + void submit(); + }} + > + {settingsHost?.typeControl?.(configChanged)} + + setDraft({ ...draft, provider })} + /> + + + setDraft({ ...draft, model })} + /> + + + setDraft({ ...draft, workspace })} + /> + + ({ + value: item.id, + label: item.name, + description: item.configured ? "Set up" : item.description, + })), + ], + }, + ]} + onValueChange={(provider) => onChange({ provider, model: "" })} + /> + {listing.status === "error" && ( +
+ +
+ )} + {provider && ( + + )} + onChange({ ...value, model })} + /> +
+ + {tested && tested !== "testing" && ( +

+ {tested.message} +

+ )} +
+ + ); + } + + /** A provider's key and other fields, saved into goose's config. One that + * signs in through goose is set up there. */ + function ProviderKeys({ + provider, + disabled, + onSaved, + }: Readonly<{ provider: Provider; disabled: boolean; onSaved(): void }>) { + const [saved, reread] = useRead( + () => setup.saved(provider.id), + provider.id, + ); + const [draft, setDraft] = useState>({}); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(""); + const [changing, setChanging] = useState(false); + const entered = Object.fromEntries( + Object.entries(draft).filter(([, text]) => text.trim()), + ); + if (provider.signsIn && !provider.configured) + return ( +

+ {provider.name} signs in through goose. Run{" "} + goose configure to set it up, then choose it here. +

+ ); + if (provider.signsIn || !provider.keys.length) return null; + if (provider.configured && !changing) + return ( +
+

+ {provider.name} is set up in goose. +

+ +
+ ); + return ( + <> + {provider.keys.map((key) => { + const isSet = saved.value?.has(key.name); + return ( + + + setDraft({ ...draft, [key.name]: text }) + } + /> + + ); + })} +
+ + {error && ( +

+ {error} +

+ )} +
+ + ); + } + + /** One session's turns, or a place's across its sessions, read on open and + * on Refresh. A refresh keeps the last result until the new one arrives. */ + function TranscriptView({ + agent, + row, + working, + }: { + agent: AgentViewProps["agent"]; + row: Row; + working?: string | undefined; + }) { + const workspace = readConfig(agent.config).workspace; + const [read, setRead] = useState< + | { status: "ready"; transcript: GooseTranscript } + | { status: "missing" } + | { status: "error"; error: string } + >(); + const [loading, setLoading] = useState(true); + const [attempt, setAttempt] = useState(0); + // A turn ending rereads, so its last steps show without a Refresh. + // biome-ignore lint/correctness/useExhaustiveDependencies: attempt is the explicit Refresh. + useEffect(() => { + const abort = new AbortController(); + setLoading(true); + history + .transcript( + { + agent: agent.pubkey, + workspace, + sessionId: row.sessionId, + ...(row.channelId ? { place: placeOf(row.key) } : {}), + ...(working ? { working } : {}), + }, + abort.signal, + ) + .then( + (transcript) => + transcript + ? ({ status: "ready", transcript } as const) + : ({ status: "missing" } as const), + (reason: unknown) => + ({ status: "error", error: message(reason) }) as const, + ) + .then((next) => { + if (abort.signal.aborted) return; + setRead(next); + setLoading(false); + }); + return () => abort.abort(); + }, [row, workspace, working, attempt]); + return ( +
+
+ +
+ {!read ? ( +

+ Reading the goose session… +

+ ) : read.status === "missing" ? ( +

+ goose has no session for this conversation on this computer. It may + have been deleted, or it ran in a different workspace. +

+ ) : read.status === "error" ? ( +

+ Could not read the goose session: {read.error} +

+ ) : ( + (working ? "working" : undefined)} + before={ + read.transcript.more && ( +

+ Showing the latest {TURN_LIMIT} turns. +

+ ) + } + /> + )} +
+ ); + } + + /** The session of the conversation the profile was opened from, and the + * agent's other recent sessions, each with its transcript. */ + function Conversations({ + agent, + conversation, + }: { + agent: AgentViewProps["agent"]; + conversation?: AgentViewProps["conversation"]; + }) { + const workspace = readConfig(agent.config).workspace; + const live = useSyncExternalStore( + (listener) => runtime.subscribe(listener), + () => runtime.sessions(agent.pubkey), + ); + const running = live.map( + (session): Row => ({ + key: session.key, + sessionId: session.sessionId ?? "", + title: "", + ...placeOf(session.key), + }), + ); + const { + session: here, + searching, + error: lookupError, + } = useOpenedSession( + running.filter((row) => row.sessionId), + conversation, + async (place, signal) => { + const found = await history.find( + agent.pubkey, + workspace, + place, + signal, + ); + return ( + found && { + ...found, + ...place, + key: place.root + ? `${place.channelId}/${place.root}` + : place.channelId, + } + ); + }, + ); + const [recent, reload] = useRead(() => { + const abort = new AbortController(); + return history.recent(agent.name, workspace, abort.signal); + }, `${agent.name}\u0000${workspace}`); + const busy = (sessionId: string) => + live.find((session) => session.sessionId === sessionId)?.busy; + const status = (sessionId: string) => { + const working = busy(sessionId); + return working === undefined ? undefined : working ? "Working" : "Idle"; + }; + const label = (row: Row) => + row.title + ? titled(agent.name, row.title) + : `${row.channelId.slice(0, 8)} · ${row.root ? `thread ${row.root.slice(0, 8)}` : "channel"}`; + const { open, show, back } = useOpenTranscript(conversation, label); + const rows = (recent.value ?? []) + .filter((row) => row.sessionId !== here?.sessionId) + .slice(0, RECENT_LIMIT) + .map((row): Row => ({ ...row, key: row.sessionId, channelId: "" })); + return ( +
+ {conversation && ( + here && show(here)} + /> + )} +
+
+

Recent sessions

+ +
+ {recent.status === "error" ? ( +

+ Could not list goose's sessions: {recent.error} +

+ ) : rows.length ? ( +
    + {rows.map((row) => ( +
  • + + {label(row)} + + {[ + status(row.sessionId), + row.at && new Date(row.at).toLocaleString(), + ] + .filter(Boolean) + .join(" · ")} + + + +
    • + ))} +
+ ) : recent.status === "ready" ? ( +

+ None. A conversation's session starts when the agent is mentioned. +

+ ) : null} +
+ {open && ( + + + + )} +
+ ); + } + + /** The provider, model and instructions a new agent starts with. */ + /** Whether New agent's provider dialog is open. The footer menu opens it + * and closes before it shows, so Create, which stays, renders it. */ + const choosing = { + open: false, + listeners: new Set<() => void>(), + set(open: boolean) { + choosing.open = open; + for (const listener of choosing.listeners) listener(); + }, + subscribe(listener: () => void) { + choosing.listeners.add(listener); + return () => { + choosing.listeners.delete(listener); + }; + }, }; - return ( - { - event.preventDefault(); - void submit(); - }} - > - {settingsHost?.typeControl?.(configChanged)} - - setDraft({ ...draft, provider })} + + /** The instructions a new agent starts with, and the provider dialog its + * footer menu opens. */ + function Create({ config, onChange, disabled }: AgentCreateProps) { + const current = readConfig(config); + const open = useSyncExternalStore(choosing.subscribe, () => choosing.open); + return ( + <> + onChange({ ...current, instructions })} /> - - choosing.set(false)} + onChange={(choice) => { + onChange({ ...current, ...choice }); + choosing.set(false); + }} + /> + )} + + ); + } + + /** A provider, its key and its model, chosen apart from the card. */ + function ChooseDialog({ + value, + onChange, + onClose, + }: Readonly<{ + value: Choice; + onChange(choice: Choice): void; + onClose(): void; + }>) { + const [draft, setDraft] = useState({ + provider: value.provider, + model: value.model, + }); + return ( + { + if (!open) onClose(); + }} + title="Provider and model" + description="goose's own config is the default. Keys are saved in goose, for every agent and the goose CLI." + finalFocus={false} + actions={ + <> + + + + } > - setDraft({ ...draft, model })} - /> - - + + + + ); + } + + /** goose's own default and the agent's choice, in New agent's type menu. */ + function CreateModelMenu({ + config, + onChange, + disabled, + active = true, + }: AgentCreateModelMenuProps) { + const current = readConfig(config); + const [listing] = useRead(() => setup.providers(), "providers"); + const defaults = listing.value?.defaults; + const named = (id: string | undefined) => + listing.value?.providers.find((item) => item.id === id)?.name ?? id; + const fallback = [named(defaults?.provider), defaults?.model] + .filter(Boolean) + .join(" · "); + const chosen = current.provider || current.model; + const icon = - setDraft({ ...draft, workspace })} + /> + + ({ + value: item.id, + label: item.name, + ...(item.configured ? { description: "Set up" } : {}), + })), + ], + }, + ]} + onValueChange={(provider) => onChange({ provider, model: "" })} + /> + {listing.status === "error" && ( +
+ +
+ )} + {provider && ( + + )} + onChange({ ...value, model })} + /> +
+ + {tested && tested !== "testing" && ( +

+ {tested.message} +

+ )} +
+ + ); + } + + /** A provider's API key, saved with Pi's login. One that signs in without + * a key is set up with the `pi` command. */ + function ProviderKey({ + provider, + disabled, + onSaved, + }: Readonly<{ provider: Provider; disabled: boolean; onSaved(): void }>) { + const [key, setKey] = useState(""); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(""); + const [changing, setChanging] = useState(false); + if (provider.configured && (!provider.key || !changing)) + return ( +
+

+ {provider.name} is set up in Pi. +

+ {provider.key && ( + + )} +
+ ); + if (!provider.key) + return ( +

+ {provider.name} signs in through Pi. Run pi, then{" "} + /login, to set it up, then choose it here. +

+ ); + return ( + <> + + + +
+ + {error && ( +

+ {error} +

+ )} +
+ + ); + } + + /** Whether New agent's provider dialog is open. The footer menu opens it + * and closes before it shows, so Create, which stays, renders it. */ + const choosing = { + open: false, + listeners: new Set<() => void>(), + set(open: boolean) { + choosing.open = open; + for (const listener of choosing.listeners) listener(); + }, + subscribe(listener: () => void) { + choosing.listeners.add(listener); + return () => { + choosing.listeners.delete(listener); + }; + }, + }; + + /** The instructions a new agent starts with, and the provider dialog its + * footer menu opens. */ + function Create({ config, onChange, disabled }: AgentCreateProps) { + const current = readConfig(config); + const open = useSyncExternalStore(choosing.subscribe, () => choosing.open); + return ( + <> + onChange({ ...current, instructions })} + /> + {open && ( + choosing.set(false)} + onChange={(choice) => { + onChange({ ...current, ...choice }); + choosing.set(false); + }} + /> + )} + + ); + } + + /** A provider, its key and its model, chosen apart from the card. */ + function ChooseDialog({ + value, + onChange, + onClose, + }: Readonly<{ + value: Choice; + onChange(choice: Choice): void; + onClose(): void; + }>) { + const [draft, setDraft] = useState({ + provider: value.provider, + model: value.model, + }); + const [listing] = useRead(() => setup.providers(), "providers"); + return ( + { + if (!open) onClose(); + }} + title="Provider and model" + description="Pi's own settings are the default. Keys are saved in Pi, for every Pi agent and the pi command." + finalFocus={false} + actions={ + <> + + + + } + > +
+ +
+ + ); + } + + /** Pi's own default and the agent's choice, in New agent's type menu. */ + function CreateModelMenu({ + config, + onChange, + disabled, + active = true, + }: AgentCreateModelMenuProps) { + const current = readConfig(config); + const [listing] = useRead(() => setup.providers(), "providers"); + const defaults = listing.value?.defaults; + const named = (id: string | undefined) => + listing.value?.providers.find((item) => item.id === id)?.name ?? id; + const fallback = [named(defaults?.provider), defaults?.model] + .filter(Boolean) + .join(" · "); + const chosen = current.provider || current.model; + const unknown = unknownProvider({ provider: "", model: "" }, listing.value); + const icon =