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/core/coordinator-gateway.ts b/src/core/coordinator-gateway.ts new file mode 100644 index 00000000..853853b4 --- /dev/null +++ b/src/core/coordinator-gateway.ts @@ -0,0 +1,45 @@ +/** + * 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; + /** 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 { + 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. 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; + 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. */ + async onLostCoordinator(): Promise { + if (!this.connected) return; + this.connected = false; + await this.deps.disconnectHub(); + } +} 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..fb8e4e02 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]; @@ -87,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) { @@ -98,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); @@ -136,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/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..4c197dbc 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,19 @@ 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?.(); + }, + onError: (error) => { + this.onError?.(error); + }, + }); + this.peerLifecycle = new PeerLifecycle({ peerInfo: this.peerInfo, agents: this.agents, @@ -278,6 +298,7 @@ export class MeshStore implements CommsStore { roomProtocol: this.roomProtocol, deliveryEngine: this.deliveryEngine, staleAgentChecker: this.staleAgentChecker, + coordinatorGateway: this.coordinatorGateway, }); } @@ -348,6 +369,7 @@ export class MeshStore implements CommsStore { this.coordinatorPort, ); this.staleAgentChecker.start(); + await this.coordinatorGateway.onBecameCoordinator(); connected = true; } catch (coordErr) { const msg = @@ -750,6 +772,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/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/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(); + } +}); diff --git a/src/test/coordinator-gateway.test.ts b/src/test/coordinator-gateway.test.ts new file mode 100644 index 00000000..1fda5663 --- /dev/null +++ b/src/test/coordinator-gateway.test.ts @@ -0,0 +1,124 @@ +/** + * 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; + 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, + }; +} + +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); + }); + + 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", () => { + 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); + }); +}); 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/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..88c69584 100644 --- a/src/test/hub-mode-session.integration.test.ts +++ b/src/test/hub-mode-session.integration.test.ts @@ -1,30 +1,24 @@ // 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; +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 }); } @@ -41,114 +35,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 () => { @@ -196,25 +82,59 @@ describe("connectToHub", () => { await transportA.shutdown(); 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(); + 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(); }); -} +}); 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));`, diff --git a/src/test/mesh-store-orchestration.test.ts b/src/test/mesh-store-orchestration.test.ts index fb81c801..698ae01d 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,67 @@ 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(); + }); + + 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", () => { @@ -675,4 +738,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); }); }); 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(); + } + }); +});