feat(core): coordinator holds the hub connection (gateway role)

src/core/bridge-mesh.ts

@@ -31,9 +31,10 @@ export interface BridgeMeshSync extends BridgeMesh {
 export function createBridgeMeshSync(
   slot: Readonly<IdentitySlot>,
   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<IdentitySlot>,
   coordinatorPort?: number,
+  hubUrl?: string,
 ): Promise<BridgeMesh> {
   const { store, tool, attachIdentity } = createBridgeMeshSync(
     slot,
     coordinatorPort,
+    hubUrl,
   );
   await attachIdentity();
   return { store, tool };

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<void>;
+  /** Drops the held hub connection, if any. Backed by the transport's own optional disconnectHub. */
+  disconnectHub: () => Promise<void>;
+  /** 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<CoordinatorGatewayDeps>) {}
+
+  /** 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<void> {
+    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<void> {
+    if (!this.connected) return;
+    this.connected = false;
+    await this.deps.disconnectHub();
+  }
+}

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<string, PeerAdvert>,
+  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);
+    }
+  }
+}

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<void> {
+    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<ArrayBuffer> {
   }
   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";
+}

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<T extends { readBy: string[] }>(entry: T): T {
   return { ...entry, readBy: [...entry.readBy] };

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<string, AgentIdentity>();
   private readonly rooms = new Map<string, Room>();
@@ -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();
   }
 }

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<string, PeerInfo>;
   agents: Map<string, AgentIdentity>;
@@ -26,6 +27,8 @@ export interface PeerLifecycleDeps {
   roomProtocol: Pick<RoomProtocol, "flushPendingRoomRequests">;
   deliveryEngine: Pick<DeliveryEngine, "applyStateSync" | "applyPatch">;
   staleAgentChecker: Pick<StaleAgentChecker, "start">;
+  /** 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<CoordinatorGateway, "onBecameCoordinator">;
 }
 
 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<ConnectionHandle>): void {

src/core/transport.ts

@@ -259,4 +259,14 @@ export interface MeshTransport {
     deviceId: string;
     advert: Readonly<Record<string, unknown>>;
   }[];
+
+  /**
+   * 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<void>;
+
+  /**
+   * Drops this side's own held hub connection, if any. A no-op when none is live. Same optionality caveat as connectHub.
+   */
+  disconnectHub?: () => Promise<void>;
 }