From 1995981dcb16fdffc94495b20d22dc62f04046a9 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:30:22 +0100 Subject: [PATCH 1/3] feat(core): resolve gossip-discovered agents in getAgent and sendDm AgentRegistry.getAgent only checked the local agents map, so a device known solely through gossip (never registered or state-synced locally) was unresolvable even though listAgents already merged it into its own returned array. Extend getAgent to fall back to the same listDiscoverableAgents merge listAgents uses, via a shared synthesiseDiscoveredAgent helper. RoomMessaging.sendDm checked recipient existence via a direct agents.get lookup, which can never see a gossip-only device. Route it through a new resolveAgent dependency (wired to AgentRegistry.getAgent) instead, so a DM to a gossip-discovered recipient is no longer rejected as AGENT_NOT_FOUND before ever reaching the transport layer. This is a prerequisite for cross-machine DMs (agent-comms#155), where a remote agent learned via the hub's gossip will never exist as a locally registered record. --- src/core/agent-registry.ts | 45 +++++++++----- src/core/mesh-store.ts | 3 +- src/core/room-messaging.ts | 6 +- src/test/agent-registry.test.ts | 62 ++++++++++++++++++++ src/test/room-messaging-durable-send.test.ts | 4 +- src/test/room-messaging.test.ts | 34 +++++++++-- 6 files changed, 128 insertions(+), 26 deletions(-) diff --git a/src/core/agent-registry.ts b/src/core/agent-registry.ts index a77e6cd6..abdbb10c 100644 --- a/src/core/agent-registry.ts +++ b/src/core/agent-registry.ts @@ -107,9 +107,17 @@ export class AgentRegistry { return agent; } + /** Resolves an agent by id, falling back to a gossip-discovered device (see listDiscoverableAgents' own doc) not otherwise locally known -- the same merge listAgents already applies to its returned array, but for a single lookup rather than the whole list. This is what lets a caller resolve a remote, hub-learned agent (agent-comms#155) that will never appear in deps.agents at all, since nothing replicates full agent records cross-machine the way a local peer's agent_upsert broadcast does. */ async getAgent(id: string): Promise { await Promise.resolve(); - return this.deps.agents.get(id); + const known = this.deps.agents.get(id); + if (known !== undefined) return known; + const discovered = this.listDiscoverableAgents().find( + (candidate) => candidate.deviceId === id, + ); + return discovered === undefined + ? undefined + : AgentRegistry.synthesiseDiscoveredAgent(discovered); } async updateAgent( @@ -156,23 +164,32 @@ export class AgentRegistry { } for (const discovered of this.listDiscoverableAgents()) { if (this.deps.agents.has(discovered.deviceId)) continue; - result.push({ - id: discovered.deviceId, - version: 0, - name: discovered.advert.name, - harness: discovered.advert.harness, - cwd: discovered.advert.cwd, - pid: discovered.advert.pid, - startedAt: discovered.advert.startedAt, - visibility: "visible", - status: discovered.status ?? "active", - tags: discovered.advert.tags, - subscribedRooms: discovered.advert.subscribedRooms, - }); + result.push(AgentRegistry.synthesiseDiscoveredAgent(discovered)); } return result; } + /** Builds the placeholder-shaped AgentIdentity a gossip-discovered device (never locally registered) is represented as -- shared by listAgents' own array merge and getAgent's single-lookup fallback, so both resolve an identical shape for the identical discovered device. */ + private static synthesiseDiscoveredAgent(discovered: { + deviceId: string; + advert: AgentSelfAdvert; + status: AgentStatus | undefined; + }): AgentIdentity { + return { + id: discovered.deviceId, + version: 0, + name: discovered.advert.name, + harness: discovered.advert.harness, + cwd: discovered.advert.cwd, + pid: discovered.advert.pid, + startedAt: discovered.advert.startedAt, + visibility: "visible", + status: discovered.status ?? "active", + tags: discovered.advert.tags, + subscribedRooms: discovered.advert.subscribedRooms, + }; + } + /** * Every agent this store has heard gossiped by another device but never registered or otherwise locally recorded -- the read half of P3.8's eventual agent register/update/offline retirement (agent-comms#48), mirroring listRooms' own room-discovery merge (#138). Never merged into this.deps.agents: a gossip hint is not the same as a real registration, and this store has nothing else authoritative to report for it. Only ever an agent that gossiped itself as "visible" (MeshStore's own selfAgentAdvert getter never advertises a hidden or ghost agent this way), so no ghost-filtering is needed here the way listAgents' own local-agent check needs. */ diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index ff09532e..f2be1dc4 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -233,9 +233,10 @@ export class MeshStore implements CommsStore { rooms: this.rooms, messages: this.messages, dms: this.dms, - agents: this.agents, requireIdentity: () => this.requireIdentity(), roomProtocol: this.roomProtocol, + // AgentRegistry doesn't exist yet at this point in the constructor -- deferred the same lazy-`this`-capture way DeliveryEngine's own sendRoomRequestToMember closure above is. + resolveAgent: async (id) => this.agentRegistry.getAgent(id), }); this.roomLifecycle = new RoomLifecycle({ diff --git a/src/core/room-messaging.ts b/src/core/room-messaging.ts index f2f6dff7..018c2529 100644 --- a/src/core/room-messaging.ts +++ b/src/core/room-messaging.ts @@ -18,14 +18,14 @@ import type { StreamingBehavior, } from "./types.js"; -/** The state and collaborators RoomMessaging needs from MeshStore. rooms/messages/dms/agents are direct references into MeshStore's own fields; roomProtocol is the already-constructed instance (construction order: ... -\> roomProtocol -\> roomMessaging -\> ...), narrowed to what sending a message or DM ever needs. */ +/** The state and collaborators RoomMessaging needs from MeshStore. rooms/messages/dms are direct references into MeshStore's own fields; roomProtocol is the already-constructed instance (construction order: ... -\> roomProtocol -\> roomMessaging -\> ...), narrowed to what sending a message or DM ever needs; resolveAgent is AgentRegistry's own getAgent (deferred the same lazy-closure way DeliveryEngine's sendRoomRequestToMember closure is, since AgentRegistry doesn't exist yet at RoomMessaging's own construction point) rather than a bare `agents.get` lookup, so sendDm's own existence check also resolves a gossip-discovered agent (a remote, hub-learned one included, agent-comms#155) that will never appear in the agents map directly. */ export interface RoomMessagingDeps { rooms: Map; messages: Map; dms: Map; - agents: Map; requireIdentity: () => MeshStoreIdentity; roomProtocol: Pick; + resolveAgent: (id: string) => Promise; } export class RoomMessaging { @@ -132,7 +132,7 @@ export class RoomMessaging { streamingBehavior?: StreamingBehavior, ): Promise { if (to !== from) { - const recipient = this.deps.agents.get(to); + const recipient = await this.deps.resolveAgent(to); if (!recipient) throw new CommsError(`Agent ${to} not found`, "AGENT_NOT_FOUND"); if (recipient.visibility === "ghost") diff --git a/src/test/agent-registry.test.ts b/src/test/agent-registry.test.ts index cdd58c17..5084486b 100644 --- a/src/test/agent-registry.test.ts +++ b/src/test/agent-registry.test.ts @@ -165,6 +165,68 @@ describe("AgentRegistry — getAgent", () => { await expect(registry.getAgent(a.id)).resolves.toEqual(a); await expect(registry.getAgent("nobody")).resolves.toBeUndefined(); }); + + it("falls back to a gossip-discovered device not locally registered, as the same placeholder shape listAgents synthesises (agent-comms#155: a remote, hub-learned agent never lands in deps.agents at all)", async () => { + const { registry, deps } = makeHarness(); + deps.requireTransport = () => + ({ + listKnownDevices: () => [ + { + deviceId: "remote-device", + advert: { + "agent/self": { + name: "remote-agent", + harness: "codex", + cwd: "/tmp/remote", + pid: 42, + startedAt: "2026-03-03T00:00:00.000Z", + tags: ["from-hub"], + subscribedRooms: [], + }, + "presence/status": "idle", + }, + }, + ], + }) as unknown as ReturnType; + + await expect(registry.getAgent("remote-device")).resolves.toMatchObject({ + id: "remote-device", + name: "remote-agent", + harness: "codex", + cwd: "/tmp/remote", + pid: 42, + visibility: "visible", + status: "idle", + tags: ["from-hub"], + }); + }); + + it("prefers a locally-known agent over a same-id gossip-discovered placeholder", async () => { + const { registry, deps } = makeHarness(); + const local = agent({ id: "dual-known", name: "local-copy" }); + deps.agents.set(local.id, local); + deps.requireTransport = () => + ({ + listKnownDevices: () => [ + { + deviceId: "dual-known", + advert: { + "agent/self": { + name: "stale-gossip-copy", + harness: "codex", + cwd: "/tmp", + pid: 1, + startedAt: "2026-01-01T00:00:00.000Z", + tags: [], + subscribedRooms: [], + }, + }, + }, + ], + }) as unknown as ReturnType; + + await expect(registry.getAgent("dual-known")).resolves.toEqual(local); + }); }); describe("AgentRegistry — updateAgent", () => { diff --git a/src/test/room-messaging-durable-send.test.ts b/src/test/room-messaging-durable-send.test.ts index 415bea96..47839415 100644 --- a/src/test/room-messaging-durable-send.test.ts +++ b/src/test/room-messaging-durable-send.test.ts @@ -23,7 +23,7 @@ import { RoomMessaging, type RoomMessagingDeps, } from "../core/room-messaging.js"; -import type { AgentIdentity, Room } from "../core/types.js"; +import type { Room } from "../core/types.js"; const NO_DELEGATIONS_REMAINING = 0; const MINUTES_PER_HOUR = 60; @@ -79,7 +79,6 @@ async function makeHarness() { rooms: new Map([[roomPath, room]]), messages: new Map(), dms: new Map(), - agents: new Map(), requireIdentity: () => ({ slot, clock, @@ -88,6 +87,7 @@ async function makeHarness() { dataStorage, }), roomProtocol: { sendRoomRequestToMember: async () => undefined }, + resolveAgent: async () => undefined, }; return { diff --git a/src/test/room-messaging.test.ts b/src/test/room-messaging.test.ts index 0aab88ba..23f887f3 100644 --- a/src/test/room-messaging.test.ts +++ b/src/test/room-messaging.test.ts @@ -66,11 +66,15 @@ function agent(overrides: Partial = {}): AgentIdentity { function makeHarness() { const sendRoomRequestToMember = vi.fn().mockResolvedValue(undefined); + // Backs the default resolveAgent mock below -- a plain lookup Map standing in for AgentRegistry.getAgent's own real "known, else gossip-discovered" resolution, which RoomMessaging itself never sees; it only calls resolveAgent opaquely. + const agents = new Map(); + const resolveAgent = vi + .fn() + .mockImplementation(async (id: string) => agents.get(id)); const deps: RoomMessagingDeps = { rooms: new Map(), messages: new Map(), dms: new Map(), - agents: new Map(), requireIdentity: () => ({ slot: { harness: "pi", cwd: "/tmp" }, clock: { now: () => NOW_MS }, @@ -79,8 +83,11 @@ function makeHarness() { dataStorage: {} as never, }), roomProtocol: { sendRoomRequestToMember }, + resolveAgent, }; return { + agents, + resolveAgent, deps, messaging: new RoomMessaging(deps), sendRoomRequestToMember, @@ -358,7 +365,7 @@ describe("RoomMessaging — sendDm", () => { it("throws AGENT_NOT_FOUND naming the exact recipient id when the recipient has ghost visibility", async () => { const h = makeHarness(); - h.deps.agents.set(TO_DEVICE_ID, agent({ visibility: "ghost" })); + h.agents.set(TO_DEVICE_ID, agent({ visibility: "ghost" })); await expect( h.messaging.sendDm(FROM_DEVICE_ID, TO_DEVICE_ID, "hi"), ).rejects.toMatchObject({ @@ -367,6 +374,21 @@ describe("RoomMessaging — sendDm", () => { }); }); + it("resolves the recipient via resolveAgent, not a direct agents-map lookup -- so a remote, gossip-discovered recipient (agent-comms#155) is a valid DM target", async () => { + const h = makeHarness(); + h.resolveAgent.mockResolvedValue(agent({ visibility: "visible" })); + vi.mocked(loadRoomTokens).mockReturnValue({ + [dmRoomPath(FROM_DEVICE_ID, TO_DEVICE_ID)]: FAKE_TOKEN, + }); + const message = await h.messaging.sendDm( + FROM_DEVICE_ID, + TO_DEVICE_ID, + "hi", + ); + expect(h.resolveAgent).toHaveBeenCalledWith(TO_DEVICE_ID); + expect(message.content).toBe("hi"); + }); + it("skips recipient validation entirely for a self-DM", async () => { const h = makeHarness(); const message = await h.messaging.sendDm( @@ -390,7 +412,7 @@ describe("RoomMessaging — sendDm", () => { it("stores a cross-agent DM under the sorted dmRoomPath key and dials the recipient", async () => { const h = makeHarness(); - h.deps.agents.set(TO_DEVICE_ID, agent()); + h.agents.set(TO_DEVICE_ID, agent()); vi.mocked(loadRoomTokens).mockReturnValue({ [dmRoomPath(FROM_DEVICE_ID, TO_DEVICE_ID)]: FAKE_TOKEN, }); @@ -441,7 +463,7 @@ describe("RoomMessaging — sendDm", () => { it("throws NOT_MEMBER naming the exact dm key when no room:member token is persisted for a cross-agent DM", async () => { const h = makeHarness(); - h.deps.agents.set(TO_DEVICE_ID, agent()); + h.agents.set(TO_DEVICE_ID, agent()); vi.mocked(loadRoomTokens).mockReturnValue({}); const key = dmRoomPath(FROM_DEVICE_ID, TO_DEVICE_ID); @@ -455,7 +477,7 @@ describe("RoomMessaging — sendDm", () => { it("includes streaming-behavior in the wire params only when given, for a cross-agent DM", async () => { const h = makeHarness(); - h.deps.agents.set(TO_DEVICE_ID, agent()); + h.agents.set(TO_DEVICE_ID, agent()); vi.mocked(loadRoomTokens).mockReturnValue({ [dmRoomPath(FROM_DEVICE_ID, TO_DEVICE_ID)]: FAKE_TOKEN, }); @@ -471,7 +493,7 @@ describe("RoomMessaging — sendDm", () => { it("omits streaming-behavior from the wire params when not given, for a cross-agent DM", async () => { const h = makeHarness(); - h.deps.agents.set(TO_DEVICE_ID, agent()); + h.agents.set(TO_DEVICE_ID, agent()); vi.mocked(loadRoomTokens).mockReturnValue({ [dmRoomPath(FROM_DEVICE_ID, TO_DEVICE_ID)]: FAKE_TOKEN, }); From 51cb202fe5c6c3d21826c8f70c014088d0913aee Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 17:48:51 +0100 Subject: [PATCH 2/3] feat(core): forward gossip and room-domain requests between the local mesh and the hub The gateway (agent-comms#154) held a hub connection but did nothing with it: no local agent was ever advertised onto it, nothing learned from it was merged back, and a room-domain request addressed to a device this side had no local session for simply failed as not_connected. This wires the actual forwarding, both directions. Outbound: whenever a local peer session's own gossiped directory changes, forward every agent/self-bearing entry (by construction only ever a visible agent, since MeshStore's selfAgentAdvert getter never gossips that key for a hidden or ghost one) onto the hub's own raw connection as a gossip frame, keyed by each device's own device-id. sendGossipUpdate can only ever advertise a session's own single device, so this rides HubSession's own tracked raw Connection instead. connectHub also pushes an immediate catch-up of every already-known local device, so a coordinator handover doesn't wait for each local peer's own next periodic gossip tick to re-advertise. Remote directory merge: HubSession now surfaces every remote directory entry the hub's gossip and catch-up carry, merged into the transport's own mesh-wide knownDevices via the same mergeKnownDevices helper a local peer session's directory already uses -- a hub-learned agent therefore surfaces in listAgents/getAgent with no separate lookup path. Routing: WireMeshTransport.sendRoomRequest falls back to routing through the hub's relay-connect/relay-data pairing when its target isn't a local peer session, bounded by a timeout since the hub silently drops a relay-connect to an unknown target-device with no error frame. HubSession's own inbound dispatch now sends a real room-domain verb (room.send, room.join, room.notify, ...) to the same roomVerbHandlers a local peer session already uses, with a per-request ConnectionHandle keyed by request.fromDevice -- the same "attribute to the true sender, never the hub" discipline already applied to the legacy opaque-frame path, extended to this one. A legacy state_sync or state_update carried this way is still dropped before it ever reaches onMessage, unchanged: it has no per-request capability token to verify it against, unlike a real room-domain verb. reportPresenceAdvert and the outbound-forwarding/routing helpers move out to gossip-directory.ts and a new hub-forwarding.ts respectively to keep wire-mesh-transport.ts under this repo's max-lines cap. --- src/core/gossip-directory.ts | 17 +++++- src/core/hub-forwarding.ts | 57 ++++++++++++++++++++ src/core/hub-session.ts | 95 ++++++++++++++++++++++++++++----- src/core/wire-mesh-transport.ts | 55 ++++++++++--------- 4 files changed, 181 insertions(+), 43 deletions(-) create mode 100644 src/core/hub-forwarding.ts diff --git a/src/core/gossip-directory.ts b/src/core/gossip-directory.ts index 34c3d2b5..08f93f32 100644 --- a/src/core/gossip-directory.ts +++ b/src/core/gossip-directory.ts @@ -1,10 +1,12 @@ /** - * 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. + * gossip-directory — the mesh-wide gossip-directory aggregation WireMeshTransport's own listKnownDevices reads from, plus per-event directory lookups (a specific peer's presence advert). 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"; +import { AgentStatus } from "./types.js"; +import { PRESENCE_GOSSIP_KEY } from "./wire-mesh-transport.js"; /** 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( @@ -22,3 +24,16 @@ export function mergeKnownDevices( } } } + +/** Reads a presence extension from one specific device's own gossiped self-advert, if this event's directory carries a fresh one for exactly that device-id -- never for any other device-id a multi-hop directory might mention, since only the session's own authenticated peer's advert is that session's business to report. Returns undefined for a missing presence/status key, or a value that isn't a recognised AgentStatus -- an advert simply not participating in this convention, not an error (the same verifier obligation peer-advert's own open extension tail is documented under). */ +export function findPresenceAdvert( + deviceIdHex: string, + directory: readonly DirectoryEntry[], +): AgentStatus | undefined { + const entry = directory.find( + (candidate) => deviceIdToHex(candidate.device) === deviceIdHex, + ); + if (entry === undefined) return undefined; + const status: unknown = entry.advert[PRESENCE_GOSSIP_KEY]; + return AgentStatus.is(status) ? status : undefined; +} diff --git a/src/core/hub-forwarding.ts b/src/core/hub-forwarding.ts new file mode 100644 index 00000000..cfd73681 --- /dev/null +++ b/src/core/hub-forwarding.ts @@ -0,0 +1,57 @@ +/** + * hub-forwarding -- agent-comms#155's gateway forwarding, split from wire-mesh-transport.ts under the repo's max-lines cap. Outbound: filters a directory of gossiped local devices down to the ones eligible for cross-machine advertisement (an agent/self-bearing entry, meaning only ever a "visible" agent -- see AGENT_SELF_GOSSIP_KEY's own doc for why no separate visibility check is needed here) and pushes them onto a connected HubSession, best-effort. Local-to-remote routing: falls a room-domain request back through the hub's own relay-connect/relay-data pairing when its target isn't a local peer session. + */ + +import type { + CapabilityScope, + CapabilityToken, + ManageCommand, + PeerAdvert, +} from "wire-mesh-core/generated/protocol"; +import type { + DirectoryEntry, + ManageOutcome, +} from "wire-mesh-core/domain/mesh-session"; +import type { HubSession } from "./hub-session.js"; +import { AGENT_SELF_GOSSIP_KEY } from "./wire-mesh-transport.js"; + +/** Forwards every directory entry carrying an agent/self extension onto hub, if hub currently holds a live connection -- a no-op otherwise, which is every local peer except whichever one is currently the gateway. Called both from WireMeshTransport's own watchForDisconnect (as each local peer session's directory changes) and from connectHub's own initial catch-up push, so this side's already-known local devices reach the hub immediately on taking over the gateway role rather than waiting for their own next periodic gossip tick. Forwards unconditionally on every call (no caching or dedup against a previous call), mirroring relay-hub.ts's own "forward every gossip frame, as received" philosophy: an advert's own snapshot-seconds/addresses are meant to keep propagating as a liveness heartbeat, so suppressing a "duplicate" would silently stop legitimate freshness updates from reaching remote gateways. A forward that fails is reported via onError and swallowed, matching every other best-effort gossip send in this codebase. */ +export function forwardAdvertsToHub( + hub: Readonly>, + directory: readonly DirectoryEntry[], + onError: ((error: Error) => void) | undefined, +): void { + if (!hub.isConnected) return; + const eligible = directory.filter( + (entry) => entry.advert[AGENT_SELF_GOSSIP_KEY] !== undefined, + ); + if (eligible.length === 0) return; + hub.advertiseDevices(eligible).catch((error: unknown) => { + onError?.(error instanceof Error ? error : new Error(String(error))); + }); +} + +/** Catches the hub up with every local device already known at the moment this side becomes the gateway (connectHub's own trailing call, right after hub.connect resolves) -- without this, a device whose own last gossip arrived before this coordinator took over the gateway role would never be (re-)advertised until its own next periodic gossip tick (hub-side state is rebuilt from scratch on every takeover, per coordinator-gateway.ts's own class doc). Reuses forwardAdvertsToHub's own eligibility filter, so only ever a visible, agent/self-bearing device is pushed, exactly as an ordinary directory-change forward would. */ +export function pushHubCatchUp( + hub: Readonly>, + knownDevices: ReadonlyMap>, + onError: ((error: Error) => void) | undefined, +): void { + const catchUp = [...knownDevices.values()].map((advert) => ({ + device: advert.device, + advert, + })); + forwardAdvertsToHub(hub, catchUp, onError); +} + +/** Routes a room-domain request through the hub's relay-connect/relay-data pairing when memberId isn't a local peer session -- WireMeshTransport.sendRoomRequest's own fallback, since the member may be a remote agent reachable only via this machine's gateway connection (agent-comms#155's local-to-remote leg). Resolves the same not_connected outcome sendRoomRequest already returned before the hub existed at all when this side isn't currently the gateway. */ +export async function routeRoomRequestViaHub( + hub: Readonly>, + memberId: string, + command: ManageCommand, + scope: Readonly, + token?: CapabilityToken, +): Promise { + if (!hub.isConnected) return { result: "error", code: "not_connected" }; + return hub.sendRoomRequest(memberId, command, scope, token); +} diff --git a/src/core/hub-session.ts b/src/core/hub-session.ts index fb8e4e02..f0c1a8f8 100644 --- a/src/core/hub-session.ts +++ b/src/core/hub-session.ts @@ -5,16 +5,32 @@ import { acceptMeshSession, type AcceptedMeshSession, + type DirectoryEntry, + type IncomingManageRequest, + type ManageOutcome, } 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 { + CapabilityScope, + CapabilityToken, + Frame, + ManageCommand, +} 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 { ConnectionHandle, 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"; +import { + buildCommand, + DOMAIN, + FRAME_SCOPE, + FRAME_VERB, +} from "./wire-mesh-transport.js"; + +/** How long a room-domain request routed through the hub's relay-connect/relay-data pairing waits for a response before giving up. Unlike an ordinary local peer session, a relay-connect naming an unknown target-device is silently dropped by the hub (spec/relay-hub's own documented behaviour -- no error frame exists for "no such device"), so a request to a device that turns out not to be reachable via any gateway would otherwise hang forever rather than surfacing as a normal "not reachable" outcome. */ +const HUB_ROOM_REQUEST_TIMEOUT_MS = 15_000; export interface HubSessionDeps { /** Resolves this node's own identity port once ready. */ @@ -31,10 +47,19 @@ export interface HubSessionDeps { ) => void | Promise; /** Tracks the session for shutdown -- every session the transport ever creates, always. */ trackForShutdown: (session: AcceptedMeshSession) => void; + /** Fires with every remote (non-self) directory entry the hub's own gossip/catch-up surfaces, every time the session's directory changes (agent-comms#155's own remote-directory-merge leg) -- the transport merges these into its own mesh-wide knownDevices the same way it already merges a local peer session's directory, so a hub-learned agent surfaces in listAgents/getAgent with no separate lookup path. */ + onDirectory: (entries: readonly DirectoryEntry[]) => void; + /** Dispatches an already-received, non-legacy-frame request (a real core/room verb: room.send, room.notify, room.join, ...) to WireMeshTransport's own roomRouter, the identical dispatch a local peer session's own drainSession already uses -- the outbound half of agent-comms#155's "remote to local" routing leg. Deferred the same lazy-`this`-capture way DeliveryEngine's own sendRoomRequestToMember closure is, since roomRouter is constructed after this class's own instance in WireMeshTransport's constructor. */ + handleRoomRequest: ( + request: IncomingManageRequest, + handle: Readonly, + ) => Promise; } export class HubSession { private session: AcceptedMeshSession | undefined; + /** The raw connection underlying `session` -- MeshSession's own sendGossipUpdate only ever advertises this side's own single device, so advertising OTHER (local mesh) devices onto the hub (agent-comms#155's outbound leg) needs a raw gossip frame sent directly, bypassing that per-session single-self-advert limit. */ + private connection: Connection | undefined; private readonly hubPeersKnown = new Set(); constructor(private readonly deps: Readonly) {} @@ -55,6 +80,7 @@ export class HubSession { const session = this.session; if (session === undefined) return; this.session = undefined; + this.connection = undefined; await session.close(); } @@ -80,17 +106,20 @@ export class HubSession { return; } this.session = session; + this.connection = connection; 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. + // Merge the hub's directory (its catch-up arrives as the first session events) and keep refreshing it on every subsequent one. Every remote entry is also surfaced via onDirectory, so the transport can merge it into its own mesh-wide knownDevices (agent-comms#155's remote-directory-merge leg). void (async () => { + const ownHex = deviceIdToHex(identity.deviceId); 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); - } + const remoteEntries = event.directory.filter( + (entry) => deviceIdToHex(entry.device) !== ownHex, + ); + for (const entry of remoteEntries) { + this.hubPeersKnown.add(deviceIdToHex(entry.device)); } + if (remoteEntries.length > 0) this.deps.onDirectory(remoteEntries); if (event.state.status === "closed") break; } })(); @@ -100,7 +129,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. 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). */ + /** 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 every downstream consumer sees the true origin, never the hub -- the same discipline extends to a real room-domain verb (agent-comms#155's "remote to local" leg) as it already applied to the legacy opaque-frame path. A legacy FRAME_VERB carrying 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). A real room-domain verb (room.send, room.join, room.notify, ...) carries no equivalent risk -- it is independently gated by its own room:member capability token, verified regardless of which transport path it arrived over -- so it is dispatched to the same roomVerbHandlers a local peer session's own drainSession uses, via handleRoomRequest. */ private consume(session: AcceptedMeshSession): void { void (async () => { for await (const request of session.incomingManageRequests) { @@ -110,11 +139,16 @@ export class HubSession { ? deviceIdToHex(request.fromDevice) : "hub-peer"; this.hubPeersKnown.add(senderHex); - const message = extractMessage(request.command); - if (message !== undefined && !isStateMutatingMessage(message)) { - this.deps.events.onMessage({ id: senderHex }, message); + const handle: Readonly = { id: senderHex }; + if (request.command.verb === FRAME_VERB) { + const message = extractMessage(request.command); + if (message !== undefined && !isStateMutatingMessage(message)) { + this.deps.events.onMessage(handle, message); + } + await request.respond({ result: "ok" }).catch(() => undefined); + continue; } - await request.respond({ result: "ok" }).catch(() => undefined); + await this.deps.handleRoomRequest(request, handle); } })(); } @@ -125,6 +159,7 @@ export class HubSession { } if (this.session === session) { this.session = undefined; + this.connection = undefined; } } @@ -140,6 +175,38 @@ export class HubSession { hexToBytes(peerDeviceHex), ); } + + /** Gossips a raw `gossip` frame carrying every given entry's own advert onto the hub, over the raw connection rather than sendGossipUpdate (which can only ever advertise this side's own single device) -- the outbound leg of agent-comms#155's gateway forwarding. relay-hub.ts registers each advert's own `device` field against the CONNECTION it arrives on, so every entry passed here becomes reachable, cross-machine, as "via this gateway" -- callers are responsible for only ever passing entries that are actually meant to be advertised (WireMeshTransport filters to local, visible-agent-bearing entries before calling this). Throws if this side isn't currently connected to a hub, matching sendToPeer's own contract -- callers gate on isConnected first. */ + async advertiseDevices(entries: readonly DirectoryEntry[]): Promise { + const connection = this.connection; + if (connection === undefined) { + throw new Error("not connected to a hub"); + } + await connection.send({ + type: "gossip", + peers: entries.map((entry) => entry.advert), + }); + } + + /** Sends a real core/room manage-request to a specific hub-reachable peer, through the hub's relay-connect/relay-data pairing -- the local-to-remote leg of agent-comms#155's routing, mirroring MeshTransport.sendRoomRequest's own not_connected/outcome contract so WireMeshTransport can fall back to this uniformly when memberId isn't a local peer session. Bounded by HUB_ROOM_REQUEST_TIMEOUT_MS (see its own doc) since an unreachable target-device is silently dropped by the hub with no error frame, unlike a local session's own connection-level failure. */ + async sendRoomRequest( + peerDeviceHex: string, + command: ManageCommand, + scope: Readonly, + token?: CapabilityToken, + ): Promise { + const session = this.session; + if (session === undefined) { + return { result: "error", code: "not_connected" }; + } + return session.sendManageRequest( + command, + scope, + hexToBytes(peerDeviceHex), + token, + HUB_ROOM_REQUEST_TIMEOUT_MS, + ); + } } function hexToBytes(hex: string): Uint8Array { diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 63419912..e141260c 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -11,11 +11,15 @@ 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 { findPresenceAdvert, mergeKnownDevices } from "./gossip-directory.js"; +import { + forwardAdvertsToHub, + pushHubCatchUp, + routeRoomRequestViaHub, +} from "./hub-forwarding.js"; import { acceptMeshSession, type AcceptedMeshSession, - type DirectoryEntry, type IncomingManageRequest, type ManageOutcome, } from "wire-mesh-core/domain/mesh-session"; @@ -52,7 +56,7 @@ import type { import type { PeerIdentity } from "./identity.js"; import { toIdentityPort } from "./wire-mesh-identity.js"; import { nanoid } from "./nanoid.js"; -import { AgentStatus } from "./types.js"; +import type { AgentStatus } from "./types.js"; import { createRoomRouter, extractMessage, @@ -82,7 +86,7 @@ const PRESENCE_READVERTISE_INTERVAL_MS = const LISTENER_ID_LENGTH = 8; /** The domain-qualified gossip extension key this transport reads/writes presence under, per wire-mesh's own gossip-extension-namespacing convention (spec/CONVENTIONS.md): `/`, never a bare name a second application's own extension could collide with. */ -const PRESENCE_GOSSIP_KEY = "presence/status"; +export const PRESENCE_GOSSIP_KEY = "presence/status"; /** The domain-qualified gossip extension key this transport writes this side's own currently-hosted public/private rooms under -- the write half of P3.8's room-discovery replacement for createRoom's own broadcastPatch (agent-comms#48). Same namespacing convention as PRESENCE_GOSSIP_KEY. */ const HOSTED_ROOMS_GOSSIP_KEY = "room/hosted"; @@ -96,7 +100,7 @@ export interface HostedRoomAdvert { } /** The domain-qualified gossip extension key this transport writes this side's own agent identity facts under -- the write half of P3.8's eventual agent register/update/offline retirement (agent-comms#48). Same namespacing convention as PRESENCE_GOSSIP_KEY/HOSTED_ROOMS_GOSSIP_KEY. */ -const AGENT_SELF_GOSSIP_KEY = "agent/self"; +export const AGENT_SELF_GOSSIP_KEY = "agent/self"; /** The lightweight, gossip-safe shape an agent advertises itself under: enough for a peer with no prior local record of this device to construct a real AgentIdentity-shaped discovery entry. Deliberately excludes status (already carried separately under presence/status, no need to duplicate it here) and visibility (this field is only ever populated for a "visible" agent in the first place -- see MeshStore's own selfAgentAdvert getter -- so a discovered entry's visibility is always exactly "visible" by construction, never something this advert needs to assert itself). */ export interface AgentSelfAdvert { @@ -271,6 +275,11 @@ export class WireMeshTransport implements MeshTransport { privateKeyPem: identity.privateKey, }); this.identityReady = toIdentityPort(identity); + // Built before this.hub below (roomRouter has no dependency on it) so the hub can be wired with a direct this.roomRouter.handleRequest reference rather than a lazy closure. + this.roomRouter = createRoomRouter({ + events, + ...(roomVerbHandlers !== undefined ? { handlers: roomVerbHandlers } : {}), + }); this.hub = new HubSession({ identityReady: this.identityReady, events: this.events, @@ -280,10 +289,10 @@ export class WireMeshTransport implements MeshTransport { trackForShutdown: (session) => { this.allSessions.add(session); }, - }); - this.roomRouter = createRoomRouter({ - events, - ...(roomVerbHandlers !== undefined ? { handlers: roomVerbHandlers } : {}), + onDirectory: (entries) => { + mergeKnownDevices(this.knownDevices, entries); + }, + handleRoomRequest: this.roomRouter.handleRequest, }); this.pendingConnectionTimeoutMs = pendingConnectionTimeoutMs; this.getCurrentPresence = getCurrentPresence; @@ -531,21 +540,6 @@ export class WireMeshTransport implements MeshTransport { })(); } - /** Surfaces a presence extension from the remote peer's own gossiped self-advert, if this event's directory carries a fresh one for exactly this session's peer -- never for any other device-id a multi-hop directory might mention, since only the session's own authenticated peer's advert is this session's business to report. A missing presence/status key, or a value that isn't a recognised AgentStatus, is silently ignored: an advert simply not participating in this convention, not an error (the same verifier obligation peer-advert's own open extension tail is documented under). */ - private reportPresenceAdvert( - handle: Readonly, - deviceIdHex: string, - directory: readonly DirectoryEntry[], - ): void { - const entry = directory.find( - (candidate) => deviceIdToHex(candidate.device) === deviceIdHex, - ); - if (entry === undefined) return; - const status: unknown = entry.advert[PRESENCE_GOSSIP_KEY]; - if (!AgentStatus.is(status)) return; - this.events.onPresenceAdvert(handle, status); - } - private watchForDisconnect( session: AcceptedMeshSession, handle: Readonly, @@ -554,7 +548,11 @@ export class WireMeshTransport implements MeshTransport { void (async () => { for await (const event of session.events) { mergeKnownDevices(this.knownDevices, event.directory); - this.reportPresenceAdvert(handle, deviceIdHex, event.directory); + forwardAdvertsToHub(this.hub, event.directory, this.events.onError); + const presence = findPresenceAdvert(deviceIdHex, event.directory); + if (presence !== undefined) { + this.events.onPresenceAdvert(handle, presence); + } if (event.state.status === "closed") { const wasTracked = this.peerSessions.get(deviceIdHex) === session; if (wasTracked) this.peerSessions.delete(deviceIdHex); @@ -787,10 +785,10 @@ export class WireMeshTransport implements MeshTransport { token?: CapabilityToken, ): Promise { const session = this.peerSessions.get(memberId); - if (session === undefined) { - return { result: "error", code: "not_connected" }; + if (session !== undefined) { + return session.sendManageRequest(command, scope, undefined, token); } - return session.sendManageRequest(command, scope, undefined, token); + return routeRoomRequestViaHub(this.hub, memberId, command, scope, token); } // ----------------------------------------------------------------------- @@ -953,6 +951,7 @@ export class WireMeshTransport implements MeshTransport { async connectHub(url: string): Promise { await this.hub.connect(url); + pushHubCatchUp(this.hub, this.knownDevices, this.events.onError); } async disconnectHub(): Promise { From 18eee5f0379af25971fc84750019a7f3c4e80fb2 Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Thu, 17 Sep 2026 18:02:25 +0100 Subject: [PATCH 3/3] test(core): cover gateway forwarding end to end over a real relay hub wireTestTransport never wired WireMeshTransport's getSelfAgentAdvert parameter, unlike bridge-mesh.ts's own production wiring -- every test built on it therefore never gossiped the agent/self extension at all, regardless of registerAgent being called, which silently made listKnownDevices/listAgents' own gossip-discovery merge untestable through the standard helper. Wire it the same way production does. Add integration coverage for agent-comms#155 using two independent local meshes, each its own coordinator dialling a shared real relay hub, never directly connected to each other: a non-gateway local peer on one machine is discovered by the other machine's gateway purely via hub gossip; a hidden or ghost agent is never advertised across it; and a DM to a remote gateway's own agent, including its two-round consent handshake, routes correctly through the hub's relay pairing. Document the one confirmed limitation this surfaced: wire-mesh-core's own session layer tracks at most one active relay pairing per remote device, so an inbound hub-relayed room-domain request is dispatched as addressed to this gateway's own agent unconditionally -- there is no target-device disambiguation available yet to route it on to a different local peer this same gateway also advertises. --- src/core/hub-session.ts | 2 +- .../gateway-forwarding.integration.test.ts | 238 ++++++++++++++++++ src/test/test-transport.ts | 3 +- 3 files changed, 241 insertions(+), 2 deletions(-) create mode 100644 src/test/gateway-forwarding.integration.test.ts diff --git a/src/core/hub-session.ts b/src/core/hub-session.ts index f0c1a8f8..30b798d9 100644 --- a/src/core/hub-session.ts +++ b/src/core/hub-session.ts @@ -129,7 +129,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 every downstream consumer sees the true origin, never the hub -- the same discipline extends to a real room-domain verb (agent-comms#155's "remote to local" leg) as it already applied to the legacy opaque-frame path. A legacy FRAME_VERB carrying 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). A real room-domain verb (room.send, room.join, room.notify, ...) carries no equivalent risk -- it is independently gated by its own room:member capability token, verified regardless of which transport path it arrived over -- so it is dispatched to the same roomVerbHandlers a local peer session's own drainSession uses, via handleRoomRequest. */ + /** 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 every downstream consumer sees the true origin, never the hub -- the same discipline extends to a real room-domain verb (agent-comms#155's "remote to local" leg) as it already applied to the legacy opaque-frame path. A legacy FRAME_VERB carrying 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). A real room-domain verb (room.send, room.join, room.notify, ...) carries no equivalent risk -- it is independently gated by its own room:member capability token, verified regardless of which transport path it arrived over -- so it is dispatched to the same roomVerbHandlers a local peer session's own drainSession uses, via handleRoomRequest, against THIS side's own local mesh state. Known limitation, inherited from wire-mesh-core's own session layer rather than something agent-comms can fix here: a session tracks at most one active relay pairing per remote device (mesh-session.ts's own single relayPeerDevice slot), so a request relayed here is dispatched as "addressed to this gateway's own agent" unconditionally -- there is no target-device disambiguation available to route it on to a DIFFERENT local peer this gateway also advertises. Forwarding this gateway's own agent's traffic is therefore correct; a remote request genuinely meant for another local peer behind this same gateway is not yet distinguishable from one meant for this gateway's own agent. */ private consume(session: AcceptedMeshSession): void { void (async () => { for await (const request of session.incomingManageRequests) { diff --git a/src/test/gateway-forwarding.integration.test.ts b/src/test/gateway-forwarding.integration.test.ts new file mode 100644 index 00000000..504c7d1c --- /dev/null +++ b/src/test/gateway-forwarding.integration.test.ts @@ -0,0 +1,238 @@ +/** + * Integration tests for the gateway actually forwarding between the local mesh and the hub (agent-comms#155), the core leg of the cross-machine mesh epic (#153) on top of #154's own connection-lifecycle-only gateway. Two independent local meshes ("machine A" and "machine B"), each its own coordinator dialling a shared real relay hub (createRelayHub, the same domain logic the production mesh.exadev.io Durable Object runs, served over local WebSockets via hub-helpers.ts) -- neither mesh is ever directly connected to the other, so anything either side learns about the other can only have arrived via the hub. + * + * Machine A's coordinator (a1) additionally has a second, ordinary local peer (a2) -- proving outbound advertisement and the remote-directory merge work for a local peer that is NOT itself the gateway, not just for the gateway's own agent (which #154 already exercised end-to-end before this issue). + */ + +import { afterEach, describe, expect, it } from "vitest"; +import { MeshStore } from "../core/mesh-store.js"; +import { dmRoomPath } from "../core/room-path.js"; +import { realHubOverWs } from "./hub-helpers.js"; +import { wireTestTransport } from "./test-transport.js"; + +let nextPort = 22_400; +function freshPort(): number { + nextPort += 1; + return nextPort; +} + +/** Short enough that a gossip re-advertisement (which carries the registerAgent'd agent/self extension) fires within a test's own poll budget, unlike the 20s production default. */ +const FAST_GOSSIP_INTERVAL_MS = 50; + +const POLL_ATTEMPTS = 100; +const POLL_INTERVAL_MS = 50; +const sleep = async (ms: number): Promise => + new Promise((resolve) => { + setTimeout(resolve, ms); + }); + +/** Polls an async predicate until it holds, matching broadcast-window.integration.test.ts's own established pattern for this -- the shared test-transport.ts waitFor takes a synchronous condition, which can't await a MeshStore's own async listAgents/serialise-derived checks. */ +async function waitFor( + what: string, + check: () => Promise, +): Promise { + for (let i = 0; i < POLL_ATTEMPTS; i++) { + if (await check()) return; + await sleep(POLL_INTERVAL_MS); + } + expect(false, `timed out waiting for ${what}`).toBeTruthy(); +} + +const cleanups: (() => Promise)[] = []; + +afterEach(async () => { + for (const close of cleanups.splice(0)) { + await close(); + } +}); + +describe("gateway forwarding", () => { + it("advertises a local peer's agent (not just the gateway's own) onto the hub, and a separate machine's gateway merges it into its own knownDevices/listAgents", async () => { + const hub = await realHubOverWs(); + cleanups.push(hub.close); + + const a1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(a1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await a1.init(); + cleanups.push(async () => a1.shutdown()); + + const a2 = new MeshStore(a1.coordinatorPort, hub.url); + await wireTestTransport(a2, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await a2.init(); + cleanups.push(async () => a2.shutdown()); + await a2.registerAgent({ + name: "a2-local-peer", + harness: "test", + cwd: "/test/a2", + pid: process.pid, + visibility: "visible", + tags: ["from-a2"], + }); + + const b1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(b1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await b1.init(); + cleanups.push(async () => b1.shutdown()); + + await waitFor( + "b1 (a separate machine's gateway) to learn of a2 (a non-gateway local peer on machine A) via the hub", + async () => { + const agents = await b1.listAgents(b1.peerId); + return agents.some((agent) => agent.id === a2.peerId); + }, + ); + + const agents = await b1.listAgents(b1.peerId); + const discovered = agents.find((agent) => agent.id === a2.peerId); + expect(discovered).toMatchObject({ + id: a2.peerId, + name: "a2-local-peer", + harness: "test", + cwd: "/test/a2", + visibility: "visible", + tags: ["from-a2"], + }); + }); + + it("never advertises a hidden or ghost agent across the hub", async () => { + const hub = await realHubOverWs(); + cleanups.push(hub.close); + + const a1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(a1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await a1.init(); + cleanups.push(async () => a1.shutdown()); + + const aHidden = new MeshStore(a1.coordinatorPort, hub.url); + await wireTestTransport( + aHidden, + undefined, + undefined, + FAST_GOSSIP_INTERVAL_MS, + ); + await aHidden.init(); + cleanups.push(async () => aHidden.shutdown()); + await aHidden.registerAgent({ + name: "hidden-peer", + harness: "test", + cwd: "/test/hidden", + pid: process.pid, + visibility: "hidden", + tags: [], + }); + + const aGhost = new MeshStore(a1.coordinatorPort, hub.url); + await wireTestTransport( + aGhost, + undefined, + undefined, + FAST_GOSSIP_INTERVAL_MS, + ); + await aGhost.init(); + cleanups.push(async () => aGhost.shutdown()); + await aGhost.registerAgent({ + name: "ghost-peer", + harness: "test", + cwd: "/test/ghost", + pid: process.pid, + visibility: "ghost", + tags: [], + }); + + // A visible control peer on the identical mesh, registered after the hidden/ghost ones -- its own arrival on b1's side is the proof gossip genuinely had time to propagate, ruling out "nothing arrived at all" as a false-negative explanation for hidden/ghost never showing up below. + const aVisible = new MeshStore(a1.coordinatorPort, hub.url); + await wireTestTransport( + aVisible, + undefined, + undefined, + FAST_GOSSIP_INTERVAL_MS, + ); + await aVisible.init(); + cleanups.push(async () => aVisible.shutdown()); + await aVisible.registerAgent({ + name: "visible-peer", + harness: "test", + cwd: "/test/visible", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + const b1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(b1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await b1.init(); + cleanups.push(async () => b1.shutdown()); + + await waitFor( + "b1 to see the visible control peer, proving gossip had time to propagate", + async () => { + const agents = await b1.listAgents(b1.peerId); + return agents.some((agent) => agent.id === aVisible.peerId); + }, + ); + + const agents = await b1.listAgents(b1.peerId); + expect(agents.some((agent) => agent.id === aHidden.peerId)).toBe(false); + expect(agents.some((agent) => agent.id === aGhost.peerId)).toBe(false); + }); + + it("routes a DM to a remote gateway's own agent through the hub's relay pairing, delivered with the true sender attributed", async () => { + const hub = await realHubOverWs(); + cleanups.push(hub.close); + + const a1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(a1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await a1.init(); + cleanups.push(async () => a1.shutdown()); + await a1.registerAgent({ + name: "a1-gateway", + harness: "test", + cwd: "/test/a1", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + const b1 = new MeshStore(freshPort(), hub.url); + await wireTestTransport(b1, undefined, undefined, FAST_GOSSIP_INTERVAL_MS); + await b1.init(); + cleanups.push(async () => b1.shutdown()); + await b1.registerAgent({ + name: "b1-gateway", + harness: "test", + cwd: "/test/b1", + pid: process.pid, + visibility: "visible", + tags: [], + }); + + // Two-round DM consent (section 6), exactly like the local-mesh-only dm-admission.integration.test.ts -- b1's own outbound request is what's actually exercised via the new hub-relay fallback, since a1 is not a local peer of b1's own mesh. + const dmPath = dmRoomPath(b1.peerId, a1.peerId); + const requestPromise = b1.requestDmAccess(a1.peerId); + await waitFor("a1 to see b1's pending DM request via the hub", async () => { + await Promise.resolve(); + return a1.listPendingRoomJoins().length === 1; + }); + const [pending] = a1.listPendingRoomJoins(); + expect(pending?.roomPath).toBe(dmPath); + expect(pending?.requesterId).toBe(b1.peerId); + a1.acceptRoomJoin(dmPath, b1.peerId); + await requestPromise; + + const message = await b1.sendDm(b1.peerId, a1.peerId, "hello from b1"); + + await waitFor("a1 to receive the DM via the hub", async () => { + const a1Dms = a1.serialise().dms[dmPath] ?? []; + return a1Dms.some((m) => m.id === message.id); + }); + + const delivered = (a1.serialise().dms[dmPath] ?? []).find( + (m) => m.id === message.id, + ); + expect(delivered).toMatchObject({ + from: b1.peerId, + to: a1.peerId, + content: "hello from b1", + }); + }); +}); diff --git a/src/test/test-transport.ts b/src/test/test-transport.ts index e913ad73..98ed7ea5 100644 --- a/src/test/test-transport.ts +++ b/src/test/test-transport.ts @@ -17,7 +17,7 @@ import type { MeshStore } from "../core/mesh-store.js"; /** Short enough to keep generated test identity slots readable, long enough that two concurrent test runs won't collide on the same throwaway cwd. */ const TEST_IDENTITY_CWD_ID_LENGTH = 8; -/** Wires store onto a fresh WireMeshTransport and a persisted identity slot, returning the slot so a test can inspect (or reuse) the persisted room tokens directly via loadRoomTokens(). Defaults to a throwaway temp-dir slot per call -- pass an explicit slot when a test needs the same identity to survive across more than one wireTestTransport call (e.g. simulating a restart). pendingConnectionTimeoutMs overrides WireMeshTransport's own default 5-minute connect_request expiry -- a test proving that expiry behaviour needs it far shorter than any real approval window. presenceReadvertiseIntervalMs likewise overrides the default 20s presence re-advertisement cadence -- a test proving that behaviour needs it far shorter too, or a test that doesn't care about presence at all wants it long enough to never fire spuriously mid-test. */ +/** Wires store onto a fresh WireMeshTransport and a persisted identity slot, returning the slot so a test can inspect (or reuse) the persisted room tokens directly via loadRoomTokens(). Defaults to a throwaway temp-dir slot per call -- pass an explicit slot when a test needs the same identity to survive across more than one wireTestTransport call (e.g. simulating a restart). pendingConnectionTimeoutMs overrides WireMeshTransport's own default 5-minute connect_request expiry -- a test proving that expiry behaviour needs it far shorter than any real approval window. presenceReadvertiseIntervalMs likewise overrides the default 20s presence re-advertisement cadence -- a test proving that behaviour needs it far shorter too, or a test that doesn't care about presence at all wants it long enough to never fire spuriously mid-test. Wires getSelfAgentAdvert (`() => store.selfAgentAdvert`) the same way createBridgeMesh does -- a test relying on gossip carrying the registered agent/self extension (e.g. remote-directory-merge coverage, agent-comms#155) needs this wired exactly like production does. */ export async function wireTestTransport( store: MeshStore, slot?: Readonly, @@ -44,6 +44,7 @@ export async function wireTestTransport( presenceReadvertiseIntervalMs, undefined, dataStorage, + () => store.selfAgentAdvert, ), ); store.setIdentity({