Hub relay mode: connect agents through a wire-mesh relay hub

src/core/hub-session.ts

@@ -0,0 +1,138 @@
+/**
+ * HubSession -- the relay-hub connection mode (agent-comms#151), split from wire-mesh-transport.ts under the max-lines cap the same way connection-approval.ts and its siblings were. A hub connection (wss://mesh.exadev.io/) is a RELAY, not a coordinator: no connect_request/introduce approval applies. The session's own self-advert is forwarded by the hub to every other connected agent and the hub answers with a catch-up of everyone already there, so peers discover each other purely through gossip. Messages ride relay-connect pairings -- sendManageRequest's own targetDevice routing, which the session layer wraps as relay-data to exactly that device.
+ */
+
+import {
+  acceptMeshSession,
+  type AcceptedMeshSession,
+} from "wire-mesh-core/domain/mesh-session";
+import { deviceIdToHex } from "wire-mesh-core/domain/device-id";
+import type { Frame } from "wire-mesh-core/generated/protocol";
+import type { IdentityPort } from "wire-mesh-core/ports/identity";
+import type { Connection } from "wire-mesh-core/ports/transport";
+import type { TransportEvents } from "./transport.js";
+import type { MeshMessage } from "./wire-protocol.js";
+import { extractMessage } from "./room-router.js";
+import { connectWsUrl } from "./ws-dial.js";
+import { buildCommand, DOMAIN, FRAME_SCOPE } from "./wire-mesh-transport.js";
+
+export interface HubSessionDeps {
+  /** Resolves this node's own identity port once ready. */
+  identityReady: Promise<Readonly<IdentityPort>>;
+  events: Readonly<TransportEvents>;
+  /** The transport's own shutdown gate -- checked around every await, the same discipline the transport itself applies. */
+  isShuttingDown: () => boolean;
+  /** The addresses this node advertises in its own gossip self-advert. */
+  advertisedAddresses: readonly string[];
+  /** Registers a frame observer + dispatch for raw frames arriving on the hub connection (the transport's own handleDataFrame path). */
+  onFrame: (
+    connection: Readonly<Connection>,
+    frame: Readonly<Frame>,
+  ) => void | Promise<void>;
+  /** Tracks the session for shutdown -- every session the transport ever creates, always. */
+  trackForShutdown: (session: AcceptedMeshSession) => void;
+}
+
+export class HubSession {
+  private session: AcceptedMeshSession | undefined;
+  private readonly hubPeersKnown = new Set<string>();
+
+  constructor(private readonly deps: Readonly<HubSessionDeps>) {}
+
+  /** This node's own device-id, hex-encoded -- the identifier hub peers address it by. */
+  async ownDeviceHex(): Promise<string> {
+    const identity = await this.deps.identityReady;
+    return deviceIdToHex(identity.deviceId);
+  }
+
+  /** The device ids (hex) of peers discovered through the hub's gossiped directory. */
+  peers(): readonly string[] {
+    return [...this.hubPeersKnown];
+  }
+
+  /** Dials the hub and participates as a peer (see the class doc for the discovery and routing model). */
+  async connect(url: string): Promise<void> {
+    const connection = await connectWsUrl(url);
+    if (this.deps.isShuttingDown()) {
+      await connection.close();
+      return;
+    }
+    const identity = await this.deps.identityReady;
+    const session = await acceptMeshSession(connection, identity, [DOMAIN], {
+      onFrame: async (conn, frame) => this.deps.onFrame(conn, frame),
+      addresses: [...this.deps.advertisedAddresses],
+    });
+    if (this.deps.isShuttingDown()) {
+      await session.close();
+      return;
+    }
+    this.session = session;
+    this.deps.trackForShutdown(session);
+    // Merge the hub's directory (its catch-up arrives as the first session events) and keep refreshing it on every subsequent one.
+    void (async () => {
+      for await (const event of session.events) {
+        if (this.deps.isShuttingDown()) break;
+        for (const entry of event.directory) {
+          const hex = deviceIdToHex(entry.device);
+          if (hex !== deviceIdToHex(identity.deviceId)) {
+            this.hubPeersKnown.add(hex);
+          }
+        }
+        if (event.state.status === "closed") break;
+      }
+    })();
+    this.consume(session);
+    void (async () => {
+      await this.watchDisconnect(session);
+    })();
+  }
+
+  /** 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. */
+  private consume(session: AcceptedMeshSession): void {
+    void (async () => {
+      for await (const request of session.incomingManageRequests) {
+        if (this.deps.isShuttingDown()) break;
+        const senderHex =
+          request.fromDevice !== undefined
+            ? deviceIdToHex(request.fromDevice)
+            : "hub-peer";
+        this.hubPeersKnown.add(senderHex);
+        const message = extractMessage(request.command);
+        if (message !== undefined) {
+          this.deps.events.onMessage({ id: senderHex }, message);
+        }
+        await request.respond({ result: "ok" }).catch(() => undefined);
+      }
+    })();
+  }
+
+  private async watchDisconnect(session: AcceptedMeshSession): Promise<void> {
+    for await (const event of session.events) {
+      if (event.state.status === "closed") break;
+    }
+    if (this.session === session) {
+      this.session = undefined;
+    }
+  }
+
+  /** Sends one message to a hub-discovered peer through the hub's relay pairing. */
+  async sendToPeer(peerDeviceHex: string, message: MeshMessage): Promise<void> {
+    const session = this.session;
+    if (session === undefined) {
+      throw new Error("not connected to a hub");
+    }
+    await session.sendManageRequest(
+      buildCommand(message),
+      FRAME_SCOPE,
+      hexToBytes(peerDeviceHex),
+    );
+  }
+}
+
+function hexToBytes(hex: string): Uint8Array<ArrayBuffer> {
+  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;
+}

src/core/wire-mesh-transport.ts

@@ -10,6 +10,7 @@
 
 import { createTlsTransport } from "wire-mesh-core/adapters/tls-transport";
 import { connectWsUrl } from "./ws-dial.js";
+import { HubSession } from "./hub-session.js";
 import {
   acceptMeshSession,
   type AcceptedMeshSession,
@@ -194,6 +195,9 @@ export class WireMeshTransport implements MeshTransport {
   // -- The session dialled via connectToCoordinator, when this instance is not itself the coordinator --
   private coordinatorSession: AcceptedMeshSession | undefined;
 
+  // -- The hub relay mode (agent-comms#151), owning its own file under the max-lines cap: see hub-session.ts for the full connection model.
+  readonly hub: HubSession;
+
   // -- Every live session, keyed by the peer's authenticated device-id hex (== ConnectionHandle.id) -- covers coordinator-client, coordinator-accepted, and peer data sessions alike, since send()/broadcast() must reach whichever kind of session a peer happens to be reachable through. A single-slot-per-key map by construction: mesh formation genuinely establishes TWO independent sessions to the same peer (see the dataDials comment below), and the second one registered here simply overwrites the first as far as addressing goes -- fine for send()/broadcast() (either socket reaches the same peer), but NOT fine for shutdown, which must close every live session regardless of whether it's still reachable through this map. allSessions below exists specifically so shutdown never leaks the one this map's overwrite silently stopped tracking.
   private readonly peerSessions = new Map<string, AcceptedMeshSession>();
 
@@ -280,6 +284,16 @@ export class WireMeshTransport implements MeshTransport {
       privateKeyPem: identity.privateKey,
     });
     this.identityReady = toIdentityPort(identity);
+    this.hub = new HubSession({
+      identityReady: this.identityReady,
+      events: this.events,
+      isShuttingDown: this.isShuttingDown.bind(this),
+      advertisedAddresses: this.advertisedAddresses,
+      onFrame: async (conn, frame) => this.handleDataFrame(conn, frame),
+      trackForShutdown: (session) => {
+        this.allSessions.add(session);
+      },
+    });
     this.roomRouter = createRoomRouter({
       events,
       ...(roomVerbHandlers !== undefined ? { handlers: roomVerbHandlers } : {}),
@@ -405,7 +419,9 @@ export class WireMeshTransport implements MeshTransport {
     const deviceIdHex = deviceIdToHex(peerDeviceId);
     const identity = await this.identityReady;
     const session = await acceptMeshSession(connection, identity, [DOMAIN], {
-      onFrame: async (conn, frame) => this.handleDataFrame(conn, frame),
+      onFrame: async (conn, frame) => {
+        await this.handleDataFrame(conn, frame);
+      },
       addresses: this.advertisedAddresses,
     });
     if (this.isShuttingDown()) {
@@ -620,7 +636,9 @@ export class WireMeshTransport implements MeshTransport {
     );
     const identity = await this.identityReady;
     const session = await acceptMeshSession(connection, identity, [DOMAIN], {
-      onFrame: async (conn, frame) => this.handleDataFrame(conn, frame),
+      onFrame: async (conn, frame) => {
+        await this.handleDataFrame(conn, frame);
+      },
       addresses: this.advertisedAddresses,
     });
     this.coordinatorSession = session;
@@ -719,7 +737,9 @@ export class WireMeshTransport implements MeshTransport {
     }
     const identity = await this.identityReady;
     const session = await acceptMeshSession(connection, identity, [DOMAIN], {
-      onFrame: async (conn, frame) => this.handleDataFrame(conn, frame),
+      onFrame: async (conn, frame) => {
+        await this.handleDataFrame(conn, frame);
+      },
       addresses: this.advertisedAddresses,
     });
     if (this.isShuttingDown()) {
@@ -814,7 +834,9 @@ export class WireMeshTransport implements MeshTransport {
       : await this.wireTransport.connect(`${host}:${String(port)}`);
     const identity = await this.identityReady;
     const session = await acceptMeshSession(connection, identity, [DOMAIN], {
-      onFrame: async (conn, frame) => this.handleDataFrame(conn, frame),
+      onFrame: async (conn, frame) => {
+        await this.handleDataFrame(conn, frame);
+      },
       addresses: this.advertisedAddresses,
     });
     const outcome = await session.sendManageRequest(

src/test/hub-helpers.ts

@@ -0,0 +1,7 @@
+// 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";
+import type { Frame } from "wire-mesh-core/generated/protocol";
+
+export function cbor2ToBytes(frame: Frame): Uint8Array {
+  return new Uint8Array(encode(frame, cdeEncodeOptions));
+}