From eec62e1bfd3cc5cfd5803deea4d505c57b8f7e16 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 12:27:21 -0400 Subject: [PATCH] feat(agents): forward tools the screens' page adds to the agents port Builderlab, GitHub link_pr and installed plugins such as loganj.branches add agent tools from their UI entry. Once agents run in another page, the screens' page lists each tool with the service, and the service adds a stand-in that forwards each call back. The call carries the agent's client as a handle: its values, and its methods, which the screens' page calls back over the port, with bytes as base64 on the JSON path. A reloaded service is told the tools again; a call left unanswered by a screens' page that reloaded ends with an error. Nothing uses it yet. Co-authored-by: Larry <627498bd4bd1f281a16431e3c6cce3b5c25b6692798c78672298aefbf2f8f8b5@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Logan Johnson --- src/buzz-mcp/app-client.ts | 4 +- src/features/agents/port.test.ts | 101 ++++++++++- src/features/agents/port.ts | 110 +++++++++++- src/features/agents/service.test.ts | 54 ++++++ src/features/agents/tool-port.ts | 250 ++++++++++++++++++++++++++++ 5 files changed, 510 insertions(+), 9 deletions(-) create mode 100644 src/features/agents/tool-port.ts diff --git a/src/buzz-mcp/app-client.ts b/src/buzz-mcp/app-client.ts index e40d6359f7..e56a6ae543 100644 --- a/src/buzz-mcp/app-client.ts +++ b/src/buzz-mcp/app-client.ts @@ -201,13 +201,13 @@ export async function readFile( return fromBase64(output.replace(/\s/g, "")); } -function toBase64(data: Uint8Array) { +export function toBase64(data: Uint8Array) { let binary = ""; for (let i = 0; i < data.length; i += 0x8000) binary += String.fromCharCode(...data.subarray(i, i + 0x8000)); return btoa(binary); } -function fromBase64(text: string) { +export function fromBase64(text: string) { const binary = atob(text); const data = new Uint8Array(binary.length); for (let i = 0; i < binary.length; i++) data[i] = binary.charCodeAt(i); diff --git a/src/features/agents/port.test.ts b/src/features/agents/port.test.ts index 2dc453c59f..6d74dcfcbc 100644 --- a/src/features/agents/port.test.ts +++ b/src/features/agents/port.test.ts @@ -1,4 +1,6 @@ import { afterEach, expect, it, vi } from "vitest"; +import type { BuzzClient } from "../../buzz-mcp/client"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; import type { Events, Frame } from "./event-port"; import { eventPort } from "./event-port"; import { keyedRead } from "./handles"; @@ -86,9 +88,10 @@ function connected(agents: ReturnType) { return client; } -it("serves screens in another webview, whichever page loads first", async () => { +/** Tauri events between webviews, each delivered later and as JSON. */ +function jsonEvents(): Events { const handlers = new Set<(event: { payload: Frame }) => void>(); - const events: Events = { + return { emitTo: async (_webview, _event, frame) => { const payload = JSON.parse(JSON.stringify(frame)) as Frame; setTimeout(() => { @@ -100,6 +103,10 @@ it("serves screens in another webview, whichever page loads first", async () => return () => handlers.delete(handler); }, }; +} + +it("serves screens in another webview, whichever page loads first", async () => { + const events = jsonEvents(); const agents = fakeAgents(); const client = connectAgents(eventPort("main", "agents", events)); await new Promise((resolve) => setTimeout(resolve, 10)); @@ -127,6 +134,96 @@ it("serves screens in another webview, whichever page loads first", async () => ); }); +it("carries a tool's bytes as JSON, and lists the tools again with a reloaded service", async () => { + const events = jsonEvents(); + const agents = fakeAgents() as unknown as Agents; + const added = new Map(); + const addTool = (tool: AddedTool) => { + added.set(tool.name, tool); + return () => added.delete(tool.name); + }; + const client = connectAgents(eventPort("main", "agents", events)); + client.addTool( + { + name: "files_copy", + description: "Copy a file.", + inputSchema: { type: "object" }, + call: async (agent) => { + const data = await agent.read("/tmp/a"); + return (await agent.upload(data, "text/plain")).url; + }, + }, + { id: "example", revision: "1" }, + ); + const first = serveAgents( + agents, + eventPort("agents", "main", events), + addTool, + ); + await vi.waitFor(() => expect(added.has("files_copy")).toBe(true)); + const upload = vi.fn(async (data: Uint8Array) => ({ + url: `https://media.example/${data.length}`, + })); + const agent = { + pubkey: "p", + key: "k", + read: async () => new Uint8Array([1, 2, 3]), + upload, + } as unknown as BuzzClient; + await expect(added.get("files_copy")?.call(agent, {}, {})).resolves.toBe( + "https://media.example/3", + ); + expect(upload).toHaveBeenCalledWith(new Uint8Array([1, 2, 3]), "text/plain"); + // The agents page reloads: its tools go with it, and come back listed. + first(); + expect(added.size).toBe(0); + const second = serveAgents( + agents, + eventPort("agents", "main", events), + addTool, + ); + stops.push(() => { + client.close(); + second(); + }); + await vi.waitFor(() => expect(added.has("files_copy")).toBe(true)); +}); + +it("ends a tool's call when the screens' page that answers it reloads", async () => { + const events = jsonEvents(); + const added = new Map(); + const stop = serveAgents( + fakeAgents() as unknown as Agents, + eventPort("agents", "main", events), + (tool) => { + added.set(tool.name, tool); + return () => added.delete(tool.name); + }, + ); + const tool = { + name: "notes_wait", + description: "Wait.", + inputSchema: { type: "object" }, + call: () => new Promise(() => {}), + } as const; + const owner = { id: "example", revision: "1" }; + const before = connectAgents(eventPort("main", "agents", events)); + before.addTool(tool, owner); + await vi.waitFor(() => expect(added.has("notes_wait")).toBe(true)); + const waiting = added + .get("notes_wait") + ?.call({ pubkey: "p", key: "k" } as unknown as BuzzClient, {}, {}); + await new Promise((resolve) => setTimeout(resolve, 10)); + before.close(); + const after = connectAgents(eventPort("main", "agents", events)); + after.addTool(tool, owner); + stops.push(() => { + after.close(); + stop(); + }); + await expect(waiting).rejects.toThrow(/reloaded/); +}); + it("reads as loading until the service's first push, then as the service", async () => { const client = connected(fakeAgents()); expect(client.snapshot().status).toBe("loading"); diff --git a/src/features/agents/port.ts b/src/features/agents/port.ts index 4be5157e00..dd84e2bea8 100644 --- a/src/features/agents/port.ts +++ b/src/features/agents/port.ts @@ -12,6 +12,15 @@ import { type SessionPlace, type SessionState, } from "./service"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; +import type { Reply } from "../../buzz-mcp/tools"; +import { + serveTools, + type ToolCall, + type ToolListing, + type ToolOwner, + toolsClient, +} from "./tool-port"; import { serveTypes, type TypeCall, @@ -64,9 +73,13 @@ type State = Readonly<{ types: TypesState; /** Each session state a screen asked for, by `placeKey`. */ sessions: Readonly>; + /** This serving of the service. A new one has not heard what the screens' + * page asked of the last. */ + instance: string; }>; type ToClient = | Readonly<{ kind: "state"; state: State }> + | Readonly<{ kind: "tool"; call: ToolCall }> | Readonly<{ kind: "result"; id: number; value: unknown }> | Readonly<{ kind: "error"; id: number; message: string }>; type ToService = @@ -76,7 +89,21 @@ type ToService = | Readonly<{ kind: "type"; id: number; call: TypeCall; args: unknown[] }> | Readonly<{ kind: "watch"; watch: TypeWatch }> | Readonly<{ kind: "session"; key: string; place: SessionPlace }> - | Readonly<{ kind: "abort"; signal: number }>; + | Readonly<{ kind: "abort"; signal: number }> + | Readonly<{ kind: "list"; tool: ToolListing }> + | Readonly<{ kind: "withdraw"; name: string }> + | Readonly<{ + kind: "settle"; + id: number; + outcome: { reply: Reply } | { error: string }; + }> + | Readonly<{ + kind: "client"; + id: number; + client: number; + method: string; + args: unknown[]; + }>; /** A `MessagePort`, or a link to a page in another webview. It delivers each * message once, in the order sent. */ @@ -89,7 +116,12 @@ export type AgentsPort = { /** `AgentsClient`, and `present` for what this page shows of each type (by * type key). */ export type AgentsPortClient = AgentsClient & - Readonly<{ present: ReturnType["present"] }>; + Readonly<{ + present: ReturnType["present"]; + /** Adds `owner`'s `tool` for the service's agents, answered in this + * page, until the returned withdraw. */ + addTool(tool: AddedTool, owner: ToolOwner): () => void; + }>; const placeKey = (key: string, { channelId, root }: SessionPlace) => JSON.stringify([key, channelId, root ?? ""]); @@ -98,6 +130,7 @@ function stateOf( agents: Agents, types: TypesState, places: ReadonlyMap, + instance: string, ): State { const snapshot = agents.snapshot(); const memberships: Record> = {}; @@ -131,13 +164,26 @@ function stateOf( agents.sessionState(key, place), ]), ), + instance, }; } /** Serves `agents` on `port` until the returned stop. It pushes state before - * each answer, so a call's result never arrives before what it changed. */ -export function serveAgents(agents: Agents, port: AgentsPort) { + * each answer, so a call's result never arrives before what it changed. A + * tool the screens' page lists is added with `addTool` until it withdraws + * it; without `addTool`, its tools are not added. */ +export function serveAgents( + agents: Agents, + port: AgentsPort, + addTool?: (tool: AddedTool, owner: ToolOwner) => () => void, +) { const types = serveTypes(agents); + const instance = crypto.randomUUID(); + const tools = addTool + ? serveTools(addTool, (call) => + port.postMessage({ kind: "tool", call } satisfies ToClient), + ) + : undefined; const places = new Map(); // The types' own stores, which change without the service's notice. const stores = new Map<(listener: () => void) => () => void, () => void>(); @@ -155,7 +201,7 @@ export function serveAgents(agents: Agents, port: AgentsPort) { if (!stores.has(subscribe)) stores.set(subscribe, subscribe(changed)); port.postMessage({ kind: "state", - state: stateOf(agents, types.state(), places), + state: stateOf(agents, types.state(), places, instance), } satisfies ToClient); }; // Several changes in one task are one push. @@ -192,6 +238,26 @@ export function serveAgents(agents: Agents, port: AgentsPort) { types.abort(data.signal); return; } + if (data.kind === "list") { + tools?.list(data.tool); + return; + } + if (data.kind === "withdraw") { + tools?.withdraw(data.name); + return; + } + if (data.kind === "settle") { + tools?.settle(data.id, data.outcome); + return; + } + if (data.kind === "client") { + const { client, method, args } = data; + answer(data.id, () => { + if (!tools) throw new Error("That tool call has finished"); + return tools.client(client, method, args); + }); + return; + } if (data.kind === "type") { answer(data.id, () => types.call(data.call, data.args)); return; @@ -220,6 +286,7 @@ export function serveAgents(agents: Agents, port: AgentsPort) { flush(); return () => { unsubscribe(); + tools?.stop(); for (const stop of stores.values()) stop(); stores.clear(); port.onmessage = null; @@ -249,6 +316,7 @@ const STARTING: State = Object.freeze({ scopes: Object.freeze({}), }), sessions: Object.freeze({}), + instance: "", }); /** `next`, keeping each part of `prior` it equals, so screens that compare by @@ -312,9 +380,18 @@ export function connectAgents(port: AgentsPort) { if (data.kind === "state") { const prior = state; state = shared(prior, data.state); + if (prior !== STARTING && state.instance !== prior.instance) { + // A new serving, as when its page reloaded: tell it again. + asked.clear(); + tools.relist(); + } if (state !== prior) for (const listener of listeners) listener(); return; } + if (data.kind === "tool") { + void tools.answer(data.call); + return; + } const pending = waiting.get(data.id); if (!pending) return; waiting.delete(data.id); @@ -388,6 +465,28 @@ export function connectAgents(port: AgentsPort) { const sends = Object.fromEntries( SENDS.map((method) => [method, send(method)]), ); + const tools = toolsClient({ + list: (tool) => { + if (!closed) post({ kind: "list", tool }); + }, + withdraw: (name) => { + if (!closed) post({ kind: "withdraw", name }); + }, + settle: (id, outcome) => { + if (!closed) post({ kind: "settle", id, outcome }); + }, + call: (client, method, args) => + new Promise((resolve, reject) => { + const id = ++next; + waiting.set(id, { resolve, reject }); + try { + post({ kind: "client", id, client, method, args }); + } catch (error) { + waiting.delete(id); + reject(error); + } + }), + }); const client = { ...calls, ...sends, @@ -395,6 +494,7 @@ export function connectAgents(port: AgentsPort) { subscribe, types: types.types, present: types.present, + addTool: tools.add, find: (pubkey: string) => state.snapshot.agents.find((agent) => agent.pubkey === pubkey), classifier: () => state.snapshot.classifier, diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index c772a30b1c..2c4aee3426 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -13,6 +13,7 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { forgetDefaultAgent } from "../solo/default-agent"; import { AgentConfigStore, type Subject } from "./agent-config"; import { connectAgents, serveAgents } from "./port"; +import type { BuzzClient } from "../../buzz-mcp/client"; import type { Mode } from "./agent-start"; import { channelTemplates } from "../channel-templates/local"; import { eventDto, type ReadFilter, type RelayEvent } from "../relay/events"; @@ -7439,6 +7440,59 @@ it("adds plugins' tools to every session, each name once", async () => { ]); }); +it("answers a tool the screens' page adds there, calling the agent's client back", async () => { + const { service, ctx } = await setup(); + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(service, port1, (tool, owner) => { + const fiber = ctx + .extend({ pluginOwner: Object.freeze({ ...owner }) }) + .plugin({ + inject: ["agents"], + apply: (plugin: Context) => plugin.agents.addTool(tool), + }); + return () => void fiber.dispose(); + }); + const client = connectAgents(port2); + const withdraw = client.addTool( + { + name: "notes_add", + description: "Add a note.", + inputSchema: { type: "object" }, + call: async (agent, context, args) => { + const found = await agent.query([{ kinds: [1] }]); + const bytes = await agent.read("/tmp/note"); + if (args.fail) throw new Error("No notebook"); + return `${agent.pubkey} in ${context.channel}: ${found.length} found, ${bytes.length} bytes, ${String(args.text)}`; + }, + }, + { id: "example", revision: "1" }, + ); + await vi.waitFor(() => + expect(service.tools()).toMatchObject([ + { name: "notes_add", pluginId: "example" }, + ]), + ); + const agent = { + pubkey: bot, + key: "k", + query: vi.fn(async () => [{ id: "e" }]), + read: vi.fn(async () => new Uint8Array([1, 2, 3])), + } as unknown as BuzzClient; + const tool = service.tools().find((item) => item.name === "notes_add"); + if (!tool) throw new Error("The tool is not listed"); + await expect( + tool.call(agent, { channel: "c" }, { text: "hi" }), + ).resolves.toBe(`${bot} in c: 1 found, 3 bytes, hi`); + expect(agent.query).toHaveBeenCalledWith([{ kinds: [1] }]); + await expect( + tool.call(agent, { channel: "c" }, { fail: true }), + ).rejects.toThrow("No notebook"); + withdraw(); + await vi.waitFor(() => expect(service.tools()).toEqual([])); + client.close(); + stop(); +}); + it("reserves a tool name while its plugin is still starting", async () => { const ctx = new Context(); ctx.provide("pluginStatus", { diff --git a/src/features/agents/tool-port.ts b/src/features/agents/tool-port.ts new file mode 100644 index 0000000000..8821736101 --- /dev/null +++ b/src/features/agents/tool-port.ts @@ -0,0 +1,250 @@ +// Agent tools a plugin adds from the screens' page, for agents the service +// runs in another page. The screens' page lists each tool with the service +// and answers its calls. A call carries the agent's client as a handle: its +// values, and its methods, which the screens' page calls back over the port. +// So a plugin that adds tools without an agent entry keeps working, though +// its tools wait on the screens' page. +import type { BuzzClient, Context } from "../../buzz-mcp/client"; +import { fromBase64, toBase64 } from "../../buzz-mcp/app-client"; +import type { AddedTool } from "../../buzz-mcp/session-tools"; +import type { Reply } from "../../buzz-mcp/tools"; + +/** The plugin that adds a tool, as its page's plugin runtime names it. */ +export type ToolOwner = Readonly<{ id: string; revision: string }>; +export type ToolListing = Pick< + AddedTool, + "name" | "description" | "inputSchema" +> & + Readonly<{ owner: ToolOwner }>; + +const VALUES = [ + "pubkey", + "key", + "owner", + "relayAuthor", + "signingAuthority", +] as const satisfies readonly (keyof BuzzClient)[]; +const METHODS = [ + "agentTypes", + "createAgent", + "updateAgent", + "handoff", + "query", + "searchProfiles", + "publish", + "upload", + "media", + "gitAuth", + "read", + "memories", + "remember", + "templates", +] as const satisfies readonly (keyof BuzzClient)[]; +type Method = (typeof METHODS)[number]; + +/** The agent's client as one call carries it. Its thread tabs and attention + * stay behind: what they read is the service's page's. */ +export type ClientHandle = Readonly<{ + id: number; + values: Readonly>>; + methods: readonly Method[]; +}>; +export type ToolCall = Readonly<{ + id: number; + name: string; + client: ClientHandle; + context: Context; + args: Record; +}>; + +type Bytes = Readonly<{ $bytes: string }>; +const isBytes = (value: unknown): value is Bytes => + typeof value === "object" && + value !== null && + typeof (value as Bytes).$bytes === "string"; +/** Bytes as JSON carries them. */ +export const wire = (value: unknown) => + value instanceof Uint8Array ? { $bytes: toBase64(value) } : value; +export const unwire = (value: unknown) => + isBytes(value) ? fromBase64(value.$bytes) : value; + +/** The service's side: each listed tool, added with `addTool` until it is + * withdrawn, forwards its calls with `post`. */ +export function serveTools( + addTool: (tool: AddedTool, owner: ToolOwner) => () => void, + post: (call: ToolCall) => void, +) { + const added = new Map void>(); + const clients = new Map(); + const waiting = new Map< + number, + Readonly<{ + name: string; + resolve(reply: Reply): void; + reject(error: Error): void; + }> + >(); + let next = 0; + /** Ends the calls of tools `gone` picks, which will not be answered. */ + const abandon = (gone: (name: string) => boolean, why: string) => { + for (const pending of waiting.values()) + if (gone(pending.name)) pending.reject(new Error(why)); + }; + const forward = async ( + name: string, + client: BuzzClient, + context: Context, + args: Record, + ) => { + const id = ++next; + clients.set(id, client); + try { + return await new Promise((resolve, reject) => { + waiting.set(id, { name, resolve, reject }); + post({ + id, + name, + client: { + id, + values: Object.fromEntries( + VALUES.flatMap((value) => + typeof client[value] === "string" + ? [[value, client[value]]] + : [], + ), + ), + methods: METHODS.filter( + (method) => typeof client[method] === "function", + ), + }, + context, + args, + }); + }); + } finally { + waiting.delete(id); + clients.delete(id); + } + }; + return { + list(tool: ToolListing) { + // Listed again: the page that added it reloaded, or added it anew. + abandon( + (name) => name === tool.name, + `The page that adds agent tool ${tool.name} reloaded`, + ); + added.get(tool.name)?.(); + try { + added.set( + tool.name, + addTool( + { + name: tool.name, + description: tool.description, + inputSchema: tool.inputSchema, + call: (client, context, args) => + forward(tool.name, client, context, args), + }, + tool.owner, + ), + ); + } catch (error) { + added.delete(tool.name); + console.error(`Agent tool ${tool.name} was not added`, error); + } + }, + withdraw(name: string) { + added.get(name)?.(); + added.delete(name); + }, + settle(id: number, outcome: { reply: Reply } | { error: string }) { + const pending = waiting.get(id); + if (!pending) return; + if ("reply" in outcome) pending.resolve(outcome.reply); + else pending.reject(new Error(outcome.error)); + }, + /** A call the screens' page makes on the client of a call in progress. */ + async client(id: number, method: string, args: readonly unknown[]) { + const client = clients.get(id); + const run = client?.[method as Method] as + | ((...args: unknown[]) => unknown) + | undefined; + if (!client || !METHODS.includes(method as Method) || !run) + throw new Error("That tool call has finished"); + return wire(await run.apply(client, args.map(unwire))); + }, + stop() { + for (const remove of added.values()) remove(); + added.clear(); + abandon(() => true, "The agents connection is closed."); + }, + }; +} + +/** The screens' page's side: it lists each tool it adds with `list`, and + * answers the service's calls of them with `settle`, calling the agent's + * client back with `call`. */ +export function toolsClient( + send: Readonly<{ + list(tool: ToolListing): void; + withdraw(name: string): void; + settle(id: number, outcome: { reply: Reply } | { error: string }): void; + call(client: number, method: Method, args: unknown[]): Promise; + }>, +) { + const tools = new Map< + string, + Readonly<{ tool: AddedTool; listing: ToolListing }> + >(); + const handle = ({ id, values, methods }: ClientHandle): BuzzClient => + Object.freeze({ + ...values, + ...Object.fromEntries( + methods.map((method) => [ + method, + (...args: unknown[]) => { + const answer = send.call(id, method, args.map(wire)).then(unwire); + if (method !== "handoff") return answer; + // It answers nothing; a failure is the service's to report. + answer.catch(() => {}); + }, + ]), + ), + }) as unknown as BuzzClient; + return { + /** Lists `owner`'s `tool` with the service until the returned withdraw. */ + add(tool: AddedTool, owner: ToolOwner) { + const listing: ToolListing = { + name: tool.name, + description: tool.description, + inputSchema: tool.inputSchema, + owner, + }; + const entry = Object.freeze({ tool, listing }); + tools.set(tool.name, entry); + send.list(listing); + return () => { + if (tools.get(tool.name) !== entry) return; + tools.delete(tool.name); + send.withdraw(tool.name); + }; + }, + async answer({ id, name, client, context, args }: ToolCall) { + const tool = tools.get(name)?.tool; + try { + if (!tool) throw new Error(`Agent tool ${name} is no longer added`); + send.settle(id, { + reply: await tool.call(handle(client), context, args), + }); + } catch (error) { + send.settle(id, { + error: error instanceof Error ? error.message : String(error), + }); + } + }, + /** The service's page reloaded: it lists them again. */ + relist() { + for (const { listing } of tools.values()) send.list(listing); + }, + }; +}