From 46fb1ba410702448107bc3c94f9f8db60d7160e5 Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 11:02:11 -0400 Subject: [PATCH] feat(agents): serve the agents service to screens over a MessagePort serveAgents pushes what screens read synchronously (the snapshot, each shown agent's memberships, joinable agents, which hosted agents can be interrupted, and the creation and update requests) and answers their calls. It pushes state before each answer, so a call's result never arrives before what it changed. connectAgents keeps the pushed state with structural sharing, so an unchanged agent keeps its identity. Nothing uses it yet. Types, session state and the retype preview call plugin code and come in a later change. 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 | 125 ++++++++++++ src/features/agents/port.ts | 302 ++++++++++++++++++++++++++++ src/features/agents/service.test.ts | 34 ++++ 3 files changed, 461 insertions(+) create mode 100644 src/features/agents/port.test.ts create mode 100644 src/features/agents/port.ts diff --git a/src/features/agents/port.test.ts b/src/features/agents/port.test.ts new file mode 100644 index 0000000000..e98f016120 --- /dev/null +++ b/src/features/agents/port.test.ts @@ -0,0 +1,125 @@ +import { afterEach, expect, it, vi } from "vitest"; +import { connectAgents, serveAgents, shared } from "./port"; +import type { Agents, AgentsSnapshot } from "./service"; + +it("keeps each part of the prior value that the next one equals", () => { + const a = Object.freeze({ key: "a", name: "A", config: { x: 1 } }); + const b = Object.freeze({ key: "b", name: "B", config: { x: 2 } }); + const prior = Object.freeze({ + agents: Object.freeze([a, b]), + status: "ready", + }); + // As a structured clone arrives: equal, but none of it the same object. + const same = structuredClone(prior); + expect(shared(prior, same)).toBe(prior); + // b changed and moved first; a is still a. + const next = shared(prior, { + agents: [{ key: "b", name: "Bee", config: { x: 2 } }, structuredClone(a)], + status: "ready", + }); + expect(next).not.toBe(prior); + expect(next.agents[1]).toBe(a); + expect(next.agents[0]).not.toBe(b); + expect(next.agents[0]?.config).toBe(b.config); + expect(Object.isFrozen(next.agents[0])).toBe(true); + expect(shared(prior, { ...same, error: "x" })).not.toBe(prior); + expect(shared({ status: "ready", error: "x" }, { status: "ready" })).toEqual({ + status: "ready", + }); +}); + +const snapshot = (patch: Partial = {}): AgentsSnapshot => ({ + status: "ready", + classifier: "available", + classifierKey: "saved", + agents: [], + running: [], + interrupted: [], + teams: [], + ...patch, +}); +function fakeAgents() { + let state = snapshot(); + const listeners = new Set<() => void>(); + const agents = { + snapshot: () => state, + subscribe: (listener: () => void) => { + listeners.add(listener); + return () => listeners.delete(listener); + }, + memberships: () => [], + joinable: () => [], + interruptible: () => false, + creationRequests: () => [], + updateRequests: () => [], + resumeInterrupted: vi.fn(), + saveTeam: vi.fn(async (team: { name: string }) => { + state = snapshot({ + teams: [{ id: "t", name: team.name, instructions: "", members: [] }], + }); + for (const listener of listeners) listener(); + return state.teams[0]; + }), + removeTeam: vi.fn(async () => { + throw new Error("No such team"); + }), + }; + return agents; +} +let stops: (() => void)[] = []; +afterEach(() => { + for (const stop of stops) stop(); + stops = []; +}); +function connected(agents: ReturnType) { + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(agents as unknown as Agents, port1); + const client = connectAgents(port2); + stops.push(() => { + client.close(); + stop(); + }); + return client; +} + +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"); + const changed = vi.fn(); + client.subscribe(changed); + await vi.waitFor(() => expect(client.snapshot().status).toBe("ready")); + expect(changed).toHaveBeenCalledOnce(); + expect(client.classifier()).toBe("available"); +}); + +it("answers a call after the state it changed, and carries a refusal", async () => { + const agents = fakeAgents(); + const client = connected(agents); + const team = await client.saveTeam({ + name: "Crew", + instructions: "", + members: [], + }); + expect(team).toMatchObject({ id: "t", name: "Crew" }); + expect(client.snapshot().teams.map((item) => item.name)).toEqual(["Crew"]); + await expect(client.removeTeam("t")).rejects.toThrow("No such team"); +}); + +it("sends a handler's call without the click event", async () => { + const agents = fakeAgents(); + const client = connected(agents); + // As `onClick={agents.resumeInterrupted}` calls it. + (client.resumeInterrupted as (event: Event) => void)(new Event("click")); + await vi.waitFor(() => expect(agents.resumeInterrupted).toHaveBeenCalled()); + expect(agents.resumeInterrupted).toHaveBeenCalledWith(); +}); + +it("refuses calls once closed, and fails those still waiting", async () => { + const agents = fakeAgents(); + agents.removeTeam.mockReturnValueOnce(new Promise(() => {})); + const client = connected(agents); + const waiting = client.removeTeam("t"); + client.close(); + await expect(waiting).rejects.toThrow("closed"); + await expect(client.removeTeam("t")).rejects.toThrow("closed"); +}); diff --git a/src/features/agents/port.ts b/src/features/agents/port.ts new file mode 100644 index 0000000000..27c4a53a2c --- /dev/null +++ b/src/features/agents/port.ts @@ -0,0 +1,302 @@ +// The agents service and the screens that use it can run in different pages. +// The service answers over a MessagePort; the client holds what screens read +// synchronously, as the service last pushed it, and sends everything else. +import type { + Agent, + AgentCreationRequest, + Agents, + AgentsClient, + AgentsSnapshot, + AgentUpdateRequest, +} from "./service"; + +/** Calls a screen waits for. Each takes and returns structured-cloneable data. */ +const CALLS = [ + "create", + "approveCreation", + "approveUpdate", + "finish", + "portable", + "port", + "portTeam", + "save", + "retype", + "remove", + "forget", + "join", + "leave", + "interrupt", + "session", + "unarchive", + "setClassifierKey", + "clearClassifierKey", + "saveTeam", + "removeTeam", +] as const satisfies readonly (keyof AgentsClient)[]; +/** Calls a screen makes without waiting for an answer. */ +const SENDS = [ + "resumeInterrupted", + "dismissInterrupted", + "dismissCreation", + "dismissUpdate", +] as const satisfies readonly (keyof AgentsClient)[]; +type Call = (typeof CALLS)[number]; +type Send = (typeof SENDS)[number]; + +/** What screens read synchronously. */ +type State = Readonly<{ + snapshot: AgentsSnapshot; + memberships: Readonly>>; + joinable: readonly Agent[]; + interruptible: readonly string[]; + creationRequests: readonly AgentCreationRequest[]; + updateRequests: readonly AgentUpdateRequest[]; +}>; +type ToClient = + | Readonly<{ kind: "state"; state: State }> + | Readonly<{ kind: "result"; id: number; value: unknown }> + | Readonly<{ kind: "error"; id: number; message: string }>; +type ToService = + | Readonly<{ kind: "call"; id: number; method: Call; args: unknown[] }> + | Readonly<{ kind: "send"; method: Send; args: unknown[] }>; + +/** What the client carries of `AgentsClient` so far. Types, session state and + * the retype preview call into plugin code, which a later change gives the + * screens' page. */ +export type AgentsPortClient = Omit< + AgentsClient, + "types" | "sessionState" | "previewRetype" +>; + +function stateOf(agents: Agents): State { + const snapshot = agents.snapshot(); + const memberships: Record> = {}; + const joinable = agents.joinable(); + // Every agent a screen can show: listed, running, joinable or on a team. + const shown = [ + ...[...snapshot.agents, ...snapshot.running, ...joinable].map( + (agent) => agent.pubkey, + ), + ...snapshot.teams.flatMap((team) => team.members), + ]; + for (const pubkey of shown) + memberships[pubkey] ??= agents.memberships(pubkey); + return { + snapshot, + memberships, + joinable, + interruptible: [ + ...new Set( + snapshot.agents + .filter((agent) => agents.interruptible(agent.pubkey)) + .map((agent) => agent.pubkey), + ), + ], + creationRequests: agents.creationRequests(), + updateRequests: agents.updateRequests(), + }; +} + +/** 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: MessagePort) { + let pending = true; + const flush = () => { + if (!pending) return; + pending = false; + port.postMessage({ + kind: "state", + state: stateOf(agents), + } satisfies ToClient); + }; + // Several changes in one task are one push. + const changed = () => { + if (pending) return; + pending = true; + queueMicrotask(flush); + }; + const unsubscribe = agents.subscribe(changed); + const calls = new Set(CALLS); + const sends = new Set(SENDS); + port.onmessage = ({ data }: MessageEvent) => { + if (data.kind === "send" && sends.has(data.method)) { + (agents[data.method] as (...args: unknown[]) => void)(...data.args); + return; + } + if (data.kind !== "call" || !calls.has(data.method)) return; + const { id, method, args } = data; + void (async () => + (agents[method] as (...args: unknown[]) => Promise)(...args))() + .then( + (value): ToClient => ({ kind: "result", id, value }), + (error: unknown): ToClient => ({ + kind: "error", + id, + message: error instanceof Error ? error.message : String(error), + }), + ) + .then((answer) => { + flush(); + port.postMessage(answer); + }); + }; + flush(); + return () => { + unsubscribe(); + port.onmessage = null; + port.close(); + }; +} + +const EMPTY: readonly never[] = Object.freeze([]); +const STARTING: State = Object.freeze({ + snapshot: Object.freeze({ + status: "loading", + classifier: "unavailable", + classifierKey: "checking", + agents: EMPTY, + running: EMPTY, + interrupted: EMPTY, + teams: EMPTY, + }), + memberships: Object.freeze({}), + joinable: EMPTY, + interruptible: EMPTY, + creationRequests: EMPTY, + updateRequests: EMPTY, +}); + +/** `next`, keeping each part of `prior` it equals, so screens that compare by + * identity see only what changed. Array items with a `key` match by it. */ +export function shared(prior: unknown, next: T): T { + if (Object.is(prior, next) || typeof next !== "object" || next === null) + return next; + const before = prior; + if (Array.isArray(next)) { + const old = Array.isArray(before) ? (before as unknown[]) : []; + const keyed = new Map( + old.flatMap((item) => { + const key = (item as { key?: unknown } | null)?.key; + return typeof key === "string" ? [[key, item] as const] : []; + }), + ); + const items = next.map((item: unknown, index) => { + const key = (item as { key?: unknown } | null)?.key; + return shared( + typeof key === "string" ? keyed.get(key) : old[index], + item, + ); + }); + return ( + Array.isArray(before) && + items.length === old.length && + items.every((item, index) => item === old[index]) + ? before + : Object.freeze(items) + ) as T; + } + const old = + typeof before === "object" && before !== null && !Array.isArray(before) + ? (before as Record) + : undefined; + const entries = Object.entries(next).map( + ([key, value]) => [key, shared(old?.[key], value)] as const, + ); + return ( + old && + entries.length === Object.keys(old).length && + entries.every(([key, value]) => key in old && old[key] === value) + ? old + : Object.freeze(Object.fromEntries(entries)) + ) as T; +} + +/** The agents served on `port`, for screens. Until the first push it reads + * as loading. */ +export function connectAgents(port: MessagePort) { + let state = STARTING; + let next = 0; + let closed = false; + const listeners = new Set<() => void>(); + const waiting = new Map< + number, + Readonly<{ resolve(value: unknown): void; reject(error: Error): void }> + >(); + port.onmessage = ({ data }: MessageEvent) => { + if (data.kind === "state") { + const prior = state; + state = shared(prior, data.state); + if (state !== prior) for (const listener of listeners) listener(); + return; + } + const pending = waiting.get(data.id); + if (!pending) return; + waiting.delete(data.id); + if (data.kind === "result") pending.resolve(data.value); + else pending.reject(new Error(data.message)); + }; + const post = (message: ToService) => { + if (closed) throw new Error("The agents connection is closed."); + port.postMessage(message); + }; + 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); + } + }); + const send = + (method: Send) => + (...args: unknown[]) => { + // Screens pass these as event handlers; the event is not an argument. + post({ + kind: "send", + method, + args: args.filter((arg) => !(arg instanceof Event)), + }); + }; + const calls = Object.fromEntries( + CALLS.map((method) => [method, call(method)]), + ); + const sends = Object.fromEntries( + SENDS.map((method) => [method, send(method)]), + ); + const client = { + ...calls, + ...sends, + snapshot: () => state.snapshot, + subscribe: (listener: () => void) => { + listeners.add(listener); + return () => { + listeners.delete(listener); + }; + }, + find: (pubkey: string) => + state.snapshot.agents.find((agent) => agent.pubkey === pubkey), + classifier: () => state.snapshot.classifier, + memberships: (pubkey: string) => state.memberships[pubkey] ?? EMPTY, + joinable: () => state.joinable, + interruptible: (pubkey: string) => state.interruptible.includes(pubkey), + creationRequests: () => state.creationRequests, + updateRequests: () => state.updateRequests, + } as AgentsPortClient; + return Object.freeze({ + ...client, + close() { + if (closed) return; + closed = true; + port.onmessage = null; + port.close(); + for (const pending of waiting.values()) + pending.reject(new Error("The agents connection is closed.")); + waiting.clear(); + }, + }); +} diff --git a/src/features/agents/service.test.ts b/src/features/agents/service.test.ts index 4eafa41597..c607a2eb3f 100644 --- a/src/features/agents/service.test.ts +++ b/src/features/agents/service.test.ts @@ -12,6 +12,7 @@ import { finalizeEvent, generateSecretKey } from "nostr-tools"; 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 { Mode } from "./agent-start"; import { channelTemplates } from "../channel-templates/local"; import { eventDto, type ReadFilter, type RelayEvent } from "../relay/events"; @@ -2153,6 +2154,39 @@ describe("an agent in several communities", () => { ]); }); + it("serves its screens over a port, with what each call changed there when it returns", async () => { + const { service, select, second } = await setup(); + const { port1, port2 } = new MessageChannel(); + const stop = serveAgents(service, port1); + const client = connectAgents(port2); + try { + const made = await client.create({ type: "example/echo", name: "Echo" }); + expect(made).toMatchObject({ pubkey: bot, name: "Echo" }); + expect(client.find(bot)).toEqual(service.find(bot)); + // An agent nothing changed keeps its identity across pushes. + const shown = client.find(bot); + await client.saveTeam({ name: "Crew", instructions: "", members: [] }); + expect(client.snapshot().teams.map((team) => team.name)).toEqual([ + "Crew", + ]); + expect(client.find(bot)).toBe(shown); + select(second.origin); + await vi.waitFor(() => + expect(client.joinable().map((agent) => agent.key)).toEqual([bot]), + ); + const { agent } = await client.join(bot); + expect(agent.key).toBe(away); + expect(client.joinable()).toEqual([]); + expect(client.memberships(bot)).toEqual(service.memberships(bot)); + expect(client.memberships(bot)).toHaveLength(2); + // The service's refusal, as the screen would show it. + await expect(client.join(bot)).rejects.toThrow("Echo is already here"); + } finally { + client.close(); + stop(); + } + }); + it("leaves one community and stays in its others, never its last, and comes back with its key", async () => { const { service, native, second, agent } = await joined(); await service.save(agent.key, {