From 28dda318a3f6fbf8add7c269f96fc02fc3eb1664 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:18:18 +0100 Subject: [PATCH 1/8] feat(core): add CoordinatorGateway connection-lifecycle collaborator Dials the hub when a bridge becomes the local mesh coordinator and drops the connection when it loses that role, idempotently. Not yet wired into MeshStore/PeerLifecycle or the transport's own hub session. --- src/core/coordinator-gateway.ts | 37 +++++++++++ src/test/coordinator-gateway.test.ts | 94 ++++++++++++++++++++++++++++ 2 files changed, 131 insertions(+) create mode 100644 src/core/coordinator-gateway.ts create mode 100644 src/test/coordinator-gateway.test.ts diff --git a/src/core/coordinator-gateway.ts b/src/core/coordinator-gateway.ts new file mode 100644 index 00000000..d48a4bd0 --- /dev/null +++ b/src/core/coordinator-gateway.ts @@ -0,0 +1,37 @@ +/** + * CoordinatorGateway — attaches the cross-machine gateway role to this machine's local coordinator (agent-comms#154, agent-comms#153's first leg). A bridge that becomes the mesh's local coordinator, whether by a fresh bind (MeshStore's own init()) or a takeover (PeerLifecycle's own handleBecomeCoordinator), also becomes the machine's gateway: it dials the hub and holds the connection for as long as it holds the coordinator role. Losing the role, gracefully or by crash, drops the connection; the next coordinator re-dials as part of taking over. Hub-side state is therefore rebuilt from scratch on every takeover -- messages in flight during the gap are lost, the same loss class as a coordinator crash today, now on the data path. Forwarding local agents onto the hub and merging its directory back (agent-comms#155) is deliberately not this class's concern; it owns only the connection lifecycle. + */ + +export interface CoordinatorGatewayDeps { + /** The hub URL this machine's gateway dials -- configuration, defaulting to DEFAULT_HUB_URL (mesh-store-shared.ts). */ + hubUrl: string; + /** Dials the hub. Backed by the transport's own optional connectHub -- a transport with no gateway capability is never asked to redial by anything else in this class. */ + connectHub: (url: string) => Promise; + /** Drops the held hub connection, if any. Backed by the transport's own optional disconnectHub. */ + disconnectHub: () => Promise; +} + +export class CoordinatorGateway { + private connected = false; + + constructor(private readonly deps: Readonly) {} + + /** Whether this side currently holds the gateway role: it has dialled the hub and has not since lost coordinator status. */ + get isConnected(): boolean { + return this.connected; + } + + /** Dials the hub for this machine's gateway role. Idempotent: a call while already connected is a no-op, since nothing in this codebase's own coordinator-election machinery re-fires "became coordinator" without an intervening onLostCoordinator -- this guard is defensive, not a known double-fire path. */ + async onBecameCoordinator(): Promise { + if (this.connected) return; + this.connected = true; + await this.deps.connectHub(this.deps.hubUrl); + } + + /** Drops this machine's held hub connection. A no-op if this side never became the gateway, or already lost the role -- MeshStore.shutdown() calls this unconditionally regardless of coordinator status, so this guard is what makes that safe rather than a redundant extra close. */ + async onLostCoordinator(): Promise { + if (!this.connected) return; + this.connected = false; + await this.deps.disconnectHub(); + } +} diff --git a/src/test/coordinator-gateway.test.ts b/src/test/coordinator-gateway.test.ts new file mode 100644 index 00000000..78389337 --- /dev/null +++ b/src/test/coordinator-gateway.test.ts @@ -0,0 +1,94 @@ +/** + * Direct unit tests for CoordinatorGateway -- the connect-hub-on-become-coordinator / disconnect-hub-on-lose-coordinator wiring (agent-comms#154), tested against injected connectHub/disconnectHub closures rather than a real transport or hub socket, mirroring peer-lifecycle.test.ts's own DI-based approach. + */ +import { describe, expect, it, vi } from "vitest"; +import { + CoordinatorGateway, + type CoordinatorGatewayDeps, +} from "../core/coordinator-gateway.js"; + +const HUB_URL = "wss://mesh.example.test/"; + +function makeHarness(): { + gateway: CoordinatorGateway; + connectHub: ReturnType; + disconnectHub: ReturnType; +} { + const connectHub = vi.fn().mockResolvedValue(undefined); + const disconnectHub = vi.fn().mockResolvedValue(undefined); + const deps: CoordinatorGatewayDeps = { + hubUrl: HUB_URL, + connectHub, + disconnectHub, + }; + return { gateway: new CoordinatorGateway(deps), connectHub, disconnectHub }; +} + +describe("CoordinatorGateway — onBecameCoordinator", () => { + it("dials the configured hub URL", async () => { + const { gateway, connectHub } = makeHarness(); + + await gateway.onBecameCoordinator(); + + expect(connectHub).toHaveBeenCalledWith(HUB_URL); + expect(connectHub).toHaveBeenCalledTimes(1); + }); + + it("marks isConnected true once dialled", async () => { + const { gateway } = makeHarness(); + expect(gateway.isConnected).toBe(false); + + await gateway.onBecameCoordinator(); + + expect(gateway.isConnected).toBe(true); + }); + + it("is idempotent -- a second call while already connected does not redial", async () => { + const { gateway, connectHub } = makeHarness(); + + await gateway.onBecameCoordinator(); + await gateway.onBecameCoordinator(); + + expect(connectHub).toHaveBeenCalledTimes(1); + }); +}); + +describe("CoordinatorGateway — onLostCoordinator", () => { + it("drops the hub connection when one is held", async () => { + const { gateway, disconnectHub } = makeHarness(); + await gateway.onBecameCoordinator(); + + await gateway.onLostCoordinator(); + + expect(disconnectHub).toHaveBeenCalledTimes(1); + expect(gateway.isConnected).toBe(false); + }); + + it("is a no-op when this side never became the gateway", async () => { + const { gateway, disconnectHub } = makeHarness(); + + await gateway.onLostCoordinator(); + + expect(disconnectHub).not.toHaveBeenCalled(); + }); + + it("is a no-op on a second call after already losing the role", async () => { + const { gateway, disconnectHub } = makeHarness(); + await gateway.onBecameCoordinator(); + await gateway.onLostCoordinator(); + + await gateway.onLostCoordinator(); + + expect(disconnectHub).toHaveBeenCalledTimes(1); + }); + + it("allows redialling after losing and regaining the role", async () => { + const { gateway, connectHub } = makeHarness(); + await gateway.onBecameCoordinator(); + await gateway.onLostCoordinator(); + + await gateway.onBecameCoordinator(); + + expect(connectHub).toHaveBeenCalledTimes(2); + }); +}); From 5064b6d49d6c01c27adee3e6b48525767684e176 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:21:22 +0100 Subject: [PATCH 2/8] test(core): extract real-hub-over-ws test harness into hub-helpers realHubOverWs and waitForCondition were private to the one integration test that needed them; a second hub integration test for the coordinator gateway needs the identical harness plus a connection count observable on the hub side, so it now lives in the shared helpers module both tests import. --- src/test/hub-helpers.ts | 149 +++++++++++++++++- src/test/hub-mode-session.integration.test.ts | 142 +---------------- 2 files changed, 148 insertions(+), 143 deletions(-) diff --git a/src/test/hub-helpers.ts b/src/test/hub-helpers.ts index 072afe51..815c10cf 100644 --- a/src/test/hub-helpers.ts +++ b/src/test/hub-helpers.ts @@ -1,7 +1,152 @@ -// Shared helpers for the hub-mode integration test: canonical-CBOR bytes for frames crossing the test's own ws bridge. -import { cdeEncodeOptions, encode } from "cbor2"; +// Shared helpers for hub-mode integration tests: a real wire-mesh relay hub (createRelayHub -- the same domain logic the production mesh.exadev.io Durable Object runs) served over local WebSockets, canonical-CBOR framing for the ws bridge, and a condition-polling helper for the async discovery/gossip timing these tests exercise. + +import { createServer, type Server } from "node:http"; +import { WebSocketServer, type WebSocket as WsSocket } from "ws"; +import { cdeDecodeOptions, cdeEncodeOptions, decode, encode } from "cbor2"; +import { frameSchema } from "wire-mesh-core/generated/protocol"; +import { createRelayHub } from "wire-mesh-core/domain/relay-hub"; +import type { Connection } from "wire-mesh-core/ports/transport"; import type { Frame } from "wire-mesh-core/generated/protocol"; +const SHUTDOWN_GRACE_MS = 250; +const CLOSE_NORMAL = 1000; // RFC 6455 normal closure +const POLL_INTERVAL_MS = 25; +const DEFAULT_CONDITION_TIMEOUT_MS = 5_000; + export function cbor2ToBytes(frame: Frame): Uint8Array { return new Uint8Array(encode(frame, cdeEncodeOptions)); } + +/** ws's own RawData type is `Buffer | ArrayBuffer | Buffer[]`, but every caller here sets `socket.binaryType = "arraybuffer"` before this ever fires, so a real message is always an ArrayBuffer at runtime -- narrowed with a guard rather than asserted, since ws's declared type is genuinely broader than what this specific binaryType configuration guarantees. */ +function isArrayBuffer(data: unknown): data is ArrayBuffer { + return data instanceof ArrayBuffer; +} + +function wsReceiveStream(socket: WsSocket): AsyncIterable { + const pending: Frame[] = []; + const waiters: { + resolve: (result: IteratorResult) => void; + reject: (error: unknown) => void; + }[] = []; + let ended = false; + socket.on("message", (data) => { + if (!isArrayBuffer(data)) return; + try { + const decoded: unknown = decode(new Uint8Array(data), cdeDecodeOptions); + const parsed = frameSchema.safeParse(decoded); + if (!parsed.success) return; + const waiter = waiters.shift(); + if (waiter !== undefined) { + waiter.resolve({ value: parsed.data, done: false }); + } else { + pending.push(parsed.data); + } + } catch { + // Undecodable: drop, keeping the stream alive for the well-formed frames behind it (the hub's own tolerance). + } + }); + socket.on("close", () => { + ended = true; + for (const waiter of waiters.splice(0)) { + waiter.resolve({ value: undefined, done: true }); + } + }); + socket.on("error", () => { + ended = true; + for (const waiter of waiters.splice(0)) { + waiter.resolve({ value: undefined, done: true }); + } + }); + return { + [Symbol.asyncIterator]() { + return { + async next(): Promise> { + const next = pending.shift(); + if (next !== undefined) { + return { value: next, done: false }; + } + if (ended) { + return { value: undefined, done: true }; + } + return new Promise>((resolve, reject) => { + waiters.push({ resolve, reject }); + }); + }, + }; + }, + }; +} + +function wsConnection(socket: WsSocket): Connection { + return { + // async with no internal await, matching ws-dial.ts's own identical send() shape: promise-function-async requires the interface's Promise return stay async, which is exactly what makes it have nothing to await. + send: async (frame: Frame) => { + socket.send(cbor2ToBytes(frame)); + }, + receive: () => wsReceiveStream(socket), + close: async () => { + socket.close(CLOSE_NORMAL); + }, + }; +} + +/** A real relay hub served over ws: each accepted socket is wrapped as a Connection for createRelayHub, exactly what the production Durable Object does (pre-hibernation shape). Also exposes the hub's own live-connection count, since a test asserting a client-side disconnect actually reached the far end needs an observable on the hub itself, not just the client. */ +export async function realHubOverWs(): Promise<{ + url: string; + connectionCount: () => number; + close: () => Promise; +}> { + const http: Server = createServer(); + const wss = new WebSocketServer({ server: http }); + const hub = createRelayHub(); + wss.on("connection", (socket: WsSocket) => { + socket.binaryType = "arraybuffer"; + void hub.handleConnection(wsConnection(socket)); + }); + await new Promise((resolve) => { + http.listen(0, "127.0.0.1", () => { + resolve(); + }); + }); + const address = http.address(); + if (address === null || typeof address === "string") { + throw new Error("expected a TCP listen address"); + } + return { + url: `ws://127.0.0.1:${String(address.port)}/`, + connectionCount: () => wss.clients.size, + close: async () => + new Promise((resolve) => { + for (const client of wss.clients) { + client.terminate(); + } + wss.close(); + http.close(() => { + resolve(); + }); + setTimeout(resolve, SHUTDOWN_GRACE_MS); + }), + }; +} + +/** Polls `condition` until it's true or `timeoutMs` elapses, rejecting on timeout -- the shared shape every hub integration test uses to wait out real async gossip/discovery/close propagation rather than asserting immediately after firing an action. */ +export async function waitForCondition( + condition: () => boolean, + timeoutMs = DEFAULT_CONDITION_TIMEOUT_MS, +): Promise { + return new Promise((resolve, reject) => { + const started = Date.now(); + const check = (): void => { + if (condition()) { + resolve(); + return; + } + if (Date.now() - started > timeoutMs) { + reject(new Error("condition not met within timeout")); + return; + } + setTimeout(check, POLL_INTERVAL_MS); + }; + check(); + }); +} diff --git a/src/test/hub-mode-session.integration.test.ts b/src/test/hub-mode-session.integration.test.ts index f8cc941a..0259bab5 100644 --- a/src/test/hub-mode-session.integration.test.ts +++ b/src/test/hub-mode-session.integration.test.ts @@ -1,23 +1,12 @@ // Integration: two real agent-comms WireMeshTransports discovering each other and exchanging messages through a real wire-mesh relay hub (createRelayHub -- the same domain logic the production mesh.exadev.io Durable Object runs) served over local WebSockets. This is agent-comms#151's own acceptance shape: the hub connection is a relay, not a coordinator -- no connect_request/introduce approval applies, peers discover each other via the hub's gossip forwarding + catch-up, and messages ride relay pairings (sendManageRequest's own targetDevice routing). -import { createServer, type Server } from "node:http"; -import { WebSocketServer, type WebSocket as WsSocket } from "ws"; import { afterEach, describe, expect, it } from "vitest"; -import { cdeDecodeOptions, decode } from "cbor2"; -import { frameSchema } from "wire-mesh-core/generated/protocol"; -import { cbor2ToBytes } from "./hub-helpers.js"; -import { createRelayHub } from "wire-mesh-core/domain/relay-hub"; -import type { Connection } from "wire-mesh-core/ports/transport"; -import type { Frame } from "wire-mesh-core/generated/protocol"; +import { realHubOverWs, waitForCondition } from "./hub-helpers.js"; import { generateIdentity } from "../core/identity.js"; import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; import { WireMeshTransport } from "../core/wire-mesh-transport.js"; import type { TransportEvents } from "../core/transport.js"; -const SHUTDOWN_GRACE_MS = 250; -const CLOSE_NORMAL = 1000; // RFC 6455 normal closure -const POLL_INTERVAL_MS = 25; - function recordingEvents(): TransportEvents & { messages: { from: string; text: string }[]; } { @@ -41,114 +30,6 @@ function recordingEvents(): TransportEvents & { }; } -/** A real relay hub served over ws: each accepted socket is wrapped as a Connection for createRelayHub, exactly what the production Durable Object does (pre-hibernation shape). */ -async function realHubOverWs(): Promise<{ - url: string; - close: () => Promise; -}> { - const http: Server = createServer(); - const wss = new WebSocketServer({ server: http }); - const hub = createRelayHub(); - wss.on("connection", (socket: WsSocket) => { - socket.binaryType = "arraybuffer"; - void hub.handleConnection(wsConnection(socket)); - }); - await new Promise((resolve) => { - http.listen(0, "127.0.0.1", () => { - resolve(); - }); - }); - const address = http.address(); - if (address === null || typeof address === "string") { - throw new Error("expected a TCP listen address"); - } - return { - url: `ws://127.0.0.1:${String(address.port)}/`, - close: async () => - new Promise((resolve) => { - for (const client of wss.clients) { - client.terminate(); - } - wss.close(); - http.close(() => { - resolve(); - }); - setTimeout(resolve, SHUTDOWN_GRACE_MS); - }), - }; -} - -function wsConnection(socket: WsSocket): Connection { - return { - send: async (frame: Frame) => { - socket.send(cbor2ToBytes(frame)); - }, - receive: () => wsReceiveStream(socket), - close: async () => { - socket.close(CLOSE_NORMAL); - }, - }; -} - -function wsReceiveStream(socket: WsSocket): AsyncIterable { - const pending: Frame[] = []; - const waiters: { - resolve: (result: IteratorResult) => void; - reject: (error: unknown) => void; - }[] = []; - let ended = false; - socket.on("message", (data) => { - void (async () => { - try { - const decoded: unknown = decode( - new Uint8Array(data as ArrayBuffer), - cdeDecodeOptions, - ); - const parsed = frameSchema.safeParse(decoded); - if (!parsed.success) return; - const waiter = waiters.shift(); - if (waiter !== undefined) { - waiter.resolve({ value: parsed.data, done: false }); - } else { - pending.push(parsed.data); - } - } catch { - // Undecodable: drop, keeping the stream alive for the well-formed frames behind it (the hub's own tolerance). - } - })(); - }); - socket.on("close", () => { - ended = true; - for (const waiter of waiters.splice(0)) { - waiter.resolve({ value: undefined, done: true }); - } - }); - socket.on("error", () => { - ended = true; - for (const waiter of waiters.splice(0)) { - waiter.resolve({ value: undefined, done: true }); - } - }); - return { - [Symbol.asyncIterator]() { - return { - async next(): Promise> { - const next = pending.shift(); - if (next !== undefined) { - return { value: next, done: false }; - } - if (ended) { - return { value: undefined, done: true }; - } - return new Promise>((resolve, reject) => { - waiters.push({ resolve, reject }); - }); - }, - }; - }, - }; -} - const cleanups: (() => Promise)[] = []; afterEach(async () => { @@ -197,24 +78,3 @@ describe("connectToHub", () => { await transportB.shutdown(); }); }); - -async function waitForCondition( - condition: () => boolean, - timeoutMs = 5_000, -): Promise { - return new Promise((resolve, reject) => { - const started = Date.now(); - const check = (): void => { - if (condition()) { - resolve(); - return; - } - if (Date.now() - started > timeoutMs) { - reject(new Error("condition not met within timeout")); - return; - } - setTimeout(check, POLL_INTERVAL_MS); - }; - check(); - }); -} From 3e6ef1713c5f5760ce743221dc4ae9cf496a4a04 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:27:00 +0100 Subject: [PATCH 3/8] feat(core): add connectHub/disconnectHub to the transport contract WireMeshTransport.connectHub/disconnectHub delegate to the transport's own HubSession, which now exposes isConnected and disconnect() so a caller can observe and end the held hub session without reaching into private state. Both methods are optional on MeshTransport, matching listKnownDevices' existing precedent for a WireMeshTransport-only capability. Extracts mergeKnownDevices out of WireMeshTransport into its own gossip-directory module to keep the file under the repo's max-lines cap after this addition, mirroring the existing peer-lifecycle.ts / connection-approval.ts / room-router.ts / hub-session.ts splits. --- src/core/gossip-directory.ts | 24 +++++++ src/core/hub-session.ts | 13 ++++ src/core/transport.ts | 10 +++ src/core/wire-mesh-transport.ts | 29 ++++---- src/test/gossip-directory.test.ts | 89 ++++++++++++++++++++++++ src/test/wire-mesh-transport-hub.test.ts | 56 +++++++++++++++ 6 files changed, 206 insertions(+), 15 deletions(-) create mode 100644 src/core/gossip-directory.ts create mode 100644 src/test/gossip-directory.test.ts create mode 100644 src/test/wire-mesh-transport-hub.test.ts diff --git a/src/core/gossip-directory.ts b/src/core/gossip-directory.ts new file mode 100644 index 00000000..34c3d2b5 --- /dev/null +++ b/src/core/gossip-directory.ts @@ -0,0 +1,24 @@ +/** + * mergeKnownDevices — the mesh-wide gossip-directory aggregation WireMeshTransport's own listKnownDevices reads from. Split out purely to keep wire-mesh-transport.ts under the repo's max-lines cap, the same reason connection-approval.ts, room-router.ts, hub-session.ts, and peer-lifecycle.ts were each split from their own owning file. + */ + +import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; +import type { DirectoryEntry } from "wire-mesh-core/domain/mesh-session"; +import type { PeerAdvert } from "wire-mesh-core/generated/protocol"; + +/** Merges one session event's own directory into the mesh-wide knownDevices view (mutated in place), keeping the newer advert (by snapshot-seconds) whenever a device-id is already known from an earlier event or a different session. */ +export function mergeKnownDevices( + knownDevices: Map, + directory: readonly DirectoryEntry[], +): void { + for (const entry of directory) { + const deviceIdHex = deviceIdToHex(entry.device); + const existing = knownDevices.get(deviceIdHex); + if ( + existing === undefined || + entry.advert["snapshot-seconds"] >= existing["snapshot-seconds"] + ) { + knownDevices.set(deviceIdHex, entry.advert); + } + } +} diff --git a/src/core/hub-session.ts b/src/core/hub-session.ts index d2acbbe5..5c307af6 100644 --- a/src/core/hub-session.ts +++ b/src/core/hub-session.ts @@ -45,6 +45,19 @@ export class HubSession { return deviceIdToHex(identity.deviceId); } + /** Whether a hub session is currently live (connect() has resolved and disconnect() hasn't run since, and the far end hasn't closed it -- watchDisconnect clears this.session when the hub's own event stream reports closed). */ + get isConnected(): boolean { + return this.session !== undefined; + } + + /** Drops the held hub connection. A no-op if none is live (connect() was never called, disconnect() already ran, or the hub itself already closed the session). */ + async disconnect(): Promise { + const session = this.session; + if (session === undefined) return; + this.session = undefined; + await session.close(); + } + /** The device ids (hex) of peers discovered through the hub's gossiped directory. */ peers(): readonly string[] { return [...this.hubPeersKnown]; diff --git a/src/core/transport.ts b/src/core/transport.ts index f3763564..f0334289 100644 --- a/src/core/transport.ts +++ b/src/core/transport.ts @@ -259,4 +259,14 @@ export interface MeshTransport { deviceId: string; advert: Readonly>; }[]; + + /** + * Dials the relay hub at the given URL and holds the connection (agent-comms#154's own gateway role, riding this side's HubSession -- see hub-session.ts's class doc for the connection model). Optional: WireMeshTransport is the only implementation that offers it today, matching listKnownDevices' own precedent, so a caller (CoordinatorGateway) must treat its absence as "this transport has no gateway capability," never assume every MeshTransport supports it. + */ + connectHub?: (url: string) => Promise; + + /** + * Drops this side's own held hub connection, if any. A no-op when none is live. Same optionality caveat as connectHub. + */ + disconnectHub?: () => Promise; } diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 15ca1842..63419912 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -11,6 +11,7 @@ import { createTlsTransport } from "wire-mesh-core/adapters/tls-transport"; import { connectWsUrl } from "./ws-dial.js"; import { HubSession } from "./hub-session.js"; +import { mergeKnownDevices } from "./gossip-directory.js"; import { acceptMeshSession, type AcceptedMeshSession, @@ -207,20 +208,6 @@ export class WireMeshTransport implements MeshTransport { // -- Every device-id this side has ever heard gossip from, across every session's own directory, keyed by device-id hex -- the mesh-wide aggregation P3.8's own room-discovery design and the eventual agent register/update/offline retirement both need and don't otherwise have (agent-comms#48's own 2026-09-14 investigation confirmed no such aggregation existed anywhere in this file). Merged, never cleared on disconnect: a device's last-known advert (including its own presence/status, or any future gossiped extension) stays queryable even while its session is momentarily down, the same way the legacy agents Map keeps a record after setAgentOffline rather than deleting it outright. private readonly knownDevices = new Map(); - /** Merges one session event's own directory into the mesh-wide knownDevices view, keeping the newer advert (by snapshot-seconds) whenever this device-id is already known from an earlier event or a different session. */ - private mergeKnownDevices(directory: readonly DirectoryEntry[]): void { - for (const entry of directory) { - const deviceIdHex = deviceIdToHex(entry.device); - const existing = this.knownDevices.get(deviceIdHex); - if ( - existing === undefined || - entry.advert["snapshot-seconds"] >= existing["snapshot-seconds"] - ) { - this.knownDevices.set(deviceIdHex, entry.advert); - } - } - } - /** Every device this side has ever heard gossip from, mesh-wide -- not just its own directly-connected peers -- with each one's own latest full advert (addresses, snapshot-seconds, and every open-extension field such as presence/status). */ listKnownDevices(): readonly { deviceId: string; @@ -566,7 +553,7 @@ export class WireMeshTransport implements MeshTransport { ): void { void (async () => { for await (const event of session.events) { - this.mergeKnownDevices(event.directory); + mergeKnownDevices(this.knownDevices, event.directory); this.reportPresenceAdvert(handle, deviceIdHex, event.directory); if (event.state.status === "closed") { const wasTracked = this.peerSessions.get(deviceIdHex) === session; @@ -960,6 +947,18 @@ export class WireMeshTransport implements MeshTransport { .map((tracked) => `${tracked.host}:${String(tracked.port)}`); } + // ----------------------------------------------------------------------- + // MeshTransport -- Hub (gateway role, agent-comms#154) + // ----------------------------------------------------------------------- + + async connectHub(url: string): Promise { + await this.hub.connect(url); + } + + async disconnectHub(): Promise { + await this.hub.disconnect(); + } + // ----------------------------------------------------------------------- // MeshTransport -- Shutdown / unref // ----------------------------------------------------------------------- diff --git a/src/test/gossip-directory.test.ts b/src/test/gossip-directory.test.ts new file mode 100644 index 00000000..66cd7c31 --- /dev/null +++ b/src/test/gossip-directory.test.ts @@ -0,0 +1,89 @@ +/** + * Direct unit tests for mergeKnownDevices, extracted from WireMeshTransport (see gossip-directory.ts's own header) -- previously only exercised indirectly through gossip-directory-aggregation.integration.test.ts's real-session harness. + */ +import { describe, expect, it } from "vitest"; +import { mergeKnownDevices } from "../core/gossip-directory.js"; +import type { DirectoryEntry } from "wire-mesh-core/domain/mesh-session"; +import type { PeerAdvert } from "wire-mesh-core/generated/protocol"; + +const DEVICE_ID_HEX_LENGTH = 64; +const DEVICE_A_HEX = "a".repeat(DEVICE_ID_HEX_LENGTH); +const DEVICE_B_HEX = "b".repeat(DEVICE_ID_HEX_LENGTH); +/** An arbitrary "later" snapshot-seconds value, distinct from every other one used in this file's fixtures. */ +const NEWER_SNAPSHOT_SECONDS = 5; + +function deviceIdBytes(hex: string): Uint8Array { + const bytes = new Uint8Array(new ArrayBuffer(hex.length / 2)); + for (let i = 0; i < bytes.length; i += 1) { + bytes[i] = Number.parseInt(hex.slice(i * 2, i * 2 + 2), 16); + } + return bytes; +} + +function advert(snapshotSeconds: number): PeerAdvert { + return { + addresses: [], + "snapshot-seconds": snapshotSeconds, + } as unknown as PeerAdvert; +} + +function entry(hex: string, snapshotSeconds: number): DirectoryEntry { + return { device: deviceIdBytes(hex), advert: advert(snapshotSeconds) }; +} + +describe("mergeKnownDevices", () => { + it("records a device this map has never seen before", () => { + const knownDevices = new Map(); + + mergeKnownDevices(knownDevices, [entry(DEVICE_A_HEX, 1)]); + + expect(knownDevices.get(DEVICE_A_HEX)?.["snapshot-seconds"]).toBe(1); + }); + + it("replaces an existing entry with a strictly newer advert", () => { + const knownDevices = new Map([ + [DEVICE_A_HEX, advert(1)], + ]); + + mergeKnownDevices(knownDevices, [entry(DEVICE_A_HEX, 2)]); + + expect(knownDevices.get(DEVICE_A_HEX)?.["snapshot-seconds"]).toBe(2); + }); + + it("keeps an equal-snapshot advert as the incoming one (>=, not >)", () => { + const knownDevices = new Map([ + [DEVICE_A_HEX, advert(1)], + ]); + const incoming = advert(1); + + mergeKnownDevices(knownDevices, [ + { device: deviceIdBytes(DEVICE_A_HEX), advert: incoming }, + ]); + + expect(knownDevices.get(DEVICE_A_HEX)).toBe(incoming); + }); + + it("never regresses an existing entry to an older advert", () => { + const knownDevices = new Map([ + [DEVICE_A_HEX, advert(NEWER_SNAPSHOT_SECONDS)], + ]); + + mergeKnownDevices(knownDevices, [entry(DEVICE_A_HEX, 1)]); + + expect(knownDevices.get(DEVICE_A_HEX)?.["snapshot-seconds"]).toBe( + NEWER_SNAPSHOT_SECONDS, + ); + }); + + it("merges multiple distinct devices from the same directory independently", () => { + const knownDevices = new Map(); + + mergeKnownDevices(knownDevices, [ + entry(DEVICE_A_HEX, 1), + entry(DEVICE_B_HEX, 2), + ]); + + expect(knownDevices.get(DEVICE_A_HEX)?.["snapshot-seconds"]).toBe(1); + expect(knownDevices.get(DEVICE_B_HEX)?.["snapshot-seconds"]).toBe(2); + }); +}); diff --git a/src/test/wire-mesh-transport-hub.test.ts b/src/test/wire-mesh-transport-hub.test.ts new file mode 100644 index 00000000..bdb9e0a9 --- /dev/null +++ b/src/test/wire-mesh-transport-hub.test.ts @@ -0,0 +1,56 @@ +/** + * WireMeshTransport connectHub/disconnectHub (agent-comms#154) -- split out of wire-mesh-transport.test.ts to stay under this repo's max-lines cap, the same reason wire-mesh-transport-shutdown-unref.test.ts was split. Uses the real relay-hub-over-ws harness (hub-helpers.ts) rather than a mocked HubSession, since the behaviour under test is specifically that connectHub/disconnectHub reach the real HubSession instance the transport constructs for itself. + */ + +import { test, describe, expect } from "vitest"; +import { generateIdentity } from "../core/identity.js"; +import { WireMeshTransport } from "../core/wire-mesh-transport.js"; +import type { TransportEvents } from "../core/transport.js"; +import { realHubOverWs, waitForCondition } from "./hub-helpers.js"; + +function noopEvents(): TransportEvents { + return { + onMessage: () => undefined, + onPeerConnected: () => undefined, + onPeerDisconnected: () => undefined, + onIntroduction: () => undefined, + onConnectionRequest: () => undefined, + onPeerList: () => undefined, + onPeerJoined: () => undefined, + onBecomeCoordinator: () => undefined, + onRevocationAnnounce: () => undefined, + onPresenceAdvert: () => undefined, + }; +} + +describe("WireMeshTransport connectHub/disconnectHub", () => { + test("connectHub dials the given hub URL and disconnectHub drops it, observable on the hub's own connection count", async () => { + const hub = await realHubOverWs(); + const transport = new WireMeshTransport(noopEvents(), generateIdentity()); + try { + expect(transport.hub.isConnected).toBe(false); + + await transport.connectHub?.(hub.url); + + expect(transport.hub.isConnected).toBe(true); + await waitForCondition(() => hub.connectionCount() === 1); + + await transport.disconnectHub?.(); + + expect(transport.hub.isConnected).toBe(false); + await waitForCondition(() => hub.connectionCount() === 0); + } finally { + await transport.shutdown(); + await hub.close(); + } + }); + + test("disconnectHub is a safe no-op when connectHub was never called", async () => { + const transport = new WireMeshTransport(noopEvents(), generateIdentity()); + try { + await expect(transport.disconnectHub?.()).resolves.toBeUndefined(); + } finally { + await transport.shutdown(); + } + }); +}); From 2ab8e2e9c126576b0898b6f1d6d4a526bc106af5 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:29:42 +0100 Subject: [PATCH 4/8] feat(core): wire CoordinatorGateway into MeshStore's become/lose paths MeshStore now constructs a CoordinatorGateway and calls onBecameCoordinator() from both places this store takes over as coordinator: init()'s own fresh-bind fallback and PeerLifecycle's handleBecomeCoordinator takeover. shutdown() calls onLostCoordinator() unconditionally; the gateway's own connected guard makes that a no-op unless this instance actually held the role. The hub URL defaults to DEFAULT_HUB_URL (mesh-store-shared.ts) and is overridable via a new MeshStore constructor parameter. --- src/core/mesh-store-shared.ts | 3 ++ src/core/mesh-store.ts | 24 ++++++++- src/core/peer-lifecycle.ts | 6 ++- src/test/mesh-store-orchestration.test.ts | 63 +++++++++++++++++++++++ src/test/peer-lifecycle.test.ts | 9 ++++ 5 files changed, 102 insertions(+), 3 deletions(-) diff --git a/src/core/mesh-store-shared.ts b/src/core/mesh-store-shared.ts index 7e69d7e0..2938f8cc 100644 --- a/src/core/mesh-store-shared.ts +++ b/src/core/mesh-store-shared.ts @@ -44,6 +44,9 @@ export const MAX_QUEUED_DELIVERIES_PER_AGENT = 100; /** The coordinator's own bind host -- always loopback, since the mesh coordinator role only ever needs to be reachable from other local peers on this machine. Shared between mesh-store.ts's own init() and PeerLifecycle's handleBecomeCoordinator. */ export const COORDINATOR_HOST = "127.0.0.1"; +/** The production relay hub this machine's gateway dials once it becomes the local mesh coordinator (agent-comms#154). Configuration: MeshStore's own constructor accepts an override, threaded from createBridgeMesh/createBridgeMeshSync, for tests and any future non-default deployment -- this is only the default. */ +export const DEFAULT_HUB_URL = "wss://mesh.exadev.io/"; + /** Shallow-clones an entry together with its own `readBy` array, so a merged history never shares mutable array references with either input it was built from. */ function cloneWithReadBy(entry: T): T { return { ...entry, readBy: [...entry.readBy] }; diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index a5e01eb6..607bee4a 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -14,8 +14,9 @@ import { CommsError } from "./store.js"; import { DiscoveryManager } from "./discovery.js"; import { MdnsDiscoveryBackend } from "./discovery-mdns.js"; import { TailscaleDiscoveryBackend } from "./discovery-tailscale.js"; -import { COORDINATOR_HOST } from "./mesh-store-shared.js"; +import { COORDINATOR_HOST, DEFAULT_HUB_URL } from "./mesh-store-shared.js"; import type { MeshStoreIdentity } from "./mesh-store-shared.js"; +import { CoordinatorGateway } from "./coordinator-gateway.js"; import { DeliveryEngine } from "./delivery-engine.js"; import { RoomProtocol } from "./room-protocol.js"; import { RoomMessaging } from "./room-messaging.js"; @@ -71,6 +72,7 @@ export class MeshStore implements CommsStore { peerId: string; readonly startedAt: string; readonly coordinatorPort: number; + private readonly hubUrl: string; private readonly agents = new Map(); private readonly rooms = new Map(); @@ -98,6 +100,7 @@ export class MeshStore implements CommsStore { private readonly agentRegistry: AgentRegistry; private readonly connectionApproval: ConnectionApproval; private readonly staleAgentChecker: StaleAgentChecker; + private readonly coordinatorGateway: CoordinatorGateway; private readonly peerLifecycle: PeerLifecycle; /** This store's own current AgentStatus, synchronously -- the value WireMeshTransport's presence re-advertisement timer reads on every tick. undefined before registerAgent has ever run (no self agent record exists yet), in which case there is nothing yet to advertise. */ @@ -167,10 +170,14 @@ export class MeshStore implements CommsStore { }; } - constructor(coordinatorPort: number = DEFAULT_COORDINATOR_PORT) { + constructor( + coordinatorPort: number = DEFAULT_COORDINATOR_PORT, + hubUrl: string = DEFAULT_HUB_URL, + ) { this.peerId = nanoid(PEER_ID_LENGTH); this.startedAt = new Date().toISOString(); this.coordinatorPort = coordinatorPort; + this.hubUrl = hubUrl; // Discovery manager — registers available backends this.discovery = new DiscoveryManager(); @@ -268,6 +275,16 @@ export class MeshStore implements CommsStore { this.deliveryEngine.broadcastPatch(patch), }); + this.coordinatorGateway = new CoordinatorGateway({ + hubUrl: this.hubUrl, + connectHub: async (url) => { + await this.requireTransport().connectHub?.(url); + }, + disconnectHub: async () => { + await this.requireTransport().disconnectHub?.(); + }, + }); + this.peerLifecycle = new PeerLifecycle({ peerInfo: this.peerInfo, agents: this.agents, @@ -278,6 +295,7 @@ export class MeshStore implements CommsStore { roomProtocol: this.roomProtocol, deliveryEngine: this.deliveryEngine, staleAgentChecker: this.staleAgentChecker, + coordinatorGateway: this.coordinatorGateway, }); } @@ -348,6 +366,7 @@ export class MeshStore implements CommsStore { this.coordinatorPort, ); this.staleAgentChecker.start(); + await this.coordinatorGateway.onBecameCoordinator(); connected = true; } catch (coordErr) { const msg = @@ -750,6 +769,7 @@ export class MeshStore implements CommsStore { } this.staleAgentChecker.stop(); + await this.coordinatorGateway.onLostCoordinator(); await this.requireTransport().shutdown(); } } diff --git a/src/core/peer-lifecycle.ts b/src/core/peer-lifecycle.ts index d43878c4..f2f1f2f3 100644 --- a/src/core/peer-lifecycle.ts +++ b/src/core/peer-lifecycle.ts @@ -9,13 +9,14 @@ import type { SerialisedState, } from "./wire-protocol.js"; import { COORDINATOR_HOST } from "./mesh-store-shared.js"; +import type { CoordinatorGateway } from "./coordinator-gateway.js"; import type { DeliveryEngine } from "./delivery-engine.js"; import type { RoomProtocol } from "./room-protocol.js"; import type { StaleAgentChecker } from "./stale-agent-checker.js"; import type { ConnectionHandle, MeshTransport } from "./transport.js"; import type { AgentIdentity } from "./types.js"; -/** The state and collaborators PeerLifecycle needs from MeshStore. peerInfo/agents are direct references into MeshStore's own fields; coordinatorPort is a readonly value copied once; serialise is MeshStore's own retained method (constraint: it must stay directly on MeshStore.prototype, so PeerLifecycle calls it through this closure rather than owning it); roomProtocol/deliveryEngine/staleAgentChecker are the already-constructed instances (construction order: ... -\> roomProtocol -\> ... -\> staleAgentChecker -\> peerLifecycle), narrowed to what peer-lifecycle bookkeeping ever needs. */ +/** The state and collaborators PeerLifecycle needs from MeshStore. peerInfo/agents are direct references into MeshStore's own fields; coordinatorPort is a readonly value copied once; serialise is MeshStore's own retained method (constraint: it must stay directly on MeshStore.prototype, so PeerLifecycle calls it through this closure rather than owning it); roomProtocol/deliveryEngine/staleAgentChecker/coordinatorGateway are the already-constructed instances (construction order: ... -\> roomProtocol -\> ... -\> staleAgentChecker -\> coordinatorGateway -\> peerLifecycle), narrowed to what peer-lifecycle bookkeeping ever needs. */ export interface PeerLifecycleDeps { peerInfo: Map; agents: Map; @@ -26,6 +27,8 @@ export interface PeerLifecycleDeps { roomProtocol: Pick; deliveryEngine: Pick; staleAgentChecker: Pick; + /** Dials the hub the moment this side takes over as coordinator (agent-comms#154) -- see CoordinatorGateway's own class doc. Narrowed to the one method handleBecomeCoordinator ever calls; onLostCoordinator is MeshStore.shutdown()'s own concern, not this class's. */ + coordinatorGateway: Pick; } export class PeerLifecycle { @@ -114,6 +117,7 @@ export class PeerLifecycle { void this.deps.requireTransport().connectToPeer(peer, peerId); } this.deps.staleAgentChecker.start(); + await this.deps.coordinatorGateway.onBecameCoordinator(); } handlePeerDisconnected(handle: Readonly): void { diff --git a/src/test/mesh-store-orchestration.test.ts b/src/test/mesh-store-orchestration.test.ts index fb81c801..ae1c044c 100644 --- a/src/test/mesh-store-orchestration.test.ts +++ b/src/test/mesh-store-orchestration.test.ts @@ -38,6 +38,8 @@ function fakeTransport(): MeshTransport { listListeners: vi.fn().mockReturnValue([]), shutdown: vi.fn().mockResolvedValue(undefined), unref: vi.fn<() => void>(), + connectHub: vi.fn().mockResolvedValue(undefined), + disconnectHub: vi.fn().mockResolvedValue(undefined), }; } @@ -265,6 +267,46 @@ describe("MeshStore — init()", () => { expect(transport.startDataServer).toHaveBeenCalledTimes(1); }); + + it("dials the default hub URL once it becomes coordinator on a fresh bind", async () => { + const store = new MeshStore(); + const transport = fakeTransport(); + vi.mocked(transport.connectToCoordinator).mockRejectedValue( + new Error("ECONNREFUSED"), + ); + store.setTransport(transport); + + await store.init(); + + expect(transport.connectHub).toHaveBeenCalledWith("wss://mesh.exadev.io/"); + expect(transport.connectHub).toHaveBeenCalledTimes(1); + }); + + it("dials the constructor-supplied hub URL override instead of the default", async () => { + const store = new MeshStore(undefined, "wss://hub.example.test/"); + const transport = fakeTransport(); + vi.mocked(transport.connectToCoordinator).mockRejectedValue( + new Error("ECONNREFUSED"), + ); + store.setTransport(transport); + + await store.init(); + + expect(transport.connectHub).toHaveBeenCalledWith( + "wss://hub.example.test/", + ); + }); + + it("never dials the hub when joining an existing coordinator rather than becoming one", async () => { + const store = new MeshStore(); + const transport = fakeTransport(); + store.setTransport(transport); + + await store.init(); + + expect(transport.connectToCoordinator).toHaveBeenCalledTimes(1); + expect(transport.connectHub).not.toHaveBeenCalled(); + }); }); describe("MeshStore — events getter dispatch table", () => { @@ -675,4 +717,25 @@ describe("MeshStore — shutdown()", () => { expect(broadcastSpy).not.toHaveBeenCalled(); }); + + it("drops the held hub connection when this instance had become coordinator", async () => { + vi.mocked(transport.connectToCoordinator).mockRejectedValue( + new Error("ECONNREFUSED"), + ); + await store.init(); + expect(transport.connectHub).toHaveBeenCalledTimes(1); + + await store.shutdown(); + + expect(transport.disconnectHub).toHaveBeenCalledTimes(1); + }); + + it("never touches disconnectHub when this instance never became coordinator", async () => { + await store.init(); + expect(transport.connectHub).not.toHaveBeenCalled(); + + await store.shutdown(); + + expect(transport.disconnectHub).not.toHaveBeenCalled(); + }); }); diff --git a/src/test/peer-lifecycle.test.ts b/src/test/peer-lifecycle.test.ts index 0db94faa..d367a290 100644 --- a/src/test/peer-lifecycle.test.ts +++ b/src/test/peer-lifecycle.test.ts @@ -33,6 +33,7 @@ interface Harness { applyStateSync: ReturnType; applyPatch: ReturnType; staleAgentCheckerStart: ReturnType; + coordinatorGatewayOnBecameCoordinator: ReturnType; } function makeHarness(): Harness { @@ -48,6 +49,9 @@ function makeHarness(): Harness { const applyPatch = vi.fn().mockResolvedValue(undefined); const staleAgentCheckerStart = vi.fn(); + const coordinatorGatewayOnBecameCoordinator = vi + .fn() + .mockResolvedValue(undefined); const deps: PeerLifecycleDeps = { peerInfo: new Map(), agents: new Map(), @@ -58,6 +62,9 @@ function makeHarness(): Harness { roomProtocol: { flushPendingRoomRequests }, deliveryEngine: { applyStateSync, applyPatch }, staleAgentChecker: { start: staleAgentCheckerStart }, + coordinatorGateway: { + onBecameCoordinator: coordinatorGatewayOnBecameCoordinator, + }, }; return { deps, @@ -67,6 +74,7 @@ function makeHarness(): Harness { applyStateSync, applyPatch, staleAgentCheckerStart, + coordinatorGatewayOnBecameCoordinator, }; } @@ -202,6 +210,7 @@ describe("PeerLifecycle — handleBecomeCoordinator", () => { OWNER_ID, ); expect(h.staleAgentCheckerStart).toHaveBeenCalledTimes(1); + expect(h.coordinatorGatewayOnBecameCoordinator).toHaveBeenCalledTimes(1); }); }); From 2fefaef854c964094ae0ae47e8e2ad148aafa16b Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:31:01 +0100 Subject: [PATCH 5/8] feat(core): thread hubUrl through createBridgeMesh/createBridgeMeshSync Every bridge builds its MeshStore through this factory; hubUrl now passes through to MeshStore's own constructor alongside the existing coordinatorPort parameter, so a test (or a future non-default deployment) can override which hub a bridge's coordinator gateway dials without touching MeshStore directly. --- src/core/bridge-mesh.ts | 5 ++++- src/test/bridge-mesh.test.ts | 21 +++++++++++++++++++++ 2 files changed, 25 insertions(+), 1 deletion(-) diff --git a/src/core/bridge-mesh.ts b/src/core/bridge-mesh.ts index f336a847..e07fec25 100644 --- a/src/core/bridge-mesh.ts +++ b/src/core/bridge-mesh.ts @@ -31,9 +31,10 @@ export interface BridgeMeshSync extends BridgeMesh { export function createBridgeMeshSync( slot: Readonly, coordinatorPort?: number, + hubUrl?: string, ): BridgeMeshSync { const identity = loadOrCreateIdentity(slot); - const store = new MeshStore(coordinatorPort); + const store = new MeshStore(coordinatorPort, hubUrl); store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId)); // One shared dataStorage instance for both the transport's own data-domain frame responder and the store's own durable-send mint path (P5, agent-comms#50) -- oplogDirFor(slot) needs only the slot, not the async identity below, so this can be constructed synchronously right here. const dataStorage = createNodeFsStorage({ dir: oplogDirFor(slot) }); @@ -70,10 +71,12 @@ export function createBridgeMeshSync( export async function createBridgeMesh( slot: Readonly, coordinatorPort?: number, + hubUrl?: string, ): Promise { const { store, tool, attachIdentity } = createBridgeMeshSync( slot, coordinatorPort, + hubUrl, ); await attachIdentity(); return { store, tool }; diff --git a/src/test/bridge-mesh.test.ts b/src/test/bridge-mesh.test.ts index bdab3874..9ade7844 100644 --- a/src/test/bridge-mesh.test.ts +++ b/src/test/bridge-mesh.test.ts @@ -13,6 +13,7 @@ import { type IdentitySlot, } from "../core/identity-store.js"; import { waitFor } from "./test-transport.js"; +import { realHubOverWs, waitForCondition } from "./hub-helpers.js"; // Base of the ephemeral coordinator-port range used to avoid colliding with the mesh's real well-known port. const TEST_COORDINATOR_PORT_BASE = 20_900; @@ -82,3 +83,23 @@ test("createBridgeMesh passes an explicit coordinatorPort through to MeshStore, await a.store.shutdown(); } }); + +test("createBridgeMesh passes an explicit hubUrl through to MeshStore, dialled once the store becomes coordinator and dropped on shutdown", async () => { + const hub = await realHubOverWs(); + const slot = tempSlot("test-harness-hub"); + const port = + TEST_COORDINATOR_PORT_BASE + + Math.floor(Math.random() * TEST_COORDINATOR_PORT_RANGE); + const { store } = await createBridgeMesh(slot, port, hub.url); + try { + await store.init(); + expect(store.connected).toBeTruthy(); + await waitForCondition(() => hub.connectionCount() === 1); + + await store.shutdown(); + + await waitForCondition(() => hub.connectionCount() === 0); + } finally { + await hub.close(); + } +}); From e253b4f171d7bbb1aeacdca51d426d8691f76d76 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:41:52 +0100 Subject: [PATCH 6/8] fix(core): isolate a hub dial failure from local coordinator election CoordinatorGateway.onBecameCoordinator previously let a connectHub rejection propagate all the way through MeshStore.init(), so a bridge whose becomeCoordinator() succeeded locally would still fail to start if the gateway's hub happened to be unreachable. The local mesh's own coordinator election has no dependency on internet or hub availability, so this coupled two things that were never meant to be coupled. A dial failure is now reported via an injected onError callback (wired through to MeshStore's own onError) and swallowed; isConnected stays false so the next onBecameCoordinator call retries. --- src/core/coordinator-gateway.ts | 14 +++++++--- src/core/mesh-store.ts | 3 +++ src/test/coordinator-gateway.test.ts | 32 ++++++++++++++++++++++- src/test/mesh-store-orchestration.test.ts | 21 +++++++++++++++ 4 files changed, 66 insertions(+), 4 deletions(-) diff --git a/src/core/coordinator-gateway.ts b/src/core/coordinator-gateway.ts index d48a4bd0..853853b4 100644 --- a/src/core/coordinator-gateway.ts +++ b/src/core/coordinator-gateway.ts @@ -9,6 +9,8 @@ export interface CoordinatorGatewayDeps { connectHub: (url: string) => Promise; /** Drops the held hub connection, if any. Backed by the transport's own optional disconnectHub. */ disconnectHub: () => Promise; + /** Reports a hub-dial failure. Never invoked for anything else -- onBecameCoordinator's own guarantee (see its doc comment) is that a hub problem is always reported this way, never thrown, so this is the only signal a caller gets that the gateway role didn't actually connect. */ + onError?: (error: Error) => void; } export class CoordinatorGateway { @@ -21,11 +23,17 @@ export class CoordinatorGateway { return this.connected; } - /** Dials the hub for this machine's gateway role. Idempotent: a call while already connected is a no-op, since nothing in this codebase's own coordinator-election machinery re-fires "became coordinator" without an intervening onLostCoordinator -- this guard is defensive, not a known double-fire path. */ + /** Dials the hub for this machine's gateway role. Idempotent: a call while already connected is a no-op, since nothing in this codebase's own coordinator-election machinery re-fires "became coordinator" without an intervening onLostCoordinator -- this guard is defensive, not a known double-fire path. Never throws: local coordinator election (the whole reason this side is calling this at all) must not depend on the hub being reachable, so a dial failure is reported via deps.onError and swallowed here, leaving isConnected false so a later onBecameCoordinator call retries rather than being blocked by the earlier failure's own idempotency guard. */ async onBecameCoordinator(): Promise { if (this.connected) return; - this.connected = true; - await this.deps.connectHub(this.deps.hubUrl); + try { + await this.deps.connectHub(this.deps.hubUrl); + this.connected = true; + } catch (error) { + this.deps.onError?.( + error instanceof Error ? error : new Error(String(error)), + ); + } } /** Drops this machine's held hub connection. A no-op if this side never became the gateway, or already lost the role -- MeshStore.shutdown() calls this unconditionally regardless of coordinator status, so this guard is what makes that safe rather than a redundant extra close. */ diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 607bee4a..4c197dbc 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -283,6 +283,9 @@ export class MeshStore implements CommsStore { disconnectHub: async () => { await this.requireTransport().disconnectHub?.(); }, + onError: (error) => { + this.onError?.(error); + }, }); this.peerLifecycle = new PeerLifecycle({ diff --git a/src/test/coordinator-gateway.test.ts b/src/test/coordinator-gateway.test.ts index 78389337..1fda5663 100644 --- a/src/test/coordinator-gateway.test.ts +++ b/src/test/coordinator-gateway.test.ts @@ -13,15 +13,23 @@ function makeHarness(): { gateway: CoordinatorGateway; connectHub: ReturnType; disconnectHub: ReturnType; + onError: ReturnType; } { const connectHub = vi.fn().mockResolvedValue(undefined); const disconnectHub = vi.fn().mockResolvedValue(undefined); + const onError = vi.fn<(error: Error) => void>(); const deps: CoordinatorGatewayDeps = { hubUrl: HUB_URL, connectHub, disconnectHub, + onError, + }; + return { + gateway: new CoordinatorGateway(deps), + connectHub, + disconnectHub, + onError, }; - return { gateway: new CoordinatorGateway(deps), connectHub, disconnectHub }; } describe("CoordinatorGateway — onBecameCoordinator", () => { @@ -51,6 +59,28 @@ describe("CoordinatorGateway — onBecameCoordinator", () => { expect(connectHub).toHaveBeenCalledTimes(1); }); + + it("reports a dial failure via onError rather than throwing -- local coordinator election must not depend on hub reachability", async () => { + const { gateway, connectHub, onError } = makeHarness(); + const dialError = new Error("ECONNREFUSED"); + connectHub.mockRejectedValueOnce(dialError); + + await expect(gateway.onBecameCoordinator()).resolves.toBeUndefined(); + + expect(onError).toHaveBeenCalledWith(dialError); + }); + + it("does not mark itself connected after a failed dial, so a later call can retry", async () => { + const { gateway, connectHub } = makeHarness(); + connectHub.mockRejectedValueOnce(new Error("ECONNREFUSED")); + await gateway.onBecameCoordinator(); + expect(gateway.isConnected).toBe(false); + + await gateway.onBecameCoordinator(); + + expect(gateway.isConnected).toBe(true); + expect(connectHub).toHaveBeenCalledTimes(2); + }); }); describe("CoordinatorGateway — onLostCoordinator", () => { diff --git a/src/test/mesh-store-orchestration.test.ts b/src/test/mesh-store-orchestration.test.ts index ae1c044c..698ae01d 100644 --- a/src/test/mesh-store-orchestration.test.ts +++ b/src/test/mesh-store-orchestration.test.ts @@ -307,6 +307,27 @@ describe("MeshStore — init()", () => { expect(transport.connectToCoordinator).toHaveBeenCalledTimes(1); expect(transport.connectHub).not.toHaveBeenCalled(); }); + + it("still succeeds locally (becomes coordinator) when the hub dial fails -- local coordinator election must not depend on hub reachability", async () => { + const store = new MeshStore(); + const transport = fakeTransport(); + vi.mocked(transport.connectToCoordinator).mockRejectedValue( + new Error("ECONNREFUSED"), + ); + const { connectHub } = transport; + if (connectHub === undefined) throw new Error("expected connectHub"); + const hubError = new Error("hub unreachable"); + vi.mocked(connectHub).mockRejectedValue(hubError); + store.setTransport(transport); + const onError = vi.fn<(error: Error) => void>(); + store.onError = onError; + + await expect(store.init()).resolves.toBeUndefined(); + + expect(transport.becomeCoordinator).toHaveBeenCalledTimes(1); + expect(onError).toHaveBeenCalledWith(hubError); + expect(transport.unref).toHaveBeenCalledTimes(1); + }); }); describe("MeshStore — events getter dispatch table", () => { From 35ef7fe596e6c6f1fadd51a2d3d45f318b7b6a5e Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 15:37:21 +0100 Subject: [PATCH 7/8] test(core): point the multi-process smoke test at an unreachable hub MeshStore's coordinator now dials a hub on becoming coordinator (agent-comms#154); without an explicit override this smoke test would have started making a real outbound connection to the production hub on every CI run. Pointing it at a guaranteed-unreachable local address keeps the test offline and deterministic, and exercises CoordinatorGateway's own dial-failure isolation along the way. --- src/test/mesh-smoke.runner.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/test/mesh-smoke.runner.ts b/src/test/mesh-smoke.runner.ts index 5131e348..56d4ec2b 100644 --- a/src/test/mesh-smoke.runner.ts +++ b/src/test/mesh-smoke.runner.ts @@ -12,6 +12,8 @@ import * as net from "node:net"; const SMOKE_PORT = 19877; const SMOKE_HOST = "127.0.0.1"; +// A guaranteed-unreachable local address, not the real default hub (agent-comms#154's MeshStore now dials one on becoming coordinator) -- this smoke test's own concern is the coordinator/discovery/delivery path across a real process boundary, not live connectivity to production infrastructure. Pointing at an address nothing listens on exercises CoordinatorGateway's own failure-isolation (a dial failure must never block local coordinator election) deterministically and offline, rather than this test's outcome depending on whether the CI runner can reach the internet. +const UNREACHABLE_HUB_URL = "ws://127.0.0.1:1/"; interface TestMessage { type: string; @@ -120,7 +122,7 @@ function buildScript(name: string, actions: string): string { `const fs = require("node:fs");`, `function log(msg) { process.stdout.write(JSON.stringify(msg) + "\\n"); }`, `(async () => {`, - ` const store = new MeshStore(${String(SMOKE_PORT)});`, + ` const store = new MeshStore(${String(SMOKE_PORT)}, "${UNREACHABLE_HUB_URL}");`, ` const slot = { harness: "smoke-${name}", cwd: "/test/${name}", dir: fs.mkdtempSync(path.join(os.tmpdir(), "agent-comms-smoke-${name}-")) };`, ` const identity = loadOrCreateIdentity(slot);`, ` store.peerId = deviceIdToHex(Uint8Array.from(identity.deviceId));`, From cb67f5db63a4da01e32a9c81c1078f73464d560c Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 16:00:20 +0100 Subject: [PATCH 8/8] fix(core): drop state_sync/state_update relayed by hub peers A Codex security review on PR #169 found that HubSession.consume() forwarded any hub-relayed state_sync/state_update straight into applyPatch/applyStateSync with no per-peer admission control. That path existed since #151, but #154's own change is what first made a coordinator dial the hub by default in production, turning a latent gap into a live one: any identity able to reach the configured hub could patch local mesh state, including queuing a spoofed delivery. Real per-peer admission control is #156's own deliverable. Until then, the two state-mutating message methods are dropped before they ever reach onMessage; every other message type this session relays was already inert on receipt (PeerLifecycle only reacts to these two), so this closes the whole exposure through this path. --- src/core/hub-session.ts | 9 ++- src/test/hub-mode-session.integration.test.ts | 64 ++++++++++++++++++- 2 files changed, 69 insertions(+), 4 deletions(-) diff --git a/src/core/hub-session.ts b/src/core/hub-session.ts index 5c307af6..fb8e4e02 100644 --- a/src/core/hub-session.ts +++ b/src/core/hub-session.ts @@ -100,7 +100,7 @@ export class HubSession { })(); } - /** Dispatches inbound relayed manage-requests: each is handled with a handle keyed by the SENDING device (request.fromDevice names it on relay-routed requests), so onMessage and every downstream consumer see the true origin, never the hub. Replies ride respond()'s own relay routing back. */ + /** Dispatches inbound relayed manage-requests: each is handled with a handle keyed by the SENDING device (request.fromDevice names it on relay-routed requests), so onMessage and every downstream consumer see the true origin, never the hub. Replies ride respond()'s own relay routing back. A state_sync/state_update is dropped before ever reaching onMessage/applyPatch -- see isStateMutatingMessage's own doc for why: the hub has no per-peer admission control yet (that lands in agent-comms#156), so accepting one from an arbitrary hub peer would let it directly patch this side's mesh state (a security review finding on agent-comms#169, which is what first wired a hub connection into production's default coordinator path at all). */ private consume(session: AcceptedMeshSession): void { void (async () => { for await (const request of session.incomingManageRequests) { @@ -111,7 +111,7 @@ export class HubSession { : "hub-peer"; this.hubPeersKnown.add(senderHex); const message = extractMessage(request.command); - if (message !== undefined) { + if (message !== undefined && !isStateMutatingMessage(message)) { this.deps.events.onMessage({ id: senderHex }, message); } await request.respond({ result: "ok" }).catch(() => undefined); @@ -149,3 +149,8 @@ function hexToBytes(hex: string): Uint8Array { } return bytes; } + +/** A state_sync or state_update carries authority to directly overwrite or patch this side's own mesh state (agents, rooms, messages, deliveries) -- on an ordinary peer session that authority is meaningful because the peer already passed connect_request/introduce approval or the coordinator's own trusted mesh membership. A hub-relayed sender has passed neither: today's hub accepts any self-generated identity and gates nothing per-peer (agent-comms#156's own future deliverable), so treating its state_sync/state_update as equally authoritative would let an arbitrary hub peer inject an outcome indistinguishable from a genuine mesh event, e.g. a spoofed inbound delivery. Every other legacy message method this session might relay is already inert on receipt (PeerLifecycle's own handleDataMessage only reacts to these two), so filtering exactly these two is a complete fix for this specific path, not a partial one. */ +function isStateMutatingMessage(message: MeshMessage): boolean { + return message.method === "state_sync" || message.method === "state_update"; +} diff --git a/src/test/hub-mode-session.integration.test.ts b/src/test/hub-mode-session.integration.test.ts index 0259bab5..88c69584 100644 --- a/src/test/hub-mode-session.integration.test.ts +++ b/src/test/hub-mode-session.integration.test.ts @@ -5,15 +5,20 @@ import { realHubOverWs, waitForCondition } from "./hub-helpers.js"; import { generateIdentity } from "../core/identity.js"; import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; import { WireMeshTransport } from "../core/wire-mesh-transport.js"; -import type { TransportEvents } from "../core/transport.js"; +import type { ConnectionHandle, TransportEvents } from "../core/transport.js"; +import type { MeshMessage } from "../core/wire-protocol.js"; function recordingEvents(): TransportEvents & { messages: { from: string; text: string }[]; + allMessages: MeshMessage[]; } { const messages: { from: string; text: string }[] = []; + const allMessages: MeshMessage[] = []; return { messages, - onMessage: (handle, message) => { + allMessages, + onMessage: (handle: Readonly, message: MeshMessage) => { + allMessages.push(message); if (message.method === "peer_joined") { messages.push({ from: handle.id, text: message.peer.id }); } @@ -77,4 +82,59 @@ describe("connectToHub", () => { await transportA.shutdown(); await transportB.shutdown(); }); + + it("never applies a state_sync or state_update relayed by an unauthenticated hub peer (agent-comms#169 security finding: real per-peer admission lands in #156, but nothing today should let any hub peer patch local mesh state)", async () => { + const hub = await realHubOverWs(); + cleanups.push(hub.close); + + const eventsA = recordingEvents(); + const eventsB = recordingEvents(); + const transportA = new WireMeshTransport(eventsA, generateIdentity()); + const transportB = new WireMeshTransport(eventsB, generateIdentity()); + await transportA.hub.connect(hub.url); + await transportB.hub.connect(hub.url); + + const deviceA = await transportA.hub.ownDeviceHex(); + const deviceB = await transportB.hub.ownDeviceHex(); + await waitForCondition(() => { + return ( + transportA.hub.peers().includes(deviceB) && + transportB.hub.peers().includes(deviceA) + ); + }); + + await transportA.hub.sendToPeer(deviceB, { + method: "state_update", + patch: { type: "agent_offline", agentId: "spoofed" }, + }); + await transportA.hub.sendToPeer(deviceB, { + method: "state_sync", + state: { + agents: {}, + rooms: {}, + messages: {}, + dms: {}, + deliveryQueues: {}, + }, + }); + // A message type this codebase's own onMessage handler treats as inert either way -- proves the hub connection and relay pairing genuinely delivered something to B (ruling out "nothing arrived at all" as a false-negative explanation for state_update/state_sync never showing up below), while confirming filtering is specific to the two state-mutating methods rather than a blanket drop of everything. + await transportA.hub.sendToPeer(deviceB, { + method: "peer_joined", + peer: { + id: "proof-of-delivery", + port: 0, + startedAt: "2026-01-01T00:00:00.000Z", + }, + }); + + await waitForCondition(() => eventsB.messages.length > 0); + expect( + eventsB.allMessages.some( + (m) => m.method === "state_update" || m.method === "state_sync", + ), + ).toBe(false); + + await transportA.shutdown(); + await transportB.shutdown(); + }); });