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); + }, + }; +}