From fcb2a3d4a57ddd54a84c261136a1caee04fd60f0 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 16 Sep 2026 01:53:52 +0100 Subject: [PATCH] feat(core): aggregate a mesh-wide known-devices view from gossip WireMeshTransport.listKnownDevices() merges every live session's own peer-advert directory into one device-id-keyed map, keeping the newer advert (by snapshot-seconds) whenever a device is already known. This is the prerequisite the P3.8 room-discovery design and the eventual agent register/update/offline retirement both need: a way to read every device this side has ever heard gossip from, with its full advert (addresses and any open-extension field such as presence/status), rather than only the peer directly reachable through one session. Entries are never cleared on disconnect, so a device's last-known advert stays queryable while its session is momentarily down, matching how the legacy agents Map keeps a record after setAgentOffline rather than deleting it outright. --- src/core/wire-mesh-transport.ts | 30 ++++ ...-directory-aggregation.integration.test.ts | 151 ++++++++++++++++++ 2 files changed, 181 insertions(+) create mode 100644 src/test/gossip-directory-aggregation.integration.test.ts diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 9ba9f35b..d243e7ca 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -21,6 +21,7 @@ import type { CapabilityScope, CapabilityToken, ManageCommand, + PeerAdvert, RevocationEntry, } from "wire-mesh-core/generated/protocol"; import type { @@ -161,6 +162,34 @@ export class WireMeshTransport implements MeshTransport { // -- Every live session this transport has ever created, accepted or dialled, for shutdown's own use only -- never used for addressing (peerSessions is), so it never loses track of one session to another sharing the same peer id. private readonly allSessions = new Set(); + // -- 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; + advert: Readonly; + }[] { + return Array.from(this.knownDevices, ([deviceId, advert]) => ({ + deviceId, + advert, + })); + } + /** Registers a session in both peerSessions (addressing -- last one in for a given peer wins) and allSessions (shutdown -- every session, always). */ private trackSession(key: string, session: AcceptedMeshSession): void { this.peerSessions.set(key, session); @@ -405,6 +434,7 @@ export class WireMeshTransport implements MeshTransport { ): void { void (async () => { for await (const event of session.events) { + this.mergeKnownDevices(event.directory); this.reportPresenceAdvert(handle, deviceIdHex, event.directory); if (event.state.status === "closed") { const wasTracked = this.peerSessions.get(deviceIdHex) === session; diff --git a/src/test/gossip-directory-aggregation.integration.test.ts b/src/test/gossip-directory-aggregation.integration.test.ts new file mode 100644 index 00000000..e46ef311 --- /dev/null +++ b/src/test/gossip-directory-aggregation.integration.test.ts @@ -0,0 +1,151 @@ +/** + * WireMeshTransport's mesh-wide gossip directory aggregation: listKnownDevices() merges every live session's own peer-advert directory into one device-id-keyed view, so a consumer (agent listing, room discovery) can read every device this side has heard gossip from without reaching into per-session internals. This is the prerequisite P3.8's own room-discovery design and the eventual agent register/update/offline retirement both named as missing and blocking (agent-comms#48's own issue body, 2026-09-14 investigation) -- built here as its own foundational primitive, tested directly against real WireMeshTransport instances the same way presence-readvertise.integration.test.ts already does, deliberately bypassing MeshStore for the same reason that file gives. + */ + +import { test, describe, expect } from "vitest"; +import { generateIdentity } from "../core/identity.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; +import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; +import { WireMeshTransport } from "../core/wire-mesh-transport.js"; +import type { ConnectionHandle, TransportEvents } from "../core/transport.js"; +import { waitFor } from "./test-transport.js"; + +function inertEvents(): TransportEvents { + return { + onMessage: () => undefined, + onPeerConnected: () => undefined, + onPeerDisconnected: () => undefined, + onIntroduction: () => undefined, + onConnectionRequest: () => undefined, + onPeerList: () => undefined, + onPeerJoined: () => undefined, + onBecomeCoordinator: () => undefined, + onRevocationAnnounce: () => undefined, + onPresenceAdvert: () => undefined, + }; +} + +describe("WireMeshTransport.listKnownDevices", () => { + test("reflects a directly-connected peer's own gossiped advert, including its extension fields", async () => { + const identityA = generateIdentity(); + const identityB = generateIdentity(); + const peerIdA = deviceIdToHex( + await toIdentityPort(identityA).then((p) => p.deviceId), + ); + const peerIdB = deviceIdToHex( + await toIdentityPort(identityB).then((p) => p.deviceId), + ); + + const transportA = new WireMeshTransport( + inertEvents(), + identityA, + undefined, + undefined, + () => "busy", + ); + const transportB = new WireMeshTransport(inertEvents(), identityB); + + try { + await transportA.startDataServer(); + await transportB.connectToPeer( + { + id: peerIdA, + port: transportA.dataPort, + startedAt: new Date().toISOString(), + }, + peerIdB, + ); + + await waitFor( + () => + transportB + .listKnownDevices() + .find((entry) => entry.deviceId === peerIdA)?.advert[ + "presence/status" + ] === "busy", + "B's known-devices view includes A's gossiped busy status", + ); + + const known = transportB + .listKnownDevices() + .find((entry) => entry.deviceId === peerIdA); + expect(known?.advert["presence/status"]).toBe("busy"); + } finally { + await transportB.shutdown(); + await transportA.shutdown(); + } + }); + + test("has no entries before any peer connects", () => { + const identityA = generateIdentity(); + const transportA = new WireMeshTransport(inertEvents(), identityA); + expect(transportA.listKnownDevices()).toEqual([]); + }); + + test("keeps the newest advert per device across repeated gossip re-advertisement", async () => { + const identityA = generateIdentity(); + const identityB = generateIdentity(); + const peerIdA = deviceIdToHex( + await toIdentityPort(identityA).then((p) => p.deviceId), + ); + const peerIdB = deviceIdToHex( + await toIdentityPort(identityB).then((p) => p.deviceId), + ); + + let currentStatusA: "active" | "idle" = "active"; + const SHORT_PRESENCE_INTERVAL_MS = 50; + + const transportA = new WireMeshTransport( + inertEvents(), + identityA, + undefined, + undefined, + () => currentStatusA, + SHORT_PRESENCE_INTERVAL_MS, + ); + const transportB = new WireMeshTransport(inertEvents(), identityB); + + try { + await transportA.startDataServer(); + await transportB.connectToPeer( + { + id: peerIdA, + port: transportA.dataPort, + startedAt: new Date().toISOString(), + }, + peerIdB, + ); + + await waitFor( + () => + transportB + .listKnownDevices() + .find((entry) => entry.deviceId === peerIdA)?.advert[ + "presence/status" + ] === "active", + "B first observes A's initial active status", + ); + + currentStatusA = "idle"; + + await waitFor( + () => + transportB + .listKnownDevices() + .find((entry) => entry.deviceId === peerIdA)?.advert[ + "presence/status" + ] === "idle", + "B's known-devices view converges on A's latest re-advertised status", + ); + + expect( + transportB + .listKnownDevices() + .filter((entry) => entry.deviceId === peerIdA), + ).toHaveLength(1); + } finally { + await transportB.shutdown(); + await transportA.shutdown(); + } + }); +});