From c6210672c94fde90e25f5b52062c6f5def2f1dc5 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 12:56:38 -0400 Subject: [PATCH] feat(agents): carry local agents and their work over the agents port The screens' page reads the local agent list for names, mentions and the composer's agent choices, and each request's work for the Starting, Working and Error rows and their Stop. Both come from the agents service. Once it runs in another page, the port pushes them with the rest of its state, and carries Retry, Dismiss and Stop back. 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/features/agents/port.test.ts | 82 +++++++++++++++++ src/features/agents/port.ts | 137 ++++++++++++++++++++-------- src/features/agents/service.test.ts | 48 ++++++++++ 3 files changed, 227 insertions(+), 40 deletions(-) diff --git a/src/features/agents/port.test.ts b/src/features/agents/port.test.ts index a667cf5917..19cc2bf202 100644 --- a/src/features/agents/port.test.ts +++ b/src/features/agents/port.test.ts @@ -4,6 +4,8 @@ import type { AddedTool } from "../../buzz-mcp/session-tools"; import type { Events, Frame } from "./event-port"; import { eventPort } from "./event-port"; import { keyedRead } from "./handles"; +import { createLocalAgents, type LocalAgentsSnapshot } from "./local"; +import { createLocalAgentActivity } from "./local-activity"; import { connectAgents, serveAgents, shared } from "./port"; import type { Agents, @@ -352,6 +354,86 @@ it("ends a hosted type's step when the screens' page reloads, and lists it again await vi.waitFor(() => expect(added.has("example/cloud")).toBe(true)); }); +it("shows the service's local agents' work, and carries Dismiss and Stop back", async () => { + const activity = createLocalAgentActivity(); + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(fakeAgents() as unknown as Agents, port1, { + activity: activity.reader, + }); + const client = connectAgents(port2); + stops.push(() => { + client.close(); + stop(); + }); + const place = { scope: "s", agent: "a", channelId: "c", threadRootId: "r" }; + const failed = activity.begin({ ...place, messageId: "m1" }); + const running = activity.begin({ ...place, messageId: "m2" }); + failed.finish({ ok: false, error: "No model" }); + running.started(); + running.step("Read", "notes.md"); + const halt = vi.fn(async () => {}); + running.interruptible(halt); + await vi.waitFor(() => + expect(client.activity.work()).toMatchObject([ + { messageId: "m1", state: "error", error: "No model" }, + { + messageId: "m2", + state: "working", + step: { tool: "Read", target: "notes.md" }, + stoppable: true, + }, + ]), + ); + const [shownFailure, shownRun] = client.activity.work(); + if (!shownFailure || !shownRun) throw new Error("The work is not shown"); + client.activity.dismissError?.(shownFailure); + await vi.waitFor(() => + expect(client.activity.work()[0]).toMatchObject({ dismissed: true }), + ); + await client.activity.interrupt?.(shownRun); + expect(halt).toHaveBeenCalledOnce(); + expect(client.activity.work()[1]).toMatchObject({ state: "stopped" }); +}); + +it("shows the service's local agents, and asks it to read them again", async () => { + let current: LocalAgentsSnapshot = { status: "loading", agents: [] }; + const listeners = new Set<() => void>(); + const refresh = vi.fn(async () => { + current = { + status: "ready", + agents: [{ pubkey: "p", name: "Echo", relayUrl: "wss://c", owner: "o" }], + }; + for (const listener of listeners) listener(); + }); + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(fakeAgents() as unknown as Agents, port1, { + local: { + snapshot: () => current, + subscribe: (listener) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + refresh, + }, + }); + const client = connectAgents(port2); + stops.push(() => { + client.close(); + stop(); + }); + const names = createLocalAgents(); + names.bind(client.local); + await names.agents.refresh(); + expect(refresh).toHaveBeenCalledOnce(); + expect(names.agents.snapshot()).toEqual(current); + // A service with no local agents says so. + const bare = connected(fakeAgents()); + expect(bare.local.snapshot().status).toBe("loading"); + await vi.waitFor(() => + expect(bare.local.snapshot().status).toBe("unavailable"), + ); +}); + 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 93c1fa2018..6920b96c27 100644 --- a/src/features/agents/port.ts +++ b/src/features/agents/port.ts @@ -15,6 +15,8 @@ import { type SessionState, } from "./service"; import type { AddedTool } from "../../buzz-mcp/session-tools"; +import type { LocalAgents, LocalAgentsSnapshot } from "./local"; +import type { LocalAgentSource, LocalWork } from "./local-activity"; import type { Reply } from "../../buzz-mcp/tools"; import { type HostedCall, @@ -84,7 +86,12 @@ type State = Readonly<{ /** This serving of the service. A new one has not heard what the screens' * page asked of the last. */ instance: string; + /** The agents this machine keeps, for names, mentions and choices. */ + local: LocalAgentsSnapshot; + /** What each local agent is doing with each request it got. */ + work: readonly LocalWork[]; }>; +type WorkKey = Pick; type ToClient = | Readonly<{ kind: "state"; state: State }> | Readonly<{ kind: "tool"; call: ToolCall }> @@ -126,7 +133,10 @@ type ToService = kind: "settle-type"; id: number; outcome: { value?: unknown } | { error: string }; - }>; + }> + | Readonly<{ kind: "refresh-local"; id: number }> + | Readonly<{ kind: "dismiss-work"; work: WorkKey }> + | Readonly<{ kind: "interrupt-work"; id: number; work: WorkKey }>; /** A `MessagePort`, or a link to a page in another webview. It delivers each * message once, in the order sent. */ @@ -147,16 +157,27 @@ export type AgentsPortClient = AgentsClient & /** Registers `owner`'s hosted `type` for the service, its provider steps * answered in this page, until the returned withdraw. */ register(type: AgentType, owner: ToolOwner): () => void; + /** The service's local agents, as `createLocalAgents` binds them. */ + local: LocalAgents; + /** The service's local agents' work, for status rows and Stop. */ + activity: LocalAgentSource; }>; const placeKey = (key: string, { channelId, root }: SessionPlace) => JSON.stringify([key, channelId, root ?? ""]); +const UNAVAILABLE: LocalAgentsSnapshot = Object.freeze({ + status: "unavailable", + agents: Object.freeze([]), +}); + function stateOf( agents: Agents, types: TypesState, places: ReadonlyMap, instance: string, + local: LocalAgents | undefined, + activity: LocalAgentSource | undefined, ): State { const snapshot = agents.snapshot(); const memberships: Record> = {}; @@ -191,32 +212,38 @@ function stateOf( ]), ), instance, + local: local?.snapshot() ?? UNAVAILABLE, + work: activity?.work() ?? [], }; } /** 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. A * tool or hosted type the screens' page lists is added, as its plugin's, - * with `plugins.addTool` or `plugins.register` until it withdraws it; - * without them, it is not added. */ + * with `options.addTool` or `options.register` until it withdraws it; + * without them, it is not added. `local` and `activity` are the service's + * local agents and their work, for the screens' page's names and status. */ export function serveAgents( agents: Agents, port: AgentsPort, - plugins: Readonly<{ + options: Readonly<{ addTool?: (tool: AddedTool, owner: ToolOwner) => () => void; register?: (type: AgentType, owner: ToolOwner) => () => void; + local?: LocalAgents; + activity?: LocalAgentSource; }> = {}, ) { + const { local, activity } = options; const types = serveTypes(agents); const instance = crypto.randomUUID(); - const tools = plugins.addTool - ? serveTools(plugins.addTool, (call) => + const tools = options.addTool + ? serveTools(options.addTool, (call) => port.postMessage({ kind: "tool", call } satisfies ToClient), ) : undefined; - const hosted = plugins.register + const hosted = options.register ? serveHosted( - plugins.register, + options.register, (call) => port.postMessage({ kind: "hosted", call } satisfies ToClient), (id) => port.postMessage({ kind: "hosted-abort", id } satisfies ToClient), @@ -239,7 +266,7 @@ export function serveAgents( if (!stores.has(subscribe)) stores.set(subscribe, subscribe(changed)); port.postMessage({ kind: "state", - state: stateOf(agents, types.state(), places, instance), + state: stateOf(agents, types.state(), places, instance, local, activity), } satisfies ToClient); }; // Several changes in one task are one push. @@ -249,6 +276,8 @@ export function serveAgents( queueMicrotask(flush); }; const unsubscribe = agents.subscribe(changed); + const unsubscribeLocal = local?.subscribe(changed); + const unsubscribeWork = activity?.subscribeWork(changed); const calls = new Set(CALLS); const sends = new Set(SENDS); port.onmessage = ({ data }: MessageEvent) => { @@ -304,6 +333,23 @@ export function serveAgents( hosted?.settle(data.id, data.outcome); return; } + if (data.kind === "refresh-local") { + answer(data.id, () => local?.refresh()); + return; + } + if (data.kind === "dismiss-work") { + activity?.dismissError?.(data.work); + return; + } + if (data.kind === "interrupt-work") { + const { work } = data; + answer(data.id, () => { + if (!activity?.interrupt) + throw new Error("This request cannot be stopped"); + return activity.interrupt(work); + }); + return; + } if (data.kind === "client") { const { client, method, args } = data; answer(data.id, () => { @@ -340,6 +386,8 @@ export function serveAgents( flush(); return () => { unsubscribe(); + unsubscribeLocal?.(); + unsubscribeWork?.(); tools?.stop(); hosted?.stop(); for (const stop of stores.values()) stop(); @@ -372,6 +420,8 @@ const STARTING: State = Object.freeze({ }), sessions: Object.freeze({}), instance: "", + local: Object.freeze({ status: "loading", agents: EMPTY }), + work: EMPTY, }); /** `next`, keeping each part of `prior` it equals, so screens that compare by @@ -468,19 +518,23 @@ export function connectAgents(port: AgentsPort) { }; // The service may have pushed before this page listened. post({ kind: "hello" }); + /** Posts what `message` makes of a new request id, and waits for its + * answer. */ + const request = (message: (id: number) => ToService) => + new Promise((resolve, reject) => { + const id = ++next; + waiting.set(id, { resolve, reject }); + try { + post(message(id)); + } catch (error) { + waiting.delete(id); + reject(error); + } + }); const call = (method: Call) => (...args: unknown[]) => - new Promise((resolve, reject) => { - const id = ++next; - waiting.set(id, { resolve, reject }); - try { - post({ kind: "call", id, method, args }); - } catch (error) { - waiting.delete(id); - reject(error); - } - }); + request((id) => ({ kind: "call", id, method, args })); const send = (method: Send) => (...args: unknown[]) => { @@ -504,17 +558,7 @@ export function connectAgents(port: AgentsPort) { () => state.types, subscribe, { - call: (call, args) => - new Promise((resolve, reject) => { - const id = ++next; - waiting.set(id, { resolve, reject }); - try { - post({ kind: "type", id, call, args }); - } catch (error) { - waiting.delete(id); - reject(error); - } - }), + call: (call, args) => request((id) => ({ kind: "type", id, call, args })), watch: (watch) => { if (!closed) post({ kind: "watch", watch }); }, @@ -540,16 +584,7 @@ export function connectAgents(port: AgentsPort) { 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); - } - }), + request((id) => ({ kind: "client", id, client, method, args })), }); const hosted = hostedClient( { @@ -577,6 +612,28 @@ export function connectAgents(port: AgentsPort) { present: types.present, addTool: tools.add, register: hosted.register, + local: Object.freeze({ + snapshot: () => state.local, + subscribe, + refresh: async () => { + await request((id) => ({ kind: "refresh-local", id })); + }, + }), + activity: Object.freeze({ + work: () => state.work, + subscribeWork: subscribe, + dismissError: ({ scope, agent, messageId }: WorkKey) => { + if (!closed) + post({ kind: "dismiss-work", work: { scope, agent, messageId } }); + }, + interrupt: async ({ scope, agent, messageId }: WorkKey) => { + await request((id) => ({ + kind: "interrupt-work", + id, + work: { scope, agent, messageId }, + })); + }, + }), 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 8e0fdda072..5e6e55e78b 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -7592,6 +7592,54 @@ it("creates and removes a hosted agent whose type the screens' page registers, a stop(); }); +it("shows the screens' page its local agents and their work through the port", async () => { + const { service, run, emit, activity } = await setup(); + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(service, port1, { + local: { + snapshot: service.local, + subscribe: service.subscribe, + refresh: () => service.refresh(), + }, + activity: activity.reader, + }); + const client = connectAgents(port2); + await service.create({ type: "example/echo", name: "Echo" }); + await vi.waitFor(() => + expect(client.local.snapshot()).toMatchObject({ + status: "ready", + agents: [{ pubkey: bot, name: "Echo" }], + }), + ); + let finish = () => {}; + const turn = new Promise((resolve) => { + finish = resolve; + }); + run.mockImplementationOnce(async ({ workingUntil }) => workingUntil(turn)); + const mention = event("m", { + tags: [ + ["h", channel], + ["p", bot], + ], + }); + emit({ events: [mention], channelId: channel }); + try { + await vi.waitFor(() => + expect(client.activity.work()).toMatchObject([ + { agent: bot, messageId: mention.id, state: "working" }, + ]), + ); + finish(); + await vi.waitFor(() => + expect(client.activity.work()[0]?.state).toBe("done"), + ); + } finally { + finish(); + client.close(); + stop(); + } +}); + it("reserves a tool name while its plugin is still starting", async () => { const ctx = new Context(); ctx.provide("pluginStatus", {