From 9d4223801f44f1bbcdac4e4d0d70a8b0c241c511 Mon Sep 17 00:00:00 2001 From: Joah Gerstenberg Date: Wed, 21 Oct 2026 22:05:52 -0700 Subject: [PATCH] feat(amp): add an Amp agent type A bundled desktop plugin, off by default, that runs the local `amp` CLI as a Buzz agent harness. Each conversation is one kept Amp thread continued with `amp threads continue --execute --stream-json`; the Buzz tools are a per-turn loopback MCP server. While continuing a thread, the Buzz tools refuse `initialize` until Amp reports the thread open, since Amp connects twice and the host server admits one session. AGNTOPS-664 Co-authored-by: Hertz <6dc34705e5eaffade9f039764bf81185f09d3e096c0dedc2fed3b7ba7a549620@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Joah Gerstenberg --- crates/plugin-manager/src/lib.rs | 2 + crates/plugin-manager/tests/management.rs | 7 +- src/app/pages.integration.test.mjs | 1 + src/bundled/amp/README.md | 58 ++ src/bundled/amp/amp-testing.ts | 109 ++++ src/bundled/amp/amp.live.test.ts | 187 ++++++ src/bundled/amp/amp.test.ts | 226 ++++++++ src/bundled/amp/amp.ts | 255 ++++++++ src/bundled/amp/index.ts | 99 ++++ src/bundled/amp/manifest.json | 15 + src/bundled/amp/runtime.test.ts | 232 ++++++++ src/bundled/amp/runtime.ts | 670 ++++++++++++++++++++++ src/bundled/amp/tabs.tsx | 167 ++++++ src/bundled/index.ts | 7 + 14 files changed, 2034 insertions(+), 1 deletion(-) create mode 100644 src/bundled/amp/README.md create mode 100644 src/bundled/amp/amp-testing.ts create mode 100644 src/bundled/amp/amp.live.test.ts create mode 100644 src/bundled/amp/amp.test.ts create mode 100644 src/bundled/amp/amp.ts create mode 100644 src/bundled/amp/index.ts create mode 100644 src/bundled/amp/manifest.json create mode 100644 src/bundled/amp/runtime.test.ts create mode 100644 src/bundled/amp/runtime.ts create mode 100644 src/bundled/amp/tabs.tsx diff --git a/crates/plugin-manager/src/lib.rs b/crates/plugin-manager/src/lib.rs index f5e8c4cda1..cc7d30dc2d 100644 --- a/crates/plugin-manager/src/lib.rs +++ b/crates/plugin-manager/src/lib.rs @@ -288,6 +288,8 @@ pub fn bundled_manifests() -> Vec { .expect("goose manifest"), serde_json::from_str(include_str!("../../../src/bundled/pi/manifest.json")) .expect("pi manifest"), + serde_json::from_str(include_str!("../../../src/bundled/amp/manifest.json")) + .expect("amp 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")) diff --git a/crates/plugin-manager/tests/management.rs b/crates/plugin-manager/tests/management.rs index d41fe360bb..516e35af54 100644 --- a/crates/plugin-manager/tests/management.rs +++ b/crates/plugin-manager/tests/management.rs @@ -1134,7 +1134,12 @@ fn bundled_defaults_preserve_saved_choices_and_only_channels_is_required() { let id = plugin.manifest.id.as_str(); let default = !matches!( id, - "buzz.bestie" | "buzz.pi" | "buzz.todos" | "buzz.usage" | "buzz.relay-notes" + "buzz.amp" + | "buzz.bestie" + | "buzz.pi" + | "buzz.todos" + | "buzz.usage" + | "buzz.relay-notes" ); assert_eq!(plugin.enabled, default, "{id}"); if id == "buzz.channels" { diff --git a/src/app/pages.integration.test.mjs b/src/app/pages.integration.test.mjs index 21e7e1907c..e6ec65eb51 100644 --- a/src/app/pages.integration.test.mjs +++ b/src/app/pages.integration.test.mjs @@ -71,6 +71,7 @@ test("the app runtime exposes ready bundled pages and removes them on disable", assert.equal( plugin.enabledByDefault, ![ + "buzz.amp", "buzz.bestie", "buzz.pi", "buzz.todos", diff --git a/src/bundled/amp/README.md b/src/bundled/amp/README.md new file mode 100644 index 0000000000..9bf4d1dbdb --- /dev/null +++ b/src/bundled/amp/README.md @@ -0,0 +1,58 @@ +# Amp + +Create agents that run Amp on your computer and answer you in Buzz. Each conversation continues its own Amp thread. + +## Getting started + +1. Install [Amp](https://ampcode.com) and run `amp login` once. +2. Turn on **Amp** in **Settings** → **Plugins**. +3. Choose **New agent** → **Create a new agent** and pick **Amp** as the **Harness**. +4. Mention your agent in a conversation to give it work. + +--- + +# Amp for Agents + +A desktop plugin, off until turned on in **Settings → Plugins**. It runs the +`amp` CLI found on this computer's PATH (on macOS, `~/.local/bin` too), signed +in as `amp login` left it. Agents owns the identity, admission, configuration, +and signed Buzz actions; this plugin owns Amp work after delivery. + +- Each conversation is one Amp thread, saved on this device: a Buzz thread, + or the whole channel when **Sessions** says so, and always a DM or Solo + chat. Each turn runs `amp threads continue --execute --stream-json` + once, in the agent's **Workspace** (default `~/.buzz`), and its turns run + one at a time. A thread Amp can no longer continue is started again. + Threads are kept unarchived and titled `Agent · #channel`, so they also show + in Amp's thread list. +- The Buzz tools are a loopback MCP server the app answers for each turn + (`--mcp-config`), signed by the agent's native-held key; the `amp` process + holds no Buzz key. Amp reaches them with its own `tool_search`. Continuing a + thread, Amp connects once as it starts and again once the thread loads, but + the server admits one session, so the Buzz tools refuse `initialize` until + Amp reports the thread open. Amp retries and connects. +- Amp has no system prompt option, so Buzz's prompt opens each new thread, and + opens the next turn again after it changes. +- Only the owner's events, or ones the owner delegated, start work. Amp runs + with `--dangerously-allow-all`, like Claude's `bypassPermissions`. Watches + and timers run as one-off Amp threads, which Amp archives. +- **Mode** picks Amp's agent mode (`--mode`); the composer's model picker + offers the modes for one turn. +- **Stop** on a turn's activity stops its `amp` process, and posts nothing. +- A turn that answered in text and never called `send` has its answer posted in + the thread. A turn that fails posts `⚠️ I couldn't finish that: …` and tags + the owner. +- `BUZZ_MANAGED_AGENT=1` is set for `amp`, so Amp plugins that mirror threads + to Buzz, such as amp-buzz, stay out of the way. + +Not yet supported: steering a running turn (a newer mention waits for it), an +Activity transcript, usage, compact and new-session controls, resuming turns a +reload stopped, and Windows. + +## Checks + +```sh +bin/pnpm exec vitest run src/bundled/amp +# Against the real `amp`, using Amp credits: +AMP_LIVE=1 bin/pnpm exec vitest run src/bundled/amp/amp.live.test.ts +``` diff --git a/src/bundled/amp/amp-testing.ts b/src/bundled/amp/amp-testing.ts new file mode 100644 index 0000000000..4af66c78d9 --- /dev/null +++ b/src/bundled/amp/amp-testing.ts @@ -0,0 +1,109 @@ +// A fake desktop host for Amp tests: each `amp` it starts runs `script` with +// its arguments, its prompt and the turn's MCP handler, and prints what that +// returns. +import type { + HostMcpServer, + HostProcess, + HostProcessOptions, + McpHandler, +} from "../../features/host/service"; +import type { AmpHost } from "./amp"; + +export type Script = ( + run: Readonly<{ + args: readonly string[]; + prompt: string; + mcp: McpHandler; + killed: Promise; + }>, +) => Promise>; + +export type Started = { + args: readonly string[]; + cwd?: string; + env?: HostProcessOptions["env"]; + prompt?: string; + killed: boolean; +}; + +export const line = (value: unknown) => JSON.stringify(value); +export const init = (thread: string) => + line({ type: "system", subtype: "init", session_id: thread }); +export const result = (thread: string, text: string, isError = false) => + line({ + type: "result", + subtype: isError ? "error" : "success", + is_error: isError, + result: text, + session_id: thread, + }); + +export function fakeAmp(script: Script) { + const started: Started[] = []; + const servers: { closed: boolean }[] = []; + let handler: McpHandler = async () => undefined; + const host: AmpHost = { + serve: async (next) => { + handler = next; + const server = { closed: false }; + servers.push(server); + return { + url: "http://127.0.0.1:1/mcp", + headers: [{ name: "Authorization", value: "Bearer token" }], + close: async () => { + server.closed = true; + }, + closed: new Promise(() => {}), + } satisfies HostMcpServer; + }, + spawn: async (_id, options = {}) => { + const run: Started = { + args: options.args ?? [], + ...(options.cwd ? { cwd: options.cwd } : {}), + ...(options.env ? { env: options.env } : {}), + killed: false, + }; + started.push(run); + const mcp = handler; + let kill = () => {}; + const killed = new Promise((resolve) => { + kill = resolve; + }); + let input: (prompt: string) => void = () => {}; + const prompt = new Promise((resolve) => { + input = resolve; + }); + const exited = prompt.then(async (text) => { + const output: Awaited> = await Promise.race([ + script({ args: run.args, prompt: text, mcp, killed }), + killed.then(() => ({ code: 143 })), + ]); + for (const item of output.stdout ?? []) options.onStdout?.(`${item}\n`); + if (output.stderr) options.onStderr?.(output.stderr); + return output.code ?? 0; + }); + return { + write: async () => {}, + end: async (data = "") => { + run.prompt = data; + input(data); + }, + kill: async () => { + run.killed = true; + kill(); + }, + exited, + } satisfies HostProcess; + }, + }; + return { host, started, servers }; +} + +/** A `tools/call` of the Buzz tool `name`, as Amp sends one. */ +export const call = (mcp: McpHandler, name: string, id = 1) => + mcp({ + jsonrpc: "2.0", + id, + method: "tools/call", + params: { name, arguments: {} }, + }); diff --git a/src/bundled/amp/amp.live.test.ts b/src/bundled/amp/amp.live.test.ts new file mode 100644 index 0000000000..76deade3c7 --- /dev/null +++ b/src/bundled/amp/amp.live.test.ts @@ -0,0 +1,187 @@ +// @vitest-environment node +// Drives the real `amp` on PATH, signed in as `amp login` left it: run with +// AMP_LIVE=1. The Buzz tools are listed as the app lists them, from a loopback +// HTTP MCP server that follows the app's native one; their calls are recorded, +// not sent to a relay. Each run uses Amp credits. +import { spawn as spawnChild } from "node:child_process"; +import { randomBytes } from "node:crypto"; +import { mkdtempSync } from "node:fs"; +import { createServer } from "node:http"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { describe, expect, it } from "vitest"; +import { respond } from "../../buzz-mcp/rpc"; +import { answerMcp } from "../../features/host/service"; +import { systemPrompt } from "../claude-code/prompt"; +import { type AmpHost, runTurn, type ToolServer } from "./amp"; + +const LIVE = process.env.AMP_LIVE === "1"; + +const spawn: AmpHost["spawn"] = async (_id, options = {}) => { + 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("amp", [...(options.args ?? [])], { + cwd: options.cwd, + env, + stdio: ["pipe", "pipe", "pipe"], + }); + child.stdout + .setEncoding("utf8") + .on("data", (data) => options.onStdout?.(data)); + child.stderr + .setEncoding("utf8") + .on("data", (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, + }; +}; + +/** The HTTP requests the MCP servers received, by method and status. */ +const hits: { method: string; rpc: unknown; status: number }[] = []; + +/** What the app's native listener does: a token per server, 404 for a wrong + * one, 405 for `GET`, and a JSON reply or 202 for each `POST /mcp`. The first + * `initialize` answered with a result claims the server's one session, which + * later requests name; another after that stops the server. */ +const serve: AmpHost["serve"] = async (handler) => { + const token = randomBytes(32).toString("base64url"); + const session = randomBytes(32).toString("base64url"); + let claim: "unclaimed" | "claiming" | "claimed" = "unclaimed"; + let stopped = false; + const server = createServer((request, response) => { + let rpc: unknown; + const end = (status: number, body?: string) => { + hits.push({ method: request.method ?? "", rpc, status }); + response.writeHead(status, { + ...(claim === "claimed" && (status === 200 || status === 202) + ? { "Mcp-Session-Id": session } + : {}), + ...(body ? { "Content-Type": "application/json" } : {}), + }); + response.end(body); + }; + if ( + request.url?.split("?")[0] !== "/mcp" || + request.headers.authorization !== `Bearer ${token}` + ) + return end(404); + if (request.method !== "POST") return end(405); + const named = request.headers["mcp-session-id"]; + if (named !== undefined && (named !== session || claim !== "claimed")) + return end(404); + let body = ""; + request.setEncoding("utf8").on("data", (data) => (body += data)); + request.on("end", async () => { + try { + rpc = (JSON.parse(body) as { method?: unknown }).method; + } catch {} + if (stopped) return end(404); + if (named === undefined) { + if (rpc !== "initialize") return end(404); + if (claim !== "unclaimed") { + stopped = true; + return end(404); + } + claim = "claiming"; + } + const reply = await answerMcp(handler, body); + if (claim === "claiming") + claim = + (reply as { result?: unknown } | undefined)?.result === undefined + ? "unclaimed" + : "claimed"; + end(reply === undefined ? 202 : 200, JSON.stringify(reply)); + }); + }); + await new Promise((resolve) => + server.listen(0, "127.0.0.1", () => resolve()), + ); + const address = server.address(); + const port = typeof address === "object" && address ? address.port : 0; + const closed = new Promise((resolve) => + server.on("close", () => resolve()), + ); + return { + url: `http://127.0.0.1:${port}/mcp`, + headers: [{ name: "Authorization", value: `Bearer ${token}` }], + close: async () => { + server.closeAllConnections(); + server.close(); + }, + closed, + }; +}; + +type Send = { name?: string; arguments?: { text?: string } }; + +describe.skipIf(!LIVE)("amp, live", () => { + it("answers with the Buzz tools, then continues its thread", { + timeout: 300_000, + }, async () => { + const sends: Send[] = []; + const tools: ToolServer = async (message) => { + if (message.method === "tools/call") { + sends.push(message.params as Send); + return { + jsonrpc: "2.0", + id: message.id, + result: { content: [{ type: "text", text: "sent 1234" }] }, + }; + } + return respond(undefined, {}, message); + }; + const cwd = mkdtempSync(join(tmpdir(), "buzz-amp-")); + const system = systemPrompt({ scope: "thread", cwd }); + const first = await runTurn( + { spawn, serve }, + { + cwd, + prompt: `\n${system}\n\n\nRemember the word "heron". Reply in Buzz with send, audience "everyone", text "ready", final true.`, + keep: true, + title: "Buzz live test", + mode: "low", + tools, + }, + ); + expect(first).toMatchObject({ ok: true }); + expect(first.text).toBeUndefined(); + expect(sends.at(-1)).toMatchObject({ + name: "send", + arguments: { text: "ready" }, + }); + // Amp opens one MCP session and never claims the server twice. + expect( + hits.filter((hit) => hit.rpc === "initialize" && hit.status === 200), + ).toHaveLength(1); + expect(hits.filter((hit) => hit.status === 404)).toEqual([]); + + const second = await runTurn( + { spawn, serve }, + { + cwd, + prompt: + 'Which word did I ask you to remember? Reply with send, audience "everyone", the word as text, final true.', + ...(first.thread ? { thread: first.thread } : {}), + mode: "low", + tools, + }, + ); + expect(second).toMatchObject({ ok: true, thread: first.thread }); + expect(sends.at(-1)?.arguments?.text?.toLowerCase()).toContain("heron"); + }); +}); diff --git a/src/bundled/amp/amp.test.ts b/src/bundled/amp/amp.test.ts new file mode 100644 index 0000000000..5800addb57 --- /dev/null +++ b/src/bundled/amp/amp.test.ts @@ -0,0 +1,226 @@ +import { describe, expect, it, vi } from "vitest"; +import { ampArgs, runTurn, type ToolServer } from "./amp"; +import { call, fakeAmp, init, line, result } from "./amp-testing"; + +const server = { + url: "http://127.0.0.1:1/mcp", + headers: [{ name: "Authorization", value: "Bearer token" }], +}; +const ok: ToolServer = async ({ id }) => ({ + jsonrpc: "2.0", + id, + result: { content: [{ type: "text", text: "sent" }] }, +}); + +describe("ampArgs", () => { + it("starts a kept, titled thread with the Buzz tools", () => { + const args = ampArgs({ keep: true, title: "Ada · #general" }, server); + expect(args).toEqual( + expect.arrayContaining([ + "--execute", + "--stream-json", + "--dangerously-allow-all", + "--no-archive-after-execute", + "--title", + "Ada · #general", + ]), + ); + expect(args[0]).toBe("--execute"); + const config = args[args.indexOf("--mcp-config") + 1] ?? ""; + expect(JSON.parse(config)).toEqual({ + buzz: { + url: server.url, + headers: { Authorization: "Bearer token" }, + }, + }); + }); + + it("continues a saved thread in a mode, without new-thread options", () => { + const args = ampArgs( + { thread: "T-1", keep: true, title: "x", mode: "high" }, + server, + ); + expect(args.slice(0, 3)).toEqual(["threads", "continue", "T-1"]); + expect(args).toEqual(expect.arrayContaining(["--mode", "high"])); + expect(args).not.toContain("--title"); + expect(args).not.toContain("--no-archive-after-execute"); + }); + + it("lets Amp archive a one-off thread", () => { + expect(ampArgs({}, server)).not.toContain("--no-archive-after-execute"); + }); +}); + +describe("runTurn", () => { + it("returns the thread and what Amp wrote when it answered nowhere", async () => { + const fake = fakeAmp(async () => ({ + stdout: [init("T-1"), result("T-1", " pong ")], + })); + const outcome = await runTurn(fake.host, { + cwd: "~/.buzz", + prompt: "ping", + tools: ok, + }); + expect(outcome).toEqual({ ok: true, thread: "T-1", text: "pong" }); + expect(fake.started[0]).toMatchObject({ + cwd: "~/.buzz", + prompt: "ping", + env: { BUZZ_MANAGED_AGENT: "1" }, + }); + expect(fake.servers[0]?.closed).toBe(true); + }); + + it("posts nothing more once a Buzz tool answered", async () => { + const step = vi.fn(); + const fake = fakeAmp(async ({ mcp }) => { + // A notification gets no answer. + expect( + await mcp({ jsonrpc: "2.0", method: "notifications/initialized" }), + ).toBeUndefined(); + await call(mcp, "send"); + return { stdout: [init("T-1"), result("T-1", "Sent it.")] }; + }); + const outcome = await runTurn(fake.host, { + cwd: "~/.buzz", + prompt: "ping", + tools: ok, + step, + }); + expect(outcome).toEqual({ ok: true, thread: "T-1" }); + expect(step).toHaveBeenCalledWith("send"); + }); + + it("still posts the text when a send failed", async () => { + const fake = fakeAmp(async ({ mcp }) => { + await call(mcp, "send"); + return { stdout: [init("T-1"), result("T-1", "answer")] }; + }); + const outcome = await runTurn(fake.host, { + cwd: "~/.buzz", + prompt: "ping", + tools: async ({ id }) => ({ + jsonrpc: "2.0", + id, + result: { content: [], isError: true }, + }), + }); + expect(outcome.text).toBe("answer"); + }); + + it("shows Amp's tool calls as steps, without its MCP plumbing", async () => { + const step = vi.fn(); + const fake = fakeAmp(async () => ({ + stdout: [ + init("T-1"), + line({ + type: "assistant", + message: { + content: [ + { type: "tool_use", name: "tool_search", input: {} }, + { + type: "tool_use", + name: "edit_file", + input: { path: "/a/b.ts" }, + }, + { type: "tool_use", name: "shell_command", input: { cmd: "ls" } }, + ], + }, + }), + result("T-1", ""), + ], + })); + await runTurn(fake.host, { + cwd: "~", + prompt: "x", + tools: ok, + step, + }); + expect(step.mock.calls).toEqual([ + ["edit_file", "b.ts"], + ["shell_command", "ls"], + ]); + }); + + it("refuses the Buzz tools' first connection while a thread is still loading", async () => { + const initialize = { jsonrpc: "2.0", id: 1, method: "initialize" }; + const replies: unknown[] = []; + const fake = fakeAmp(async ({ mcp }) => { + replies.push(await mcp(initialize)); + return { stdout: [init("T-1"), result("T-1", "")] }; + }); + await runTurn(fake.host, { + cwd: "~", + prompt: "x", + thread: "T-1", + tools: ok, + }); + await runTurn(fake.host, { cwd: "~", prompt: "x", tools: ok }); + expect(replies[0]).toMatchObject({ id: 1, error: { code: -32000 } }); + expect(replies[1]).toMatchObject({ id: 1, result: expect.anything() }); + }); + + it("fails with Amp's error, and says when the thread never opened", async () => { + const fake = fakeAmp(async () => ({ + stderr: "Error: This thread is archived and cannot be continued.\n", + })); + expect( + await runTurn(fake.host, { + cwd: "~", + prompt: "x", + thread: "T-1", + tools: ok, + }), + ).toEqual({ + ok: false, + thread: "T-1", + error: "This thread is archived and cannot be continued.", + unopened: true, + }); + }); + + it("fails with an error result", async () => { + const fake = fakeAmp(async () => ({ + stdout: [init("T-1"), result("T-1", "Out of credits", true)], + })); + expect( + await runTurn(fake.host, { cwd: "~", prompt: "x", tools: ok }), + ).toEqual({ ok: false, thread: "T-1", error: "Out of credits" }); + }); + + it("says how to get Amp when it cannot start", async () => { + const fake = fakeAmp(async () => ({})); + fake.host = { + ...fake.host, + spawn: async () => { + throw new Error("No such file or directory"); + }, + }; + const outcome = await runTurn(fake.host, { + cwd: "~", + prompt: "x", + tools: ok, + }); + expect(outcome.ok).toBe(false); + expect(outcome.error).toContain("amp login"); + expect(fake.servers[0]?.closed).toBe(true); + }); + + it("stops the process when stopped, and reports nothing", async () => { + const controller = new AbortController(); + const fake = fakeAmp(async ({ killed }) => { + await killed; + return {}; + }); + const outcome = runTurn(fake.host, { + cwd: "~", + prompt: "x", + thread: "T-1", + tools: ok, + signal: controller.signal, + }); + await vi.waitFor(() => expect(fake.started).toHaveLength(1)); + controller.abort(); + expect(await outcome).toEqual({ ok: true, stopped: true, thread: "T-1" }); + expect(fake.started[0]?.killed).toBe(true); + }); +}); diff --git a/src/bundled/amp/amp.ts b/src/bundled/amp/amp.ts new file mode 100644 index 0000000000..95f0480188 --- /dev/null +++ b/src/bundled/amp/amp.ts @@ -0,0 +1,255 @@ +// One Amp turn: an `amp` process in execute mode that runs one message in an +// Amp thread and exits. Amp prints Claude Code's stream-json format. The Buzz +// tools are a loopback HTTP MCP server the app answers for that turn, so the +// process holds no Buzz key. +import type { + HostMcpServer, + HostProcess, + HostProcessOptions, + McpHandler, +} from "../../features/host/service"; + +export type Spawn = ( + id: string, + options?: HostProcessOptions, +) => Promise; +export type ServeMcp = (handler: McpHandler) => Promise; +export type AmpHost = Readonly<{ spawn: Spawn; serve: ServeMcp }>; + +/** Answers one MCP message of the turn's Buzz tools with its JSON-RPC response. */ +export type ToolServer = ( + message: Readonly<{ id: string | number; method: string; params?: unknown }>, +) => Promise; + +export const TOOL_SERVER = "buzz"; +/** The Buzz tools that answer where people see it. */ +const ANSWERS = new Set(["send", "edit", "react"]); +/** Amp's own plumbing for reaching MCP tools, not steps worth showing. */ +const PLUMBING = new Set(["tool_search", "code_exec"]); +const STEP_KEYS = ["path", "cmd", "command", "pattern", "query", "url"]; +const WRITTEN_LIMIT = 20_000; +const STDERR_LIMIT = 4_000; + +export const ENV: Readonly> = { + // Amp plugins that mirror sessions to Buzz, such as amp-buzz, stay out of + // a turn Buzz already holds. + BUZZ_MANAGED_AGENT: "1", +}; + +export const MODES = ["low", "medium", "high", "ultra"] as const; + +export type TurnLaunch = Readonly<{ + /** An absolute directory or one under home written `~/…`. */ + cwd: string; + prompt: string; + /** The Amp thread to continue; absent starts one. */ + thread?: string; + /** A new thread is kept for later turns; otherwise Amp archives it. */ + keep?: boolean; + title?: string; + /** An Amp agent mode; absent is Amp's default. */ + mode?: string; + tools: ToolServer; + signal?: AbortSignal; + step?(tool: string, target?: string): void; +}>; + +export type TurnOutcome = Readonly<{ + ok: boolean; + /** The Amp thread the turn ran in, once Amp named it. */ + thread?: string; + error?: string; + /** What Amp wrote, when it answered nowhere in Buzz. */ + text?: string; + /** The owner stopped it. */ + stopped?: true; + /** It failed before Amp opened the thread, such as one that cannot be + * continued. */ + unopened?: true; +}>; + +type Wire = { + id?: string | number | null; + method?: string; + params?: unknown; + error?: unknown; + result?: { isError?: unknown }; +}; +type Line = Record & { + type?: string; + subtype?: string; + session_id?: unknown; + is_error?: unknown; + result?: unknown; + message?: { content?: unknown }; +}; + +export function ampArgs( + launch: Pick, + server: Pick, +) { + const headers = Object.fromEntries( + server.headers.map(({ name, value }) => [name, value]), + ); + return [ + ...(launch.thread ? ["threads", "continue", launch.thread] : []), + "--execute", + "--stream-json", + "--dangerously-allow-all", + "--no-ide", + "--no-notifications", + "--no-color", + "--mcp-config", + JSON.stringify({ [TOOL_SERVER]: { url: server.url, headers } }), + ...(launch.mode ? ["--mode", launch.mode] : []), + ...(launch.thread + ? [] + : [ + ...(launch.keep ? ["--no-archive-after-execute"] : []), + ...(launch.title ? ["--title", launch.title] : []), + ]), + ]; +} + +/** Runs one turn and settles once `amp` exits. Never rejects. */ +export async function runTurn( + host: AmpHost, + launch: TurnLaunch, +): Promise { + let answered = false; + // Continuing a thread, Amp connects to the Buzz tools as it starts, drops + // that connection as the thread loads, and connects again. The host's + // server admits one session and stops at a second `initialize`, so until + // Amp reports the thread open, an `initialize` is refused; Amp retries. + let opened = false; + let server: HostMcpServer; + try { + server = await host.serve(async (message) => { + const { id, method = "", params } = (message ?? {}) as Wire; + if (id === undefined || id === null) return undefined; + if (method === "initialize" && launch.thread && !opened) + return { + jsonrpc: "2.0", + id, + error: { code: -32000, message: "Not ready; connect again" }, + }; + const response = (await launch.tools({ id, method, params })) as Wire; + const name = String((params as { name?: unknown } | undefined)?.name); + if ( + method === "tools/call" && + !response.error && + !response.result?.isError + ) { + launch.step?.(name); + if (ANSWERS.has(name)) answered = true; + } + return response; + }); + } catch (error) { + return { ok: false, error: message(error) }; + } + let thread = launch.thread; + let result: Line | undefined; + let stderr = ""; + let buffer = ""; + const read = (line: string) => { + let parsed: Line; + try { + parsed = JSON.parse(line); + } catch { + return; + } + if (typeof parsed.session_id === "string") { + thread = parsed.session_id; + opened = true; + } + if (parsed.type === "result") result = parsed; + if (parsed.type === "assistant") stepsOf(parsed, launch.step); + }; + try { + let process: HostProcess; + try { + process = await host.spawn("amp", { + args: ampArgs(launch, server), + cwd: launch.cwd, + env: ENV, + onStdout: (data) => { + buffer += data; + const lines = buffer.split("\n"); + buffer = lines.pop() ?? ""; + for (const line of lines) if (line.trim()) read(line); + }, + onStderr: (data) => { + stderr = (stderr + data).slice(-STDERR_LIMIT); + }, + }); + } catch (error) { + return { + ok: false, + error: `Could not start Amp (${message(error)}). Install it from https://ampcode.com, then run \`amp login\`.`, + }; + } + const stop = () => void process.kill().catch(() => undefined); + if (launch.signal?.aborted) stop(); + launch.signal?.addEventListener("abort", stop, { once: true }); + try { + await process.end(launch.prompt); + const code = await process.exited; + if (buffer.trim()) read(buffer); + if (launch.signal?.aborted) + return { ok: true, stopped: true, ...(thread ? { thread } : {}) }; + const done = result && result.is_error !== true; + if (done) { + const text = + typeof result?.result === "string" ? result.result.trim() : ""; + return { + ok: true, + ...(thread ? { thread } : {}), + ...(!answered && text ? { text: text.slice(0, WRITTEN_LIMIT) } : {}), + }; + } + const error = + (typeof result?.result === "string" && result.result.trim()) || + stderr + .trim() + .split("\n") + .at(-1) + ?.replace(/^Error:\s*/, "") || + `Amp exited (${code ?? "signal"})`; + return { + ok: false, + error, + ...(thread ? { thread } : {}), + ...(!opened ? { unopened: true as const } : {}), + }; + } finally { + launch.signal?.removeEventListener("abort", stop); + } + } finally { + void server.close().catch(() => undefined); + } +} + +function stepsOf(line: Line, step: TurnLaunch["step"]) { + if (!step) return; + const content = line.message?.content; + if (!Array.isArray(content)) return; + for (const block of content as Record[]) { + if (block.type !== "tool_use" || typeof block.name !== "string") continue; + if (PLUMBING.has(block.name)) continue; + const input = (block.input ?? {}) as Record; + const target = STEP_KEYS.map((key) => input[key]).find( + (value): value is string => typeof value === "string" && !!value, + ); + if (target) step(block.name, shortTarget(target)); + else step(block.name); + } +} + +function shortTarget(value: string) { + return value.startsWith("/") ? (value.split("/").at(-1) ?? value) : value; +} + +function message(error: unknown) { + return error instanceof Error ? error.message : String(error); +} diff --git a/src/bundled/amp/index.ts b/src/bundled/amp/index.ts new file mode 100644 index 0000000000..a843a40aef --- /dev/null +++ b/src/bundled/amp/index.ts @@ -0,0 +1,99 @@ +// Amp agents: an agent type whose agents are Amp on this computer. Each +// conversation the agent is in is one Amp thread; Amp answers with the +// in-process Buzz tools, served to it over loopback HTTP MCP. +import { isTauri } from "@tauri-apps/api/core"; +import { repositoriesContext } from "../../features/communities/repositories"; +import type { PluginModule } from "../../plugins/api"; +import { MODES } from "./amp"; +import { AmpRuntime, type Config, config, DEFAULT_CONFIG } from "./runtime"; +import { Create, modeLabel, SettingsTab } from "./tabs"; + +export const TYPE_ID = "amp"; +/** A turn's run returns once it is queued, 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 `amp`. + if (!isTauri() || !/Mac|Linux/i.test(navigator.platform)) return; + const { agents, host } = ctx; + const typeKey = `${ctx.pluginOwner?.id ?? "buzz.amp"}/${TYPE_ID}`; + const runtime = new AmpRuntime( + host, + agents.relay, + (pubkey, signal) => agents.core(pubkey, signal), + localStorage, + () => agents.tools(), + repositoriesContext, + ); + if (host.spawn) { + const sync = () => { + const snapshot = agents.snapshot(); + if (snapshot.status === "loading") return; + runtime.sync(snapshot.running.filter((agent) => agent.type === typeKey)); + }; + const unsubscribe = agents.subscribe(sync); + sync(); + ctx.effect(() => () => { + unsubscribe(); + runtime.dispose(); + }); + } + + agents.register({ + id: TYPE_ID, + title: "Amp", + description: + "Amp on this computer. Each thread it is mentioned in is one continuing Amp thread.", + 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) => modeLabel(config(agent.config).mode), + sessions: { + list: (pubkey) => { + const busy = new Set(runtime.sessions(pubkey).map(({ key }) => key)); + return runtime.conversations(pubkey).map(({ key, at }) => { + const [channelId = key, root] = key.split("/"); + return { + channelId, + ...(root && { root }), + at, + ...(busy.has(key) && { status: "Working" }), + }; + }); + }, + subscribe: (listener) => runtime.subscribe(listener), + }, + tabs: [ + { + id: "settings", + title: "Settings", + role: "settings", + splitSettings: true, + component: SettingsTab, + }, + ], + models: (agent) => ({ + current: config(agent.config).mode, + models: MODES.map((mode) => ({ + id: mode, + label: modeLabel(mode), + efforts: [], + })), + }), + run: (delivery) => runtime.run(delivery), + runOnce: (turn) => runtime.runOnce(turn), + timeoutMs: HANDOVER_MS, + }); +}; diff --git a/src/bundled/amp/manifest.json b/src/bundled/amp/manifest.json new file mode 100644 index 0000000000..8bf19b7e51 --- /dev/null +++ b/src/bundled/amp/manifest.json @@ -0,0 +1,15 @@ +{ + "id": "buzz.amp", + "name": "Amp", + "apiVersion": 1, + "description": "Run local Amp agents in Buzz.", + "discovery": { + "category": "agents", + "tags": ["code", "coding-agents"] + }, + "host": { + "processes": [{ "id": "amp", "program": "amp" }], + "mcp": true, + "files": true + } +} diff --git a/src/bundled/amp/runtime.test.ts b/src/bundled/amp/runtime.test.ts new file mode 100644 index 0000000000..face10b37e --- /dev/null +++ b/src/bundled/amp/runtime.test.ts @@ -0,0 +1,232 @@ +import { describe, expect, it, vi } from "vitest"; +import type { Delivery } from "../../features/agents/service"; +import type { Host } from "../../features/host/service"; +import type { EventData, RelayEvent } from "../../features/relay/events"; +import type { RelayData } from "../../features/relay/service"; +import { fakeAmp, init, result, type Script } from "./amp-testing"; +import { AmpRuntime, DEFAULT_CONFIG } from "./runtime"; + +const self = "c".repeat(64); +const alice = "a".repeat(64); +const bob = "b".repeat(64); +const channel = "chan-1"; +const message = ( + id: string, + content: string, + extra: Partial = {}, +): RelayEvent => + ({ + id: id.repeat(64).slice(0, 64), + pubkey: alice, + kind: 9, + created_at: 100 + Number(id), + content, + tags: [ + ["h", channel], + ["p", self], + ], + sig: "", + ...extra, + }) as RelayEvent; +const reply = (id: string, content: string, root: string) => + message(id, content, { + tags: [ + ["h", channel], + ["e", root.repeat(64).slice(0, 64), "", "reply"], + ["p", self], + ], + }); + +function setup(script: Script) { + const fake = fakeAmp(script); + const host = { + spawn: fake.host.spawn, + serveMcp: fake.host.serve, + readFile: vi.fn(), + request: vi.fn(), + } as unknown as Host; + const session = { + read: vi.fn(async () => []), + channels: { + list: () => ({ + channels: [ + { id: channel, name: "general", channelType: "stream", members: [] }, + ], + }), + }, + names: { resolve: (pubkey: string) => (pubkey === alice ? "Alice" : "") }, + profiles: { ensure: vi.fn(async () => {}), snapshot: () => new Map() }, + channelWrites: { startingCanvas: () => undefined }, + canvas: { available: false, read: vi.fn(async () => null) }, + }; + const relay = { + snapshot: () => ({ + status: "ready", + session, + viewer: alice, + scope: `https://amp.example:${alice}`, + }), + } as unknown as RelayData; + const data = new Map(); + const storage = { + getItem: (key: string) => data.get(key) ?? null, + setItem: (key: string, value: string) => void data.set(key, value), + } as Storage; + const runtime = new AmpRuntime( + host, + () => relay, + async () => null, + storage, + ); + const published: { kind: number; content: string; tags: string[][] }[] = []; + const agent = { + pubkey: self, + name: "Ampy", + owner: alice, + publish: vi.fn(async (template: (typeof published)[number]) => { + published.push(template); + return template; + }), + }; + const working: PromiseLike[] = []; + const deliver = async (event: RelayEvent, extra: Partial = {}) => { + const ignored = vi.fn(); + await runtime.run({ + trigger: { type: "mention", event }, + channelId: channel, + agent, + config: DEFAULT_CONFIG, + signal: new AbortController().signal, + workingUntil: (done: PromiseLike) => void working.push(done), + ignored, + ...extra, + } as unknown as Delivery); + return { ignored, done: working.at(-1) }; + }; + return { fake, runtime, agent, published, deliver, storage }; +} + +describe("AmpRuntime", () => { + it("ignores a mention from someone other than its owner", async () => { + const { deliver, fake } = setup(async () => ({})); + const { ignored } = await deliver(message("1", "hi", { pubkey: bob })); + expect(ignored).toHaveBeenCalled(); + expect(fake.started).toHaveLength(0); + }); + + it("opens a thread with Buzz's prompt, then continues it in the same Buzz thread", async () => { + const { deliver, fake, published } = setup(async ({ args }) => { + const thread = args[0] === "threads" ? (args[2] ?? "") : "T-1"; + return { stdout: [init(thread), result(thread, "pong")] }; + }); + const first = message("1", "@Ampy ping"); + await (await deliver(first)).done; + expect(fake.started[0]?.args).not.toContain("continue"); + expect(fake.started[0]?.args).toContain("--no-archive-after-execute"); + expect(fake.started[0]?.prompt).toContain(""); + expect(fake.started[0]?.prompt).toContain("Content: @Ampy ping"); + expect(fake.started[0]?.cwd).toBe("~/.buzz"); + // Amp answered in text, so it is posted in the thread. + expect(published.at(-1)).toEqual({ + kind: 9, + content: "pong", + tags: [ + ["h", channel], + ["e", first.id, "", "reply"], + ], + }); + await (await deliver(reply("2", "@Ampy again", "1"))).done; + expect(fake.started[1]?.args.slice(0, 3)).toEqual([ + "threads", + "continue", + "T-1", + ]); + expect(fake.started[1]?.prompt).not.toContain("buzz-system-prompt"); + // Another Buzz thread is another Amp thread. + await (await deliver(message("3", "@Ampy elsewhere"))).done; + expect(fake.started[2]?.args).not.toContain("continue"); + }); + + it("starts the thread again when Amp cannot continue it", async () => { + let calls = 0; + const { deliver, fake, published } = setup(async () => { + calls++; + if (calls === 2) + return { stderr: "Error: This thread is archived.\n", code: 0 }; + const thread = calls === 1 ? "T-1" : "T-2"; + return { stdout: [init(thread), result(thread, `answer ${calls}`)] }; + }); + await (await deliver(message("1", "@Ampy one"))).done; + await (await deliver(reply("2", "@Ampy two", "1"))).done; + expect(fake.started).toHaveLength(3); + expect(fake.started[2]?.args).not.toContain("continue"); + expect(fake.started[2]?.prompt).toContain(""); + expect(published.at(-1)?.content).toBe("answer 3"); + await (await deliver(reply("3", "@Ampy three", "1"))).done; + expect(fake.started[3]?.args.slice(0, 3)).toEqual([ + "threads", + "continue", + "T-2", + ]); + }); + + it("says in the thread that a turn failed, tagging the owner", async () => { + const { deliver, published } = setup(async () => ({ + stdout: [init("T-1"), result("T-1", "Out of credits", true)], + })); + await (await deliver(message("1", "@Ampy hi"))).done; + expect(published.at(-1)?.content).toBe( + "⚠️ I couldn't finish that: Out of credits", + ); + expect(published.at(-1)?.tags).toContainEqual(["p", alice]); + }); + + it("runs a conversation's turns one at a time, and Stop ends one", async () => { + const releases: (() => void)[] = []; + const { deliver, fake, runtime, published } = setup( + async ({ args, killed }) => { + await Promise.race([ + new Promise((resolve) => releases.push(resolve)), + killed, + ]); + const thread = args[0] === "threads" ? (args[2] ?? "") : "T-1"; + return { stdout: [init(thread), result(thread, "done")] }; + }, + ); + let stop: (() => Promise) | undefined; + const first = await deliver(message("1", "@Ampy one"), { + interruptible: (next: () => Promise) => { + stop = next; + }, + } as Partial); + const second = await deliver(reply("2", "@Ampy two", "1")); + await vi.waitFor(() => expect(fake.started).toHaveLength(1)); + expect(runtime.sessions(self)).toEqual([ + { key: `${channel}/${"1".repeat(64)}`, busy: true }, + ]); + await stop?.(); + await first.done; + // A stopped turn posts nothing; the next starts after it. + expect(published).toHaveLength(0); + await vi.waitFor(() => expect(fake.started).toHaveLength(2)); + releases.at(-1)?.(); + await second.done; + expect(published.at(-1)?.content).toBe("done"); + }); + + it("runs a standalone turn in a one-off thread", async () => { + const { runtime, fake, agent } = setup(async () => ({ + stdout: [init("T-9"), result("T-9", "checked")], + })); + const outcome = await runtime.runOnce({ + delivery: { agent, channelId: channel, config: DEFAULT_CONFIG }, + key: "timer/x #1", + signal: new AbortController().signal, + prompt: async () => "Check the build", + } as unknown as Parameters[0]); + expect(outcome).toEqual({ ok: true }); + expect(fake.started[0]?.args).not.toContain("--no-archive-after-execute"); + expect(fake.started[0]?.prompt).toContain("Check the build"); + expect(runtime.conversations(self)).toEqual([]); + }); +}); diff --git a/src/bundled/amp/runtime.ts b/src/bundled/amp/runtime.ts new file mode 100644 index 0000000000..7aa4fc266b --- /dev/null +++ b/src/bundled/amp/runtime.ts @@ -0,0 +1,670 @@ +// Turns Agents deliveries into Amp turns. Each conversation an agent is in is +// one Amp thread, saved on this device; each turn runs one `amp` process in +// it, one at a time per conversation. A watch or timer wake runs as a one-off +// thread in its Interest's lane. The run returns once the turn is queued, so +// one agent works in several conversations at once. +import type { Context } from "../../buzz-mcp/client"; +import { + MemoryReads, + READ_TIMEOUT_MS, +} from "../../features/agent-harness/core-memory"; +import { respond } from "../../buzz-mcp/rpc"; +import { appClient } from "../../buzz-mcp/app-client"; +import { + type AddedTool, + sessionHint, + turnNote, +} from "../../buzz-mcp/session-tools"; +import { + fromOwner, + type Agent, + type AgentHandle, + type Agents, + type Delivery, + type StandaloneTurn, + type TurnResult, +} from "../../features/agents/service"; +import { ownerReads } from "../../features/agents/reads"; +import { soloChat } from "../../features/agents/solo-chat"; +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 { + type PluginTrigger, + wakeAnchor, + wakeLands, +} from "../../features/agent-harness/plugin-wake"; +import { + CONTEXT_LIMIT, + type Scope, + systemPrompt, + turnPrompt, +} from "../claude-code/prompt"; +import { sessionTitle, standalonePrompt } from "../goose/runtime"; +import { ownerCanvas, readProject } from "../../features/agent-harness/project"; +import { type AmpHost, runTurn, type ToolServer } from "./amp"; + +export type Config = Readonly<{ + /** An Amp agent mode; empty uses Amp's default. */ + mode: string; + instructions: string; + /** The agent's working directory, absolute or `~/…`. */ + workspace: string; + /** Whether each thread is its own Amp thread, or the whole channel is one. */ + scope: Scope; +}>; +export const DEFAULT_CONFIG: Config = Object.freeze({ + mode: "", + instructions: "", + workspace: "~/.buzz", + scope: "thread", +}); +export const config = (value: unknown): Config => ({ + ...DEFAULT_CONFIG, + ...(value && typeof value === "object" ? (value as Partial) : {}), +}); + +const NAMES_TIMEOUT_MS = 2_000; +const CHAT = [9, 40002]; +/** Saved conversations per agent; the least recently used are forgotten. */ +const SAVED_LIMIT = 200; +const STORE = "buzz.amp.conversations.v1"; + +/** One conversation's Amp thread: when it last ran, the newest message it was + * shown, and a hash of the system prompt it was last given. */ +export type SavedConversation = Readonly<{ + thread: string; + at: number; + seen: number; + prompt: string; +}>; + +/** FNV-1a, so a saved conversation holds no copy of its prompt. */ +function hash(text: string) { + let value = 0x811c9dc5; + for (let i = 0; i < text.length; i++) + value = Math.imul(value ^ text.charCodeAt(i), 0x01000193); + return (value >>> 0).toString(36); +} + +type Entry = { + config: Config; + /** The team sections of the latest run. */ + team: string; + name?: string; + /** The agent's handle from its latest delivery, for the tools to act with. */ + handle?: AgentHandle; +}; +type Running = { controller: AbortController }; + +const section = (tag: string, body: string) => + `<${tag}>\n${body.trim()}\n`; + +export class AmpRuntime { + private readonly agents = new Map(); + /** Turns running now, by agent and conversation. */ + private readonly running = new Map(); + /** Each conversation's latest queued turn, which the next waits for. */ + private readonly queues = new Map>(); + private readonly memoryReads = new MemoryReads(); + private readonly listeners = new Set<() => void>(); + + constructor( + private readonly host: Host, + /** The owner's connection to each agent's own community. */ + private readonly relay: (pubkey: string) => RelayData | undefined, + /** The agent's `core` memory as the host keeps it (`Agents.core`). */ + private readonly core: Agents["core"], + private readonly storage: Storage = globalThis.localStorage, + /** The tools plugins add to every session now. */ + private readonly added: () => readonly AddedTool[] = () => [], + /** Desktop-local repository instructions. */ + private readonly repositories: (scope: string | undefined) => string = () => + "", + ) {} + + private get amp(): AmpHost { + const { host } = this; + if (!host.spawn || !host.serveMcp) + throw new Error("Amp agents run only in the desktop app"); + return { + spawn: host.spawn.bind(host), + serve: host.serveMcp.bind(host), + }; + } + + /** Brings the runtime in line with this type's agents; removed agents' turns + * stop. Settings are read as each turn starts. */ + sync(agents: readonly Agent[]) { + const current = new Set(agents.map((agent) => agent.pubkey)); + for (const pubkey of this.agents.keys()) + if (!current.has(pubkey)) { + this.agents.delete(pubkey); + for (const [key, turn] of this.running) + if (key.startsWith(`${pubkey}/`)) turn.controller.abort(); + } + for (const agent of agents) { + const entry = this.entry(agent.pubkey, config(agent.config)); + entry.config = config(agent.config); + entry.name = agent.name; + } + this.notify(); + } + + /** Conversations with a turn running now. */ + sessions(pubkey: string) { + return [...this.running.keys()] + .filter((key) => key.startsWith(`${pubkey}/`)) + .map((key) => ({ key: key.slice(pubkey.length + 1), busy: true })); + } + /** The agent's saved conversations, most recently used first. */ + conversations(pubkey: string) { + return this.saved(pubkey).map(([key, { at }]) => ({ key, at })); + } + subscribe(listener: () => void) { + this.listeners.add(listener); + return () => { + this.listeners.delete(listener); + }; + } + private notify() { + for (const listener of this.listeners) listener(); + } + + dispose() { + for (const turn of this.running.values()) turn.controller.abort(); + this.running.clear(); + this.agents.clear(); + } + + async run({ + trigger, + agent, + config: raw, + channelId, + team, + model, + workingUntil, + ignored, + signal, + step, + acknowledged, + interruptible, + }: Delivery) { + // Watches and timers come as standalone turns (`runOnce`). + if (trigger.type !== "mention" && trigger.type !== "plugin") + return ignored?.(); + // A plugin's wake is the owner's installed code, not a message to check. + const wake = trigger.type === "plugin" ? trigger : undefined; + // Amp runs commands as you; mentions need owner or delegated authority. + 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, config(raw)); + entry.handle = agent; + entry.team = team ?? ""; + const event = + trigger.type === "plugin" ? wakeAnchor(trigger) : trigger.event; + const channel = + channelId ?? event.tags.find((tag) => tag[0] === "h")?.[1] ?? undefined; + if (!channel) return; + const topLevel = await soloChat( + () => this.relay(agent.pubkey), + agent.owner, + channel, + signal, + ); + if (wake && !wakeLands(wake, topLevel)) return ignored?.(); + signal.throwIfAborted(); + const key = await this.key(entry, agent.pubkey, event, channel, topLevel); + const conversation = `${agent.pubkey}/${key}`; + const controller = new AbortController(); + const done = (this.queues.get(conversation) ?? Promise.resolve()) + .catch(() => undefined) + .then(async () => { + if ( + controller.signal.aborted || + this.agents.get(agent.pubkey) !== entry + ) + return; + this.running.set(conversation, { controller }); + this.notify(); + acknowledged?.(); + try { + await this.converse(entry, agent, { + key, + event, + channel, + topLevel, + signal: controller.signal, + ...(model ? { mode: model } : {}), + ...(wake ? { wake } : {}), + ...(step ? { step } : {}), + }); + } finally { + if (this.running.get(conversation)?.controller === controller) + this.running.delete(conversation); + this.notify(); + } + }); + this.queues.set(conversation, done); + void done + .finally(() => { + if (this.queues.get(conversation) === done) + this.queues.delete(conversation); + }) + .catch((error) => console.warn("Amp turn failed", error)); + workingUntil(done); + interruptible?.(async () => controller.abort()); + } + + /** One turn in a conversation's Amp thread, started anew when it cannot be + * continued. */ + private async converse( + entry: Entry, + agent: AgentHandle, + turn: Readonly<{ + key: string; + event: EventData; + channel: string; + topLevel: boolean; + signal: AbortSignal; + mode?: string; + wake?: PluginTrigger; + step?: (tool: string, target?: string) => void; + }>, + ) { + const { key, event, channel, topLevel } = turn; + const pubkey = agent.pubkey; + const root = threadReference(event)?.rootId ?? event.id; + const noteRoot = key === channel ? undefined : root; + const notes = agent.notes?.(channel, noteRoot) ?? []; + const system = await this.systemPrompt(entry, pubkey); + const fingerprint = hash(system); + const context: Context = topLevel + ? { channel, topLevel } + : key === channel + ? { channel } + : { channel, root }; + const attempt = async (saved: SavedConversation | undefined) => { + const input = await this.turn( + entry, + pubkey, + event, + channel, + topLevel, + saved?.seen ?? 0, + ); + const body = [ + turnPrompt({ + ...input, + agent: pubkey, + label: "@mention", + ...(turn.wake ? { wake: turn.wake } : {}), + }), + ...notes, + turnNote(agent), + ] + .filter(Boolean) + .join("\n\n"); + const prompt = !saved + ? `${section("buzz-system-prompt", system)}\n\n${body}` + : saved.prompt !== fingerprint + ? `${section("buzz-system-prompt-updated", `These replace your earlier Buzz instructions.\n\n${system}`)}\n\n${body}` + : body; + const where = input.channel.name + ? `#${input.channel.name}` + : input.channel.dm + ? "DM" + : channel.slice(0, 8); + return runTurn(this.amp, { + cwd: entry.config.workspace, + prompt, + ...(saved ? { thread: saved.thread } : {}), + keep: true, + title: sessionTitle(entry.name, where, key.split("/")[1]), + ...(turn.mode || entry.config.mode + ? { mode: turn.mode || entry.config.mode } + : {}), + tools: this.tools(pubkey, context), + signal: turn.signal, + ...(turn.step ? { step: turn.step } : {}), + }); + }; + const saved = this.saved(pubkey).find(([saved]) => saved === key)?.[1]; + let outcome = await attempt(saved); + if (!outcome.ok && outcome.unopened && saved && !turn.signal.aborted) + outcome = await attempt(undefined); + if (outcome.thread) + this.save(pubkey, key, { + thread: outcome.thread, + at: Date.now(), + seen: event.created_at, + prompt: fingerprint, + }); + if (outcome.stopped) return; + if (!outcome.ok) { + if (outcome.unopened) agent.restoreNotes?.(channel, notes, noteRoot); + return this.post( + agent, + channel, + event, + `⚠️ I couldn't finish that: ${(outcome.error ?? "Amp failed").slice(0, 500)}`, + topLevel, + [["p", agent.owner]], + ); + } + // An answer Amp wrote without `send` is posted for it, so the thread is + // not left waiting. + if (outcome.text) + return this.post(agent, channel, event, outcome.text, topLevel); + } + + /** Runs a standalone turn, such as a watch's or a timer's, in a one-off + * Amp thread that joins no conversation. */ + async runOnce({ + delivery, + key, + signal, + prompt, + }: StandaloneTurn): Promise { + const { agent, team } = delivery; + const entry = this.entry(agent.pubkey, config(delivery.config)); + entry.handle = agent; + entry.team = team ?? ""; + const text = await prompt(); + if (text === undefined) return { ok: true }; + const system = await this.systemPrompt(entry, agent.pubkey); + const running = `${agent.pubkey}/${key}`; + const controller = new AbortController(); + const stop = () => controller.abort(); + signal.addEventListener("abort", stop, { once: true }); + this.running.set(running, { controller }); + this.notify(); + try { + const outcome = await runTurn(this.amp, { + cwd: entry.config.workspace, + prompt: `${section("buzz-system-prompt", system)}\n\n${standalonePrompt(agent.pubkey, text)}`, + title: sessionTitle(entry.name, key), + ...(entry.config.mode ? { mode: entry.config.mode } : {}), + tools: this.tools(agent.pubkey, {}), + signal: controller.signal, + }); + return outcome.ok + ? { ok: true } + : { ok: false, error: outcome.error ?? "Amp failed" }; + } finally { + signal.removeEventListener("abort", stop); + if (this.running.get(running)?.controller === controller) + this.running.delete(running); + this.notify(); + } + } + + /** The agent's Buzz tools for one turn, acting where `context` says. */ + private tools(pubkey: string, context: Context): ToolServer { + return async (message) => { + const handle = this.agents.get(pubkey)?.handle; + const session = this.relay(pubkey)?.snapshot().session; + const workspace = this.agents.get(pubkey)?.config.workspace; + const client = + handle && + appClient(handle, { + relayAuthor: session?.relayAuthor, + audienceAuthorityCommunity: session?.audienceAuthorityCommunity, + signingAuthority: session?.workflowAuthority, + read: async (path) => { + if (!this.host.readFile) + throw new Error("Amp agents run only in the desktop app"); + return this.host.readFile(path, { + ...(workspace ? { cwd: workspace } : {}), + }); + }, + memories: (slug) => this.memories(pubkey, slug), + remembered: () => this.forget(pubkey), + }); + const added = this.added(); + return respond(client, context, message, added, { + instructions: sessionHint(added), + }); + }; + } + + private async systemPrompt(entry: Entry, pubkey: string) { + const { workspace, instructions, scope } = entry.config; + const memory = await this.core(pubkey); + return systemPrompt({ + scope, + cwd: workspace, + repositories: this.repositories(this.relay(pubkey)?.snapshot().scope), + ...(entry.name ? { agent: { name: entry.name, pubkey } } : {}), + instructions, + team: entry.team, + ...(memory === undefined ? {} : { memory }), + }); + } + + /** Replies as the agent in the event's thread, or at a Solo chat's top + * level. */ + private post( + agent: AgentHandle, + channel: string, + event: EventData, + content: string, + topLevel: boolean, + tags: string[][] = [], + ) { + return agent + .publish({ + kind: 9, + content, + tags: [ + ["h", channel], + ...(topLevel + ? [] + : [["e", threadReference(event)?.rootId ?? event.id, "", "reply"]]), + ...tags, + ], + }) + .then(() => undefined) + .catch((failure) => console.warn("Amp could not post", failure)); + } + + /** The conversation the event belongs to: a Solo chat, a DM and a + * channel-scoped agent's channel are one, otherwise each thread is. */ + private async key( + entry: Entry, + self: string, + event: EventData, + channelId: string, + topLevel: boolean, + ) { + const session = this.session(self); + const summary = + session?.channels.list().channels.find((item) => item.id === channelId) ?? + session?.channels.get?.(channelId); + const dm = summary?.channelType === "dm"; + const rootId = threadReference(event)?.rootId ?? event.id; + return topLevel || dm || entry.config.scope === "channel" + ? channelId + : `${channelId}/${rootId}`; + } + + private session(self: string) { + const snapshot = this.relay(self)?.snapshot(); + return snapshot?.status === "ready" ? snapshot.session : undefined; + } + + /** What the event's turn prompt needs, with the conversation's messages + * newer than `seen` that the thread has not been shown. */ + private async turn( + entry: Entry, + self: string, + event: EventData, + channelId: string, + topLevel: boolean, + seen: number, + ) { + const session = this.session(self); + 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 key = await this.key(entry, self, event, channelId, topLevel); + const scope: Scope = key === channelId ? "channel" : "thread"; + const owner = entry.handle?.owner; + const reading = topLevel + ? readProject( + session && owner + ? () => + ownerCanvas( + session.canvas, + owner, + channelId, + session.channelWrites.startingCanvas(channelId), + ) + : undefined, + ) + : Promise.resolve(undefined); + let earlier: readonly EventData[] = []; + if (session && (thread || dm || topLevel)) + try { + const events = await session.read( + thread && !topLevel + ? [ + { 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("Amp could not read the conversation", error); + } + // A thread 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 ?? []; + const project = (await reading)?.section; + return { + event, + scope, + topLevel, + 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, + ...(project ? { project } : {}), + }; + } + + private entry(pubkey: string, initial: Config): Entry { + let entry = this.agents.get(pubkey); + if (!entry) { + entry = { config: initial, team: "" }; + this.agents.set(pubkey, entry); + } + return entry; + } + + private readAll(): Record> { + try { + const value = JSON.parse(this.storage.getItem(STORE) ?? "{}"); + return value && typeof value === "object" ? value : {}; + } catch { + return {}; + } + } + /** Saved conversations, most recently used first. */ + private saved(pubkey: string): [string, SavedConversation][] { + return Object.entries(this.readAll()[pubkey] ?? {}) + .filter( + ([, saved]) => + typeof saved?.thread === "string" && typeof saved.at === "number", + ) + .sort(([, a], [, b]) => b.at - a.at); + } + private save(pubkey: string, key: string, saved: SavedConversation) { + const all = this.readAll(); + all[pubkey] = Object.fromEntries( + [ + [key, saved] as [string, SavedConversation], + ...this.saved(pubkey).filter(([other]) => other !== key), + ].slice(0, SAVED_LIMIT), + ); + try { + this.storage.setItem(STORE, JSON.stringify(all)); + } catch (error) { + console.warn("Amp conversations were not saved", error); + } + } + + /** The agent's memory, or only entry `slug`, as the owner's reads of its + * community give it: it is encrypted to the owner, so no agent key is + * needed. */ + private memories(pubkey: string, slug?: string) { + const reads = ownerReads(this.relay(pubkey), pubkey); + if (!reads) return Promise.reject(new Error("Buzz is not connected")); + return this.memoryReads.read( + reads, + pubkey, + AbortSignal.timeout(READ_TIMEOUT_MS), + slug, + ); + } + private forget(pubkey: string) { + const reads = ownerReads(this.relay(pubkey), pubkey); + if (reads) this.memoryReads.forget(reads.connection, pubkey); + } +} diff --git a/src/bundled/amp/tabs.tsx b/src/bundled/amp/tabs.tsx new file mode 100644 index 0000000000..49c61998dc --- /dev/null +++ b/src/bundled/amp/tabs.tsx @@ -0,0 +1,167 @@ +// Amp's New agent choices and agent settings. +import { useEffect, useId, useRef, useState } from "react"; +import { AgentInstructions } from "../../features/agents/AgentInstructions"; +import { AgentSaveButton } from "../../features/agents/AgentSaveButton"; +import type { + AgentCreateProps, + AgentViewProps, +} from "../../features/agents/service"; +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 { SettingsGroup } from "../../shared/design-system/ui/SettingsGroup"; +import { effortName, validWorkspace } from "../codex/config"; +import { MODES } from "./amp"; +import { type Config, DEFAULT_CONFIG, config as readConfig } from "./runtime"; + +/** Select items need a value, so Amp's default mode has one of its own. */ +const DEFAULT_MODE = "default"; + +export const modeLabel = (mode: string) => + mode ? effortName(mode) : "Amp default"; + +const MODE_GROUPS = [ + { + label: "", + options: [ + { value: DEFAULT_MODE, label: modeLabel("") }, + ...MODES.map((mode) => ({ value: mode, label: modeLabel(mode) })), + ], + }, +]; + +const SCOPES = [ + { + label: "", + options: [ + { value: "thread", label: "One Amp thread per Buzz thread" }, + { value: "channel", label: "One Amp thread per channel" }, + ], + }, +]; + +const message = (error: unknown) => + error instanceof Error ? error.message : String(error); + +/** 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, + settingsSection, + settingsContent, +}: AgentViewProps) { + const form = useId(); + const saved = readConfig(agent.config); + const [draft, setDraft] = useState(saved); + const [saving, setSaving] = useState(false); + const [error, setError] = useState(""); + const saveErrorRef = useRef(null); + useEffect(() => { + if (error) saveErrorRef.current?.focus(); + }, [error]); + const configChanged = JSON.stringify(draft) !== JSON.stringify(saved); + const changed = configChanged || Boolean(settingsHost?.pendingSave); + const submit = async () => { + setSaving(true); + setError(""); + try { + const workspace = draft.workspace.trim() || DEFAULT_CONFIG.workspace; + if (!validWorkspace(workspace)) + throw new Error("Enter an absolute or ~/… workspace path."); + await save({ ...draft, workspace }); + } catch (reason) { + setError(message(reason)); + } finally { + setSaving(false); + } + }; + return ( +
{ + event.preventDefault(); + void submit(); + }} + > + {error && ( +

+ {error} +

+ )} + {settingsSection !== "runtime" && ( + <> + + {settingsHost?.typeControl?.(configChanged)} + setDraft({ ...draft, workspace })} + /> + +