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", {