From 86443384b2e870d14c5a772c6c4691e6f7992339 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 01:23:53 +0100 Subject: [PATCH] feat(core): add hub relay mode A hub connection (wss URL) is a RELAY, not a coordinator: no connect_request or introduce approval applies. The session 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 via sendManageRequest targetDevice routing. Inbound relayed manage-requests are dispatched keyed by the SENDING device (request.fromDevice), never the hub itself, and replies ride respond() routing back. The integration test runs a real createRelayHub over local WebSockets and proves the full stack: two transports discover each other via the hub gossip and exchange a message through a relay pairing. Closes #151 --- src/core/hub-session.ts | 138 +++++++++++ src/core/wire-mesh-transport.ts | 30 ++- src/test/hub-helpers.ts | 7 + src/test/hub-mode-session.integration.test.ts | 220 ++++++++++++++++++ 4 files changed, 391 insertions(+), 4 deletions(-) create mode 100644 src/core/hub-session.ts create mode 100644 src/test/hub-helpers.ts create mode 100644 src/test/hub-mode-session.integration.test.ts diff --git a/src/core/hub-session.ts b/src/core/hub-session.ts new file mode 100644 index 00000000..d2acbbe5 --- /dev/null +++ b/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>; + events: Readonly; + /** 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, + frame: Readonly, + ) => void | Promise; + /** 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(); + + constructor(private readonly deps: Readonly) {} + + /** This node's own device-id, hex-encoded -- the identifier hub peers address it by. */ + async ownDeviceHex(): Promise { + 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 { + 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 { + 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 { + 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 { + 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; +} diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index cd64911b..15ca1842 100644 --- a/src/core/wire-mesh-transport.ts +++ b/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(); @@ -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( diff --git a/src/test/hub-helpers.ts b/src/test/hub-helpers.ts new file mode 100644 index 00000000..072afe51 --- /dev/null +++ b/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)); +} diff --git a/src/test/hub-mode-session.integration.test.ts b/src/test/hub-mode-session.integration.test.ts new file mode 100644 index 00000000..f8cc941a --- /dev/null +++ b/src/test/hub-mode-session.integration.test.ts @@ -0,0 +1,220 @@ +// 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 { 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 }[]; +} { + const messages: { from: string; text: string }[] = []; + return { + messages, + onMessage: (handle, message) => { + if (message.method === "peer_joined") { + messages.push({ from: handle.id, text: message.peer.id }); + } + }, + onPeerConnected: () => undefined, + onPeerDisconnected: () => undefined, + onIntroduction: () => undefined, + onConnectionRequest: () => undefined, + onPeerList: () => undefined, + onPeerJoined: () => undefined, + onBecomeCoordinator: () => undefined, + onRevocationAnnounce: () => undefined, + onPresenceAdvert: () => undefined, + }; +} + +/** 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 () => { + for (const close of cleanups.splice(0)) { + await close(); + } +}); + +describe("connectToHub", () => { + it("two transports discover each other via the hub's gossip and exchange messages through relay pairings", 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); + + // Discovery: each side's hubPeers should eventually list the other. + 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) + ); + }); + + // Messaging: A sends to B via the hub; B's onMessage fires with A's device. + await transportA.hub.sendToPeer(deviceB, { + method: "peer_joined", + peer: { + id: "hub-says-hi", + port: 0, + startedAt: "2026-01-01T00:00:00.000Z", + }, + }); + + await waitForCondition(() => eventsB.messages.length > 0); + expect(eventsB.messages[0]?.text).toBe("hub-says-hi"); + expect(eventsB.messages[0]?.from).toBe(deviceA); + + 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(); + }); +}