From 04214673869d6ef387bd9d88b498cbed712d4655 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 16 Sep 2026 06:53:16 +0100 Subject: [PATCH] fix(bridge): stop the mesh SharedWorker merging patches on top of stale local state The worker's own applyPatch reimplemented a partial CRDT merge for agent_upsert/room_upsert -- unioning the existing local subscribedRooms/ members onto whatever the server sent, rather than replacing them. Since 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 merge) before ever broadcasting a patch, this local merge could only make things worse: it silently resurrected a room or member the server had already removed, because the worker has no version field of its own to know its local copy might already be stale. applyPatch now overwrites directly from whatever the patch carries, with no merge attempt. agent_offline, room_delete, message_add, dm_add, and message_read are untouched -- none of them had this problem. The pure reducer functions (applyPatch, applyStateSync, getStateSnapshot, the four state Maps) are now exported and covered by real unit tests -- this file had none before. The module's own self.addEventListener registration is now guarded on self actually existing, since importing the module directly for testing has no SharedWorkerGlobalScope the way a real browser environment does. --- src/bridges/user/web/frontend/mesh-worker.ts | 104 +++++++--------- .../frontend/test/mesh-worker.unit.test.ts | 111 ++++++++++++++++++ 2 files changed, 155 insertions(+), 60 deletions(-) create mode 100644 src/bridges/user/web/frontend/test/mesh-worker.unit.test.ts diff --git a/src/bridges/user/web/frontend/mesh-worker.ts b/src/bridges/user/web/frontend/mesh-worker.ts index 5479b115..92ac3d1c 100644 --- a/src/bridges/user/web/frontend/mesh-worker.ts +++ b/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(); -const rooms = new Map(); -const messages = new Map(); -const dms = new Map(); +export const agents = new Map(); +export const rooms = new Map(); +export const messages = new Map(); +export const dms = new Map(); -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); + } + }; + }); +} diff --git a/src/bridges/user/web/frontend/test/mesh-worker.unit.test.ts b/src/bridges/user/web/frontend/test/mesh-worker.unit.test.ts new file mode 100644 index 00000000..4ccd9924 --- /dev/null +++ b/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 { + 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 { + 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" }), + ]); + }); +});