From 19a2f1a25184801d6d7650d537fa3e254afe7b6c Mon Sep 17 00:00:00 2001 From: Logan Johnson Date: Tue, 27 Oct 2026 12:17:19 -0400 Subject: [PATCH] feat(agents): an agents port between webviews over Tauri events The agents protocol can now run between pages in two webviews. Frames are numbered and held until their turn, since separate emits can arrive out of order and an answer must not overtake its state push. Each page load is a session: a page announces itself once it listens, holds what it sends until it knows the other page, and drops late frames from a page that reloaded. The screens' client says hello on connect, so a reloaded screens' page gets the current state without waiting for a change. 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/event-port.test.ts | 92 ++++++++++++++++++ src/features/agents/event-port.ts | 124 +++++++++++++++++++++++++ src/features/agents/port.test.ts | 43 +++++++++ src/features/agents/port.ts | 23 ++++- 4 files changed, 279 insertions(+), 3 deletions(-) create mode 100644 src/features/agents/event-port.test.ts create mode 100644 src/features/agents/event-port.ts diff --git a/src/features/agents/event-port.test.ts b/src/features/agents/event-port.test.ts new file mode 100644 index 0000000000..a38928964b --- /dev/null +++ b/src/features/agents/event-port.test.ts @@ -0,0 +1,92 @@ +import { expect, it } from "vitest"; +import { eventPort, type Events, type Frame } from "./event-port"; + +/** Tauri events between webviews, delivered when the test says, each batch + * in reverse, and as JSON. */ +function bus() { + const handlers = new Set<(event: { payload: Frame }) => void>(); + let queue: Frame[] = []; + const events: Events = { + emitTo: async (_webview, _event, frame) => { + queue.push(JSON.parse(JSON.stringify(frame)) as Frame); + }, + listen: async (_event, handler) => { + handlers.add(handler); + return () => handlers.delete(handler); + }, + }; + return { + events, + /** Delivers what was sent, then what that sent, until nothing is. */ + async settle() { + for (let round = 0; round < 20; round += 1) { + for (let tick = 0; tick < 5; tick += 1) await Promise.resolve(); + const batch = queue; + queue = []; + if (!batch.length) continue; + for (const frame of batch.reverse()) + for (const handler of [...handlers]) handler({ payload: frame }); + } + }, + }; +} + +function received(port: ReturnType) { + const messages: unknown[] = []; + port.onmessage = ({ data }) => messages.push(data); + return messages; +} + +it("delivers each message once, in the order sent, though frames arrive out of order", async () => { + const { events, settle } = bus(); + const agents = eventPort("agents", "main", events); + const main = eventPort("main", "agents", events); + const heard = received(main); + await settle(); + for (const n of [1, 2, 3]) agents.postMessage({ n }); + await settle(); + expect(heard).toEqual([{ n: 1 }, { n: 2 }, { n: 3 }]); +}); + +it("holds what a page sends until the other page is up", async () => { + const { events, settle } = bus(); + const main = eventPort("main", "agents", events); + main.postMessage("early"); + await settle(); + const agents = eventPort("agents", "main", events); + const heard = received(agents); + await settle(); + expect(heard).toEqual(["early"]); +}); + +it("talks to a reloaded page afresh, and drops the old page's late frames", async () => { + const { events, settle } = bus(); + const agents = eventPort("agents", "main", events); + const heardByAgents = received(agents); + const before = eventPort("main", "agents", events); + await settle(); + // The old page's last message is still on its way when the page reloads. + before.postMessage("stale"); + before.close(); + const after = eventPort("main", "agents", events); + const heardByMain = received(after); + await settle(); + after.postMessage("fresh"); + agents.postMessage("hello again"); + await settle(); + expect(heardByAgents).toEqual(["fresh"]); + expect(heardByMain).toEqual(["hello again"]); +}); + +it("ignores frames for another webview", async () => { + const { events, settle } = bus(); + const agents = eventPort("agents", "main", events); + const main = eventPort("main", "agents", events); + const huddle = eventPort("huddle", "agents", events); + const heard = received(huddle); + await settle(); + agents.postMessage("for main"); + await settle(); + expect(heard).toEqual([]); + main.close(); +}); diff --git a/src/features/agents/event-port.ts b/src/features/agents/event-port.ts new file mode 100644 index 0000000000..ffbe3732dd --- /dev/null +++ b/src/features/agents/event-port.ts @@ -0,0 +1,124 @@ +// A port between pages in two webviews of this app, over Tauri events. The +// agents protocol needs each message once and in the order sent: a call's +// answer must not overtake the state push before it. Separate emits can +// arrive out of order, so each frame is numbered and held until its turn. +// +// Either page may load first, and either may reload. Each page load is a +// session. A page announces its session once it listens, and answers a new +// peer's announcement with its own; until it knows the peer's session it +// holds what it sends. Frames name the session they are for, so a frame +// meant for a page that has since reloaded is dropped. +import { emitTo, listen, type UnlistenFn } from "@tauri-apps/api/event"; +import type { AgentsPort } from "./port"; + +const EVENT = "buzz://agents-port"; + +export type Frame = Readonly< + { + /** The webviews it is for and from. */ + to: string; + by: string; + /** The sending page load. */ + from: string; + } & ( + | { kind: "ready" } + | { kind: "message"; for: string; seq: number; data: unknown } + ) +>; + +export type Events = Readonly<{ + emitTo(webview: string, event: string, payload: Frame): Promise; + listen( + event: string, + handler: (event: { payload: Frame }) => void, + ): Promise; +}>; + +/** A port from this page, in webview `from`, to the page in webview `to`. + * Messages are JSON: a value JSON cannot carry does not arrive as sent. + * What this page sends before the peer's page is up waits for it; what it + * sent to a peer page that reloaded is lost with that page. */ +export function eventPort( + from: string, + to: string, + events: Events = { emitTo, listen }, +): AgentsPort { + const session = crypto.randomUUID(); + let closed = false; + let peer: string | undefined; + let sent = 0; + let next = 0; + const waiting: unknown[] = []; + const held = new Map(); + // Peer pages that have since reloaded: their late frames are stale. + const retired = new Set(); + const emit = (frame: Frame) => + events.emitTo(to, EVENT, frame).catch((error: unknown) => { + console.error("Agents port message was not sent", error); + }); + const send = (data: unknown) => { + if (peer) + void emit({ + to, + by: from, + from: session, + kind: "message", + for: peer, + seq: sent++, + data, + }); + }; + const port: AgentsPort = { + onmessage: null, + postMessage(data) { + if (closed) return; + if (peer) send(data); + else waiting.push(data); + }, + close() { + closed = true; + waiting.length = 0; + held.clear(); + void unlisten.then((stop) => stop()); + }, + }; + // A new peer page: what was sent to the last one is gone with it. + const meet = (other: string) => { + if (peer) retired.add(peer); + peer = other; + sent = 0; + next = 0; + held.clear(); + for (const data of waiting.splice(0)) send(data); + }; + const unlisten = events + .listen(EVENT, ({ payload: frame }) => { + if (closed || frame.to !== from || frame.by !== to) return; + if (frame.from !== peer && retired.has(frame.from)) return; + if (frame.kind === "ready") { + if (frame.from === peer) return; + meet(frame.from); + void emit({ to, by: from, from: session, kind: "ready" }); + return; + } + if (frame.for !== session) return; + // It knows this page, so its announcement is on the way or was missed. + if (frame.from !== peer) meet(frame.from); + held.set(frame.seq, frame.data); + while (held.has(next) && !closed) { + const data = held.get(next); + held.delete(next); + next += 1; + port.onmessage?.(new MessageEvent("message", { data })); + } + }) + .then((stop) => { + if (!closed) void emit({ to, by: from, from: session, kind: "ready" }); + return stop; + }) + .catch((error: unknown): UnlistenFn => { + console.error("Agents port is not listening", error); + return () => {}; + }); + return port; +} diff --git a/src/features/agents/port.test.ts b/src/features/agents/port.test.ts index 5e9e32c6d2..2dc453c59f 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 { Events, Frame } from "./event-port"; +import { eventPort } from "./event-port"; import { keyedRead } from "./handles"; import { connectAgents, serveAgents, shared } from "./port"; import type { Agents, AgentsSnapshot } from "./service"; @@ -84,6 +86,47 @@ function connected(agents: ReturnType) { return client; } +it("serves screens in another webview, whichever page loads first", async () => { + const handlers = new Set<(event: { payload: Frame }) => void>(); + const events: Events = { + emitTo: async (_webview, _event, frame) => { + const payload = JSON.parse(JSON.stringify(frame)) as Frame; + setTimeout(() => { + for (const handler of [...handlers]) handler({ payload }); + }); + }, + listen: async (_event, handler) => { + handlers.add(handler); + return () => handlers.delete(handler); + }, + }; + const agents = fakeAgents(); + const client = connectAgents(eventPort("main", "agents", events)); + await new Promise((resolve) => setTimeout(resolve, 10)); + const stop = serveAgents( + agents as unknown as Agents, + eventPort("agents", "main", events), + ); + stops.push(() => { + client.close(); + stop(); + }); + await vi.waitFor(() => expect(client.snapshot().status).toBe("ready")); + await expect( + client.saveTeam({ name: "Crew", instructions: "", members: [] }), + ).resolves.toMatchObject({ name: "Crew" }); + expect(client.snapshot().teams.map((team) => team.name)).toEqual(["Crew"]); + // The screens' page reloads; the service has nothing new to push. + client.close(); + const reloaded = connectAgents(eventPort("main", "agents", events)); + stops.push(() => reloaded.close()); + await vi.waitFor(() => + expect(reloaded.snapshot().teams.map((team) => team.name)).toEqual([ + "Crew", + ]), + ); +}); + 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 e9c2434c2a..4be5157e00 100644 --- a/src/features/agents/port.ts +++ b/src/features/agents/port.ts @@ -1,5 +1,5 @@ // 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 +// The service answers over a port; the client holds what screens read // synchronously, as the service last pushed it, and sends everything else. import { type Agent, @@ -70,6 +70,7 @@ type ToClient = | Readonly<{ kind: "result"; id: number; value: unknown }> | Readonly<{ kind: "error"; id: number; message: string }>; type ToService = + | Readonly<{ kind: "hello" }> | Readonly<{ kind: "call"; id: number; method: Call; args: unknown[] }> | Readonly<{ kind: "send"; method: Send; args: unknown[] }> | Readonly<{ kind: "type"; id: number; call: TypeCall; args: unknown[] }> @@ -77,6 +78,14 @@ type ToService = | Readonly<{ kind: "session"; key: string; place: SessionPlace }> | Readonly<{ kind: "abort"; signal: number }>; +/** A `MessagePort`, or a link to a page in another webview. It delivers each + * message once, in the order sent. */ +export type AgentsPort = { + onmessage: ((event: MessageEvent) => void) | null; + postMessage(message: unknown): void; + close(): void; +}; + /** `AgentsClient`, and `present` for what this page shows of each type (by * type key). */ export type AgentsPortClient = AgentsClient & @@ -127,7 +136,7 @@ function stateOf( /** 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) { +export function serveAgents(agents: Agents, port: AgentsPort) { const types = serveTypes(agents); const places = new Map(); // The types' own stores, which change without the service's notice. @@ -159,6 +168,12 @@ export function serveAgents(agents: Agents, port: MessagePort) { const calls = new Set(CALLS); const sends = new Set(SENDS); port.onmessage = ({ data }: MessageEvent) => { + // A screens' page that loaded after the last push needs one now. + if (data.kind === "hello") { + pending = true; + flush(); + return; + } if (data.kind === "send" && sends.has(data.method)) { (agents[data.method] as (...args: unknown[]) => void)(...data.args); return; @@ -283,7 +298,7 @@ export function shared(prior: unknown, next: T): T { /** The agents served on `port`, for screens. Until the first push it reads * as loading. */ -export function connectAgents(port: MessagePort) { +export function connectAgents(port: AgentsPort) { let state = STARTING; let next = 0; let closed = false; @@ -310,6 +325,8 @@ export function connectAgents(port: MessagePort) { if (closed) throw new Error("The agents connection is closed."); port.postMessage(message); }; + // The service may have pushed before this page listened. + post({ kind: "hello" }); const call = (method: Call) => (...args: unknown[]) =>