feat(agents): an agents port between webviews over Tauri events

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<typeof eventPort>) {
+  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();
+});

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<void>;
+  listen(
+    event: string,
+    handler: (event: { payload: Frame }) => void,
+  ): Promise<UnlistenFn>;
+}>;
+
+/** 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<number, unknown>();
+  // Peer pages that have since reloaded: their late frames are stale.
+  const retired = new Set<string>();
+  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;
+}

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<typeof fakeAgents>) {
   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");

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,13 +70,22 @@ 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[] }>
   | Readonly<{ kind: "watch"; watch: TypeWatch }>
   | 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<string, readonly [string, SessionPlace]>();
   // 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<string>(CALLS);
   const sends = new Set<string>(SENDS);
   port.onmessage = ({ data }: MessageEvent<ToService>) => {
+    // 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<T>(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[]) =>