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" }), + ]); + }); +});