fix(bridge): stop the mesh SharedWorker merging patches on top of stale local state

src/bridges/user/web/frontend/mesh-worker.ts

@@ -35,7 +35,7 @@ declare const self: SharedWorkerGlobalScope;
 // Wire protocol types (inlined — mirrors core/wire-protocol.ts)
 // ---------------------------------------------------------------------------
 
-interface AgentIdentity {
+export interface AgentIdentity {
   id: string;
   name: string;
   harness: string;
@@ -48,7 +48,7 @@ interface AgentIdentity {
   subscribedRooms: string[];
 }
 
-interface Room {
+export interface Room {
   id: string;
   name: string;
   type: "public" | "private" | "secret";
@@ -130,12 +130,12 @@ type WorkerOutbound =
 // Lightweight mesh state
 // ---------------------------------------------------------------------------
 
-const agents = new Map<string, AgentIdentity>();
-const rooms = new Map<string, Room>();
-const messages = new Map<string, RoomMessage[]>();
-const dms = new Map<string, DmMessage[]>();
+export const agents = new Map<string, AgentIdentity>();
+export const rooms = new Map<string, Room>();
+export const messages = new Map<string, RoomMessage[]>();
+export const dms = new Map<string, DmMessage[]>();
 
-function applyStateSync(state: SerialisedState): void {
+export function applyStateSync(state: SerialisedState): void {
   for (const [id, agent] of Object.entries(state.agents)) {
     agents.set(id, agent);
   }
@@ -150,23 +150,14 @@ function applyStateSync(state: SerialisedState): void {
   }
 }
 
-function applyPatch(patch: MeshStatePatch): void {
+/**
+ * Applies one server-sent patch to local state. Never merges with the existing local copy -- the worker holds a single WebSocket connection to exactly one server, and the server has already computed the fully-merged, authoritative record (delivery-engine.ts's own version-gated CRDT merge) before ever broadcasting it, so whatever a patch carries IS the correct state, full stop. Merging on top of a possibly-stale local copy (the previous behaviour) can only make things worse -- e.g. resurrecting a room or member the server already removed, since the worker has no version field to know its own copy might already be the older one.
+ */
+export function applyPatch(patch: MeshStatePatch): void {
   switch (patch.type) {
-    case "agent_upsert": {
-      const existing = agents.get(patch.agent.id);
-      if (existing) {
-        const merged = patch.agent;
-        for (const r of existing.subscribedRooms) {
-          if (!merged.subscribedRooms.includes(r)) {
-            merged.subscribedRooms.push(r);
-          }
-        }
-        agents.set(merged.id, merged);
-      } else {
-        agents.set(patch.agent.id, patch.agent);
-      }
+    case "agent_upsert":
+      agents.set(patch.agent.id, patch.agent);
       break;
-    }
     case "agent_offline": {
       const agent = agents.get(patch.agentId);
       if (agent) {
@@ -175,19 +166,9 @@ function applyPatch(patch: MeshStatePatch): void {
       }
       break;
     }
-    case "room_upsert": {
-      const existing = rooms.get(patch.room.id);
-      if (existing) {
-        const merged = patch.room;
-        for (const m of existing.members) {
-          if (!merged.members.includes(m)) merged.members.push(m);
-        }
-        rooms.set(merged.id, merged);
-      } else {
-        rooms.set(patch.room.id, patch.room);
-      }
+    case "room_upsert":
+      rooms.set(patch.room.id, patch.room);
       break;
-    }
     case "room_delete":
       rooms.delete(patch.roomId);
       break;
@@ -228,7 +209,7 @@ function applyPatch(patch: MeshStatePatch): void {
   }
 }
 
-function getStateSnapshot(): StateSnapshot {
+export function getStateSnapshot(): StateSnapshot {
   return {
     agents: [...agents.values()],
     rooms: [...rooms.values()],
@@ -390,30 +371,33 @@ function isWorkerInbound(value: unknown): value is WorkerInbound {
 // Entry point — listen for SharedWorker connections
 // ---------------------------------------------------------------------------
 
-self.addEventListener("connect", (event: MessageEvent) => {
-  const rawPort = event.ports[0];
-  if (rawPort === undefined) return;
-  // MessagePort satisfies MessagePortLike (has postMessage, close, onmessage)
-  ports.add(rawPort);
-
-  // Send current state to the new port
-  try {
-    rawPort.postMessage(
-      JSON.stringify({
-        type: "state",
-        state: getStateSnapshot(),
-      } satisfies WorkerOutbound),
-    );
-  } catch {
-    // Port not ready yet — will get state on next update
-  }
+/** Registers the real SharedWorker entry point. Guarded on `self` actually existing as a SharedWorkerGlobalScope: this module is imported directly (not just bundled) by unit tests exercising the pure reducer functions above, and a plain Node test environment has no global `self` at all. */
+if (typeof self !== "undefined") {
+  self.addEventListener("connect", (event: MessageEvent) => {
+    const rawPort = event.ports[0];
+    if (rawPort === undefined) return;
+    // MessagePort satisfies MessagePortLike (has postMessage, close, onmessage)
+    ports.add(rawPort);
 
-  rawPort.onmessage = (e: MessageEvent) => {
-    const parsed: unknown = JSON.parse(
-      typeof e.data === "string" ? e.data : String(e.data),
-    );
-    if (isWorkerInbound(parsed)) {
-      handlePortMessage(parsed);
+    // Send current state to the new port
+    try {
+      rawPort.postMessage(
+        JSON.stringify({
+          type: "state",
+          state: getStateSnapshot(),
+        } satisfies WorkerOutbound),
+      );
+    } catch {
+      // Port not ready yet — will get state on next update
     }
-  };
-});
+
+    rawPort.onmessage = (e: MessageEvent) => {
+      const parsed: unknown = JSON.parse(
+        typeof e.data === "string" ? e.data : String(e.data),
+      );
+      if (isWorkerInbound(parsed)) {
+        handlePortMessage(parsed);
+      }
+    };
+  });
+}

src/bridges/user/web/frontend/test/mesh-worker.unit.test.ts

@@ -0,0 +1,111 @@
+/**
+ * Unit tests for mesh-worker.ts's own local reducer -- applyPatch/applyStateSync/getStateSnapshot. The worker connects to exactly one server over one WebSocket (never multiple mesh peers directly), and the server has already computed the fully-merged, authoritative record before ever broadcasting a patch -- so unlike the server's own delivery-engine.ts, which genuinely needs version-gated CRDT merge to reconcile concurrent writes from multiple peers, the worker needs none of that: applying a patch should simply overwrite with whatever the server sent.
+ */
+import { describe, it, expect, beforeEach } from "vitest";
+import {
+  agents,
+  rooms,
+  applyPatch,
+  applyStateSync,
+  getStateSnapshot,
+} from "../mesh-worker.js";
+import type { AgentIdentity, Room } from "../mesh-worker.js";
+
+function agent(overrides: Partial<AgentIdentity> = {}): AgentIdentity {
+  return {
+    id: "agent-1",
+    name: "agent-name",
+    harness: "pi",
+    cwd: "/tmp",
+    pid: 1,
+    startedAt: "2026-01-01T00:00:00.000Z",
+    visibility: "visible",
+    status: "active",
+    tags: [],
+    subscribedRooms: [],
+    ...overrides,
+  };
+}
+
+function room(overrides: Partial<Room> = {}): Room {
+  return {
+    id: "room-1",
+    name: "room-name",
+    type: "public",
+    owner: "owner-1",
+    createdAt: "2026-01-01T00:00:00.000Z",
+    description: "",
+    members: [],
+    invited: [],
+    ...overrides,
+  };
+}
+
+beforeEach(() => {
+  agents.clear();
+  rooms.clear();
+});
+
+describe("mesh-worker applyPatch(agent_upsert)", () => {
+  it("overwrites subscribedRooms exactly as the server sent it, never resurrecting a room the server already removed", () => {
+    agents.set(
+      "agent-1",
+      agent({ id: "agent-1", subscribedRooms: ["room-a", "room-b"] }),
+    );
+
+    applyPatch({
+      type: "agent_upsert",
+      agent: agent({ id: "agent-1", subscribedRooms: [] }),
+    });
+
+    expect(agents.get("agent-1")?.subscribedRooms).toEqual([]);
+  });
+
+  it("stores a brand-new agent as-is", () => {
+    applyPatch({
+      type: "agent_upsert",
+      agent: agent({ id: "agent-2", name: "fresh" }),
+    });
+    expect(agents.get("agent-2")?.name).toBe("fresh");
+  });
+});
+
+describe("mesh-worker applyPatch(room_upsert)", () => {
+  it("overwrites members exactly as the server sent it, never resurrecting a member the server already removed", () => {
+    rooms.set(
+      "room-1",
+      room({ id: "room-1", members: ["member-a", "member-b"] }),
+    );
+
+    applyPatch({
+      type: "room_upsert",
+      room: room({ id: "room-1", members: ["member-a"] }),
+    });
+
+    expect(rooms.get("room-1")?.members).toEqual(["member-a"]);
+  });
+
+  it("stores a brand-new room as-is", () => {
+    applyPatch({
+      type: "room_upsert",
+      room: room({ id: "room-2", name: "fresh" }),
+    });
+    expect(rooms.get("room-2")?.name).toBe("fresh");
+  });
+});
+
+describe("mesh-worker applyStateSync / getStateSnapshot", () => {
+  it("populates local state from a full snapshot, readable back via getStateSnapshot", () => {
+    applyStateSync({
+      agents: { "agent-1": agent({ id: "agent-1", name: "synced" }) },
+      rooms: { "room-1": room({ id: "room-1", name: "synced-room" }) },
+      messages: {},
+      dms: {},
+    });
+    const snapshot = getStateSnapshot();
+    expect(snapshot.agents).toEqual([agent({ id: "agent-1", name: "synced" })]);
+    expect(snapshot.rooms).toEqual([
+      room({ id: "room-1", name: "synced-room" }),
+    ]);
+  });
+});