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[]) =>