From 460c9556f044dbeae566ae6fa7eb346e4adfc91f Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 16 Sep 2026 06:31:54 +0100 Subject: [PATCH] feat(core)!: retire federation.ts in favour of wire-mesh's own connectToRemote federation.ts (674 lines) was a fully separate, parallel transport stack: its own TLS sockets, its own X.509 certificate-fingerprint trust model, and its own hand-rolled fed_* wire framing over wire-protocol.ts's legacy newline-delimited JSON -- none of it routed through WireMeshTransport at all. It predates the wire-mesh substrate migration and was never updated to use it. The cross-machine connection problem federation.ts solved for itself is already solved generically: connectToRemote/acceptConnection/ rejectConnection (already wired through WireMeshTransport, already human-approval-gated per connection) reach a peer on any host, not just localhost. The one thing federation.ts additionally provided -- scoping which rooms/agents cross a mesh boundary via Room.federated and a pre-declared trust allowlist -- has no replacement here: this is a full retirement, not a swap, and any two meshes that connect via connectToRemote now become one shared mesh, matching how agent_upsert/ room_upsert were already confirmed to be genuinely mesh-wide by design (agent-comms#48) rather than needing a narrower audience. Removes FederationManager, FederationBridge, Room.federated, the nine mesh_fed_* tool actions and their MeshOnlyFeatures methods, and the fed_* MeshMessage wire-protocol variants those actions' own wire framing used. deliverRoomMessageToMember, added in the previous PR specifically for FederationBridge's own onRoomMessage, is removed as dead code along with its only caller. --- src/core/agent-registry.ts | 12 +- src/core/bridge.ts | 22 - src/core/comms-store.ts | 2 +- src/core/delivery-engine.ts | 19 +- src/core/federation-bridge.ts | 151 ---- src/core/federation.ts | 674 ------------------ src/core/index.ts | 2 - src/core/mesh-store.ts | 71 +- src/core/room-lifecycle.ts | 22 +- src/core/room-messaging.ts | 9 +- src/core/room-router.ts | 13 +- src/core/tool.ts | 142 +--- src/core/types.ts | 29 - src/core/wire-protocol.ts | 17 +- src/test/agent-registry.test.ts | 46 +- .../delivery-engine-directed-notify.test.ts | 58 -- src/test/delivery-engine.test.ts | 21 +- src/test/federation-bridge.test.ts | 396 ---------- src/test/federation.integration.test.ts | 300 -------- src/test/mesh-store-orchestration.test.ts | 67 +- src/test/room-lifecycle-membership.test.ts | 43 -- src/test/room-lifecycle-remote.test.ts | 16 +- src/test/room-lifecycle.test.ts | 61 -- src/test/room-messaging-durable-send.test.ts | 1 - src/test/room-messaging.test.ts | 15 - 25 files changed, 18 insertions(+), 2191 deletions(-) delete mode 100644 src/core/federation-bridge.ts delete mode 100644 src/core/federation.ts delete mode 100644 src/test/federation-bridge.test.ts delete mode 100644 src/test/federation.integration.test.ts diff --git a/src/core/agent-registry.ts b/src/core/agent-registry.ts index a24635ab..a77e6cd6 100644 --- a/src/core/agent-registry.ts +++ b/src/core/agent-registry.ts @@ -4,13 +4,12 @@ import { CommsError } from "./store.js"; import type { DeliveryEngine } from "./delivery-engine.js"; -import type { FederationManager } from "./federation.js"; import type { MeshTransport } from "./transport.js"; import type { AgentSelfAdvert } from "./wire-mesh-transport.js"; import { AgentStatus } from "./types.js"; import type { AgentIdentity, Visibility } from "./types.js"; -/** The state and collaborators AgentRegistry needs from MeshStore. agents/identityCache are direct references into MeshStore's own fields; startedAt is a readonly value copied once; deliveryEngine and federation are the already-constructed instances, narrowed to what agent-lifecycle bookkeeping ever needs. */ +/** The state and collaborators AgentRegistry needs from MeshStore. agents/identityCache are direct references into MeshStore's own fields; startedAt is a readonly value copied once; deliveryEngine is the already-constructed instance, narrowed to what agent-lifecycle bookkeeping ever needs. */ export interface AgentRegistryDeps { agents: Map; identityCache: Map; @@ -24,10 +23,6 @@ export interface AgentRegistryDeps { | "notifyRoomsOfStatus" | "notifyRoomsOfNameChange" >; - federation: Pick< - FederationManager, - "broadcastAgentVisible" | "broadcastAgentGone" - >; } /** Narrows an untrusted gossiped value (WireMeshTransport.listKnownDevices' own advert["agent/self"], self-asserted by whichever peer advertised it) into an AgentSelfAdvert -- a malformed or non-conforming entry is silently skipped rather than treated as an error, the same convention room-lifecycle.ts's own isHostedRoomAdvert already established for the identical class of gossip consumption. */ @@ -109,10 +104,6 @@ export class AgentRegistry { type: "agent_upsert", agent, }); - // Broadcast presence to federated links - if (agent.visibility === "visible") { - await this.deps.federation.broadcastAgentVisible(agent); - } return agent; } @@ -227,7 +218,6 @@ export class AgentRegistry { type: "agent_offline", agentId: id, }); - await this.deps.federation.broadcastAgentGone(id); } } } diff --git a/src/core/bridge.ts b/src/core/bridge.ts index 7d8456e7..c79afc48 100644 --- a/src/core/bridge.ts +++ b/src/core/bridge.ts @@ -64,9 +64,6 @@ export const MCP_TOOL_PARAMS = z.object({ "mesh_listeners", "mesh_set_visibility", "mesh_get_visibility", - "mesh_fed_connect", - "mesh_fed_disconnect", - "mesh_fed_links", ]), name: z.string().optional(), visibility: VisibilityEnum.optional(), @@ -346,25 +343,6 @@ export function buildAction(params: Record): CommsAction { }; return result; } - case "mesh_fed_connect": { - if (p.host === undefined) - throw new BuildActionError("mesh_fed_connect", "host"); - if (p.port === undefined) - throw new BuildActionError("mesh_fed_connect", "port"); - const result: CommsAction & { action: "mesh_fed_connect" } = { - action: "mesh_fed_connect", - host: p.host, - port: p.port, - }; - if (p.name !== undefined) result.name = p.name; - return result; - } - case "mesh_fed_disconnect": - if (p.id === undefined) - throw new BuildActionError("mesh_fed_disconnect", "id"); - return { action: "mesh_fed_disconnect", linkId: p.id }; - case "mesh_fed_links": - return { action: "mesh_fed_links" }; default: return p.action satisfies never; } diff --git a/src/core/comms-store.ts b/src/core/comms-store.ts index 3e4dee19..2ef7d6ac 100644 --- a/src/core/comms-store.ts +++ b/src/core/comms-store.ts @@ -5,7 +5,7 @@ * * Bridges depend on this interface, not on a specific implementation. * - * Deliberately excludes listener management, federation, and connection approval: those are transport concerns MeshStore alone can support -- FileStore has no network transport to manage listeners on, federate through, or approve inbound connections for. Widening this interface to cover them (as it once did, via always-throwing FileStore stubs) is what forced server.ts and the bridge controller to reach past CommsStore into the concrete MeshStore anyway; CommsTool, the one consumer that genuinely needs to expose these when a MeshStore backs it, takes them as an optional extension (see MeshOnlyFeatures in tool.ts) rather than the shared interface pretending every implementation supports them. + * Deliberately excludes listener management and connection approval: those are transport concerns MeshStore alone can support -- FileStore has no network transport to manage listeners on or approve inbound connections for. Widening this interface to cover them (as it once did, via always-throwing FileStore stubs) is what forced server.ts and the bridge controller to reach past CommsStore into the concrete MeshStore anyway; CommsTool, the one consumer that genuinely needs to expose these when a MeshStore backs it, takes them as an optional extension (see MeshOnlyFeatures in tool.ts) rather than the shared interface pretending every implementation supports them. */ import type { diff --git a/src/core/delivery-engine.ts b/src/core/delivery-engine.ts index 4a30b08a..1d624f44 100644 --- a/src/core/delivery-engine.ts +++ b/src/core/delivery-engine.ts @@ -45,7 +45,7 @@ export interface DeliveryEngineDeps { ((patch: MeshStatePatch) => void | Promise) | undefined; isShutDown: () => boolean; /** - * Sends one directed room-domain request to a single member, queuing it for retry when unreachable -- RoomProtocol's own method. Deferred: RoomProtocol doesn't exist yet when DeliveryEngine is constructed (construction order: discovery -\> deliveryEngine -\> ... -\> roomProtocol), so MeshStore wires this as `(...) => this.roomProtocol.sendRoomRequestToMember(...)`, a closure over `this` that only resolves `this.roomProtocol` when markRead actually calls it at runtime, well after the constructor has finished -- the same lazy-`this`-capture pattern the constructor already uses to wire FederationManager's own callbacks before `this.federation` exists. + * Sends one directed room-domain request to a single member, queuing it for retry when unreachable -- RoomProtocol's own method. Deferred: RoomProtocol doesn't exist yet when DeliveryEngine is constructed (construction order: discovery -\> deliveryEngine -\> ... -\> roomProtocol), so MeshStore wires this as `(...) => this.roomProtocol.sendRoomRequestToMember(...)`, a closure over `this` that only resolves `this.roomProtocol` when markRead actually calls it at runtime, well after the constructor has finished. */ sendRoomRequestToMember: ( memberId: string, @@ -110,8 +110,6 @@ export class DeliveryEngine { existing.owner = incoming.owner; existing.createdAt = incoming.createdAt; existing.description = incoming.description; - if (incoming.federated !== undefined) - existing.federated = incoming.federated; existing.memberJoins = DeliveryEngine.mergeMemberOps( existing.memberJoins, incoming.memberJoins, @@ -467,21 +465,6 @@ export class DeliveryEngine { } } - /** - * Delivers a room message to one specific member -- the local-fanout half of a federated room message (federation-bridge.ts's own onRoomMessage), which has no handleRoomSend manage-response of its own to carry delivery, since the message arrives over a federation link rather than a live room.send. Emits the "delivered" receipt back to the message's own sender the same way deliverLocallyAndBroadcast used to, then delivers to the member via deliverToMember (locally for this store's own agent, or a directed room.notify otherwise). - */ - async deliverRoomMessageToMember( - roomId: string, - memberId: string, - message: RoomMessage, - ): Promise { - await this.emitDeliveryStatus(message.id, memberId, "delivered", roomId); - await this.deliverToMember(memberId, roomId, { - type: "room_message", - message, - }); - } - async notifyRoomsOfStatus( agentId: string, status: AgentStatus, diff --git a/src/core/federation-bridge.ts b/src/core/federation-bridge.ts deleted file mode 100644 index 1a42d9f0..00000000 --- a/src/core/federation-bridge.ts +++ /dev/null @@ -1,151 +0,0 @@ -/** - * FederationBridge — implements FederationManager's own FedCallbacks contract as a real class rather than a closure bag, so the inbound-from-remote-mesh handling (a federated agent becoming visible/gone, a federated room's message/join/leave) and the two outbound sync queries (getVisibleAgents/getFederatedRoomMemberships) live together instead of scattered across mesh-store.ts. Split out to reduce mesh-store.ts under the repo's max-lines cap. - */ - -import type { FedCallbacks } from "./federation.js"; -import type { DeliveryEngine } from "./delivery-engine.js"; -import type { AgentIdentity, Room, RoomMessage } from "./types.js"; - -/** The state and DeliveryEngine operations FederationBridge needs from MeshStore -- the core agents/rooms/messages Maps are direct references into MeshStore's own fields, and deliveryEngine is the already-constructed instance (construction order: ... -\> deliveryEngine -\> federationBridge -\> federation), narrowed to only the membership/broadcast/delivery operations a federation callback ever needs. */ -export interface FederationBridgeDeps { - agents: Map; - rooms: Map; - messages: Map; - deliveryEngine: Pick< - DeliveryEngine, - | "bump" - | "recordMemberOp" - | "refreshMembership" - | "broadcastPatch" - | "deliverToRoom" - | "deliverRoomMessageToMember" - >; -} - -export class FederationBridge implements FedCallbacks { - constructor(private readonly deps: FederationBridgeDeps) {} - - /** Called when a remote agent becomes visible over a federation link -- stores it locally under a `fed:`-prefixed id to avoid collisions with local agents. */ - async onAgentVisible(agent: AgentIdentity): Promise { - const remoteId = `fed:${agent.id}@${agent.harness}`; - const remoteAgent: AgentIdentity = { - ...agent, - id: remoteId, - tags: [...agent.tags, "federated"], - }; - this.deps.agents.set(remoteId, remoteAgent); - await this.deps.deliveryEngine.broadcastPatch({ - type: "agent_upsert", - agent: remoteAgent, - }); - } - - /** Called when a remote agent goes offline/disappears -- the agentId comes from the remote mesh, so this finds the locally prefixed version. */ - async onAgentGone(agentId: string): Promise { - const prefix = `fed:${agentId}@`; - for (const [localId, agent] of this.deps.agents) { - if (localId.startsWith(prefix)) { - agent.status = "offline"; - this.deps.agents.set(localId, agent); - await this.deps.deliveryEngine.broadcastPatch({ - type: "agent_offline", - agentId: localId, - }); - break; - } - } - } - - /** Called when a message arrives for a federated room -- stores it locally and delivers to every local room member. A `fed:`-prefixed member is a shadow record for a remote participant with no addressable mesh device of its own; federation.ts's own link forwarding, not this local fan-out, is what reaches them. */ - async onRoomMessage(roomId: string, message: RoomMessage): Promise { - const room = this.deps.rooms.get(roomId); - if (room?.federated !== true) return; - - const arr = this.deps.messages.get(roomId) ?? []; - arr.push(message); - this.deps.messages.set(roomId, arr); - - for (const memberId of room.members) { - if (memberId.startsWith("fed:")) continue; - await this.deps.deliveryEngine.deliverRoomMessageToMember( - roomId, - memberId, - message, - ); - } - } - - /** Called when a remote agent joins a federated room -- creates a shadow `fed:`-prefixed member and notifies local members. */ - async onRoomJoin( - roomId: string, - agentId: string, - _agentName: string, - ): Promise { - const room = this.deps.rooms.get(roomId); - if (room?.federated !== true) return; - - const remoteId = `fed:${agentId}`; - - if (!room.members.includes(remoteId)) { - this.deps.deliveryEngine.bump(room); - this.deps.deliveryEngine.recordMemberOp(room, "member", "join", remoteId); - this.deps.deliveryEngine.refreshMembership(room); - this.deps.rooms.set(roomId, room); - await this.deps.deliveryEngine.broadcastPatch({ - type: "room_upsert", - room, - }); - } - - await this.deps.deliveryEngine.deliverToRoom( - roomId, - { type: "member_joined", room: roomId, agent: remoteId }, - remoteId, - ); - } - - /** Called when a remote agent leaves a federated room. */ - async onRoomLeave(roomId: string, agentId: string): Promise { - const room = this.deps.rooms.get(roomId); - if (room?.federated !== true) return; - - const remoteId = `fed:${agentId}`; - this.deps.deliveryEngine.bump(room); - this.deps.deliveryEngine.recordMemberOp(room, "member", "leave", remoteId); - this.deps.deliveryEngine.refreshMembership(room); - this.deps.rooms.set(roomId, room); - await this.deps.deliveryEngine.broadcastPatch({ - type: "room_upsert", - room, - }); - - await this.deps.deliveryEngine.deliverToRoom(roomId, { - type: "member_left", - room: roomId, - agent: remoteId, - }); - } - - /** Get all visible agents in the local mesh (for syncing to new links) -- excludes agents already federated in from elsewhere, to avoid re-broadcasting them back out. */ - getVisibleAgents(): AgentIdentity[] { - const result: AgentIdentity[] = []; - for (const agent of this.deps.agents.values()) { - if (agent.visibility === "visible" && !agent.id.startsWith("fed:")) { - result.push(agent); - } - } - return result; - } - - /** Get all federated rooms and their local member lists (for syncing to new links). */ - getFederatedRoomMemberships(): Map { - const result = new Map(); - for (const [roomId, room] of this.deps.rooms) { - if (room.federated === true) { - const localMembers = room.members.filter((m) => !m.startsWith("fed:")); - result.set(roomId, localMembers); - } - } - return result; - } -} diff --git a/src/core/federation.ts b/src/core/federation.ts deleted file mode 100644 index 556f331c..00000000 --- a/src/core/federation.ts +++ /dev/null @@ -1,674 +0,0 @@ -/** - * FederationManager — manages TLS links between mesh coordinators. - * - * Federation links are persistent TLS connections between coordinators on - * different machines. They forward agent presence, room memberships, and - * messages for rooms marked as federated. Non-federated rooms never leave - * the local mesh. - * - * Each link uses certificate pinning (the same trust model as TlsTransport). - * The wire protocol runs over the same newline-delimited JSON framing as the - * local mesh, but with `fed_*` method types. - */ - -import * as tls from "node:tls"; -import { encode, isMeshMessage, MessageBuffer } from "./wire-protocol.js"; -import type { MeshMessage } from "./wire-protocol.js"; -import type { AgentIdentity, RoomMessage } from "./types.js"; -import { nanoid } from "./nanoid.js"; -import { fingerprintDer, generateIdentity } from "./identity.js"; -import type { PeerIdentity } from "./identity.js"; - -// --------------------------------------------------------------------------- -// Types -// --------------------------------------------------------------------------- - -/** Represents an active federation link (outbound or inbound). */ -export interface FedLink { - /** Unique ID for this link (assigned locally). */ - id: string; - /** Remote mesh ID (received during handshake). */ - remoteMeshId: string; - /** Human-readable name for the remote mesh. */ - remoteName: string; - /** The TLS socket. */ - socket: tls.TLSSocket; - /** Whether the handshake has completed successfully. */ - ready: boolean; - /** Direction of the connection. */ - direction: "outbound" | "inbound"; -} - -/** Callbacks the FederationManager uses to interact with local mesh state. */ -export interface FedCallbacks { - /** Called when a remote agent becomes visible over a federation link. */ - onAgentVisible: (agent: AgentIdentity) => Promise; - /** Called when a remote agent goes offline/disappears. */ - onAgentGone: (agentId: string) => Promise; - /** Called when a message arrives for a federated room. */ - onRoomMessage: (roomId: string, message: RoomMessage) => Promise; - /** Called when a remote agent joins a federated room. */ - onRoomJoin: ( - roomId: string, - agentId: string, - agentName: string, - ) => Promise; - /** Called when a remote agent leaves a federated room. */ - onRoomLeave: (roomId: string, agentId: string) => Promise; - /** Get all visible agents in the local mesh (for syncing to new links). */ - getVisibleAgents: () => AgentIdentity[]; - /** Get all federated rooms and their member lists (for syncing to new links). */ - getFederatedRoomMemberships: () => Map; -} - -// --------------------------------------------------------------------------- -// Constants -// --------------------------------------------------------------------------- - -const FED_PING_INTERVAL_MS = 30_000; -const FED_PING_TIMEOUT_MS = 10_000; -const FED_VERSION = "1.0.0"; -/** Length (in bytes of entropy) of a locally-assigned federation link ID. */ -const FED_LINK_ID_LENGTH = 8; -/** How long to wait for the remote side to complete the fed_handshake/fed_ack exchange before giving up. */ -const FED_HANDSHAKE_TIMEOUT_MS = 5000; -/** How long to wait for the underlying TLS socket itself to connect before giving up. */ -const FED_CONNECT_TIMEOUT_MS = 5000; -/** How often {@link FederationManager.waitForHandshake} polls link readiness. */ -const FED_HANDSHAKE_POLL_INTERVAL_MS = 50; - -// --------------------------------------------------------------------------- -// FederationManager -// --------------------------------------------------------------------------- - -export class FederationManager { - private readonly links = new Map(); - private readonly identity: PeerIdentity; - private readonly meshId: string; - private readonly meshName: string; - private readonly callbacks: FedCallbacks; - private readonly pingTimers = new Map< - string, - ReturnType - >(); - private shutDown = false; - private readonly pendingPongs = new Map< - string, - ReturnType - >(); - /** - * Certificate fingerprints this instance will federate with, inbound or outbound. Empty by default — federation trusts nobody until an operator explicitly pins a remote mesh's fingerprint, the same no-CA, pin-the-key trust model ordinary peer connections already use. - */ - private readonly trustedFingerprints = new Set(); - private listener: tls.Server | undefined; - - constructor( - meshId: string, - meshName: string, - callbacks: Readonly, - ) { - this.meshId = meshId; - this.meshName = meshName; - this.callbacks = callbacks; - this.identity = generateIdentity(); - } - - /** The TLS identity used for federation connections. */ - get tlsIdentity(): PeerIdentity { - return this.identity; - } - - // ----------------------------------------------------------------------- - // Trust — which remote mesh fingerprints this instance will federate with - // ----------------------------------------------------------------------- - - /** Pin a remote mesh's certificate fingerprint as trusted for federation. */ - addTrustedFingerprint(fingerprint: string): void { - this.trustedFingerprints.add(fingerprint); - } - - /** Remove a previously pinned fingerprint. Existing links using it are not torn down. */ - removeTrustedFingerprint(fingerprint: string): void { - this.trustedFingerprints.delete(fingerprint); - } - - /** List currently trusted fingerprints. */ - listTrustedFingerprints(): string[] { - return [...this.trustedFingerprints]; - } - - /** - * Verify the certificate a connected TLS socket presented against the trusted-fingerprint allowlist. Returns the presented fingerprint when trusted, `undefined` (and destroys the socket) otherwise. - * - * This is the check that was missing entirely before: the socket was accepted with `rejectUnauthorized: false` (required, since these are self-signed certs with no CA) but nothing then verified *which* self-signed cert was presented, so any certificate was accepted as a valid federation peer. - */ - private verifyPeerOrDestroy(socket: tls.TLSSocket): string | undefined { - const cert = socket.getPeerCertificate(); - // Node's types declare every PeerCertificate field non-optional, but the documented runtime behaviour when the peer presents no certificate at all is an empty object — not null/undefined, and not a Buffer-typed `raw`. Detect that real shape rather than trusting the declared type. - if (Object.keys(cert).length === 0) { - socket.destroy(); - return undefined; - } - const fingerprint = fingerprintDer(cert.raw); - if (!this.trustedFingerprints.has(fingerprint)) { - socket.destroy(); - return undefined; - } - return fingerprint; - } - - // ----------------------------------------------------------------------- - // Outbound links - // ----------------------------------------------------------------------- - - /** - * Establish an outbound federation link to a remote coordinator. - * Returns the local link ID once the handshake completes. - */ - async connect(host: string, port: number, name?: string): Promise { - if (this.shutDown) throw new Error("FederationManager is shut down"); - - const linkId = nanoid(FED_LINK_ID_LENGTH); - - const socket = await this.tlsConnect(host, port); - if (this.verifyPeerOrDestroy(socket) === undefined) { - throw new Error( - `Federation connection to ${host}:${String(port)} rejected: ` + - "the presented certificate is not in the trusted-fingerprint allowlist. " + - "Call addTrustedFingerprint() with the remote mesh's fingerprint first.", - ); - } - const link: FedLink = { - id: linkId, - remoteMeshId: "", - remoteName: name ?? `${host}:${String(port)}`, - socket, - ready: false, - direction: "outbound", - }; - this.links.set(linkId, link); - - // Send handshake - const handshake: MeshMessage = { - method: "fed_handshake", - meshId: this.meshId, - name: this.meshName, - version: FED_VERSION, - }; - await this.writeToSocket(socket, handshake); - - // Wire up incoming message handling - this.wireSocket(linkId, socket); - - // Wait for fed_ack (with timeout) - await this.waitForHandshake(linkId, FED_HANDSHAKE_TIMEOUT_MS); - - return linkId; - } - - // ----------------------------------------------------------------------- - // Inbound links (server) - // ----------------------------------------------------------------------- - - /** - * Handle an incoming TLS connection that sent a fed_handshake. - * Called by the transport layer when a federation connection arrives. - */ - async handleInbound(socket: tls.TLSSocket): Promise { - if (this.shutDown) { - socket.destroy(); - throw new Error("FederationManager is shut down"); - } - - if (this.verifyPeerOrDestroy(socket) === undefined) { - throw new Error( - "Inbound federation connection rejected: the presented certificate " + - "is not in the trusted-fingerprint allowlist.", - ); - } - - const linkId = nanoid(FED_LINK_ID_LENGTH); - const link: FedLink = { - id: linkId, - remoteMeshId: "", - remoteName: "unknown", - socket, - ready: false, - direction: "inbound", - }; - this.links.set(linkId, link); - - this.wireSocket(linkId, socket); - - // Wait for the remote side's handshake - await this.waitForHandshake(linkId, FED_HANDSHAKE_TIMEOUT_MS); - - return linkId; - } - - // ----------------------------------------------------------------------- - // Link management - // ----------------------------------------------------------------------- - - /** Disconnect a specific federation link. */ - async disconnect(linkId: string): Promise { - const link = this.links.get(linkId); - if (!link) return Promise.resolve(); - - this.clearLinkTimers(linkId); - link.ready = false; - link.socket.unref(); - link.socket.destroy(); - this.links.delete(linkId); - - return Promise.resolve(); - } - - /** List all active federation links. */ - listLinks(): FedLink[] { - return [...this.links.values()].filter((l) => l.ready); - } - - // ----------------------------------------------------------------------- - // Broadcasting to federated links - // ----------------------------------------------------------------------- - - /** Broadcast an agent visibility event to all ready links. */ - async broadcastAgentVisible(agent: AgentIdentity): Promise { - const msg: MeshMessage = { method: "fed_agent_visible", agent }; - await this.broadcastToReady(msg); - } - - /** Broadcast an agent gone event to all ready links. */ - async broadcastAgentGone(agentId: string): Promise { - const msg: MeshMessage = { method: "fed_agent_gone", agentId }; - await this.broadcastToReady(msg); - } - - /** Forward a room message to all ready links (federated rooms only). */ - async forwardRoomMessage( - roomId: string, - message: RoomMessage, - ): Promise { - const msg: MeshMessage = { method: "fed_room_message", roomId, message }; - await this.broadcastToReady(msg); - } - - /** Broadcast a room join to all ready links. */ - async broadcastRoomJoin( - roomId: string, - agentId: string, - agentName: string, - ): Promise { - const msg: MeshMessage = { - method: "fed_room_join", - roomId, - agentId, - agentName, - }; - await this.broadcastToReady(msg); - } - - /** Broadcast a room leave to all ready links. */ - async broadcastRoomLeave(roomId: string, agentId: string): Promise { - const msg: MeshMessage = { method: "fed_room_leave", roomId, agentId }; - await this.broadcastToReady(msg); - } - - // ----------------------------------------------------------------------- - // Inbound listener — accepts federation links from remote coordinators - // ----------------------------------------------------------------------- - - /** - * Start listening for inbound federation connections. Every accepted connection is routed through `handleInbound()`, which enforces the trusted-fingerprint check before a link is ever created — nothing here bypasses that check. - * - * Previously nothing in the shipped product called `handleInbound()` at all: it existed only as a function the integration test invoked directly against a hand-rolled `tls.createServer`. This is that server, promoted to real code. - */ - async listen(host: string, port: number): Promise { - if (this.listener) { - throw new Error("FederationManager is already listening"); - } - return new Promise((resolve, reject) => { - const server = tls.createServer( - { - key: this.identity.privateKey, - cert: this.identity.certificate, - // Same as the outbound side: no CA, so we don't ask Node to verify the chain. requestCert is what makes the connecting peer's own certificate available to verifyPeerOrDestroy() inside handleInbound() — without it there is nothing to check. - rejectUnauthorized: false, - requestCert: true, - }, - (socket) => { - this.handleInbound(socket).catch(() => { - // Rejected (untrusted fingerprint, or shutting down) — the socket is already destroyed inside handleInbound/verifyPeerOrDestroy. - }); - }, - ); - - server.listen(port, host, () => { - this.listener = server; - resolve(); - }); - server.on("error", reject); - }); - } - - /** Stop accepting new inbound federation connections. Existing links are unaffected. */ - async stopListening(): Promise { - const server = this.listener; - if (!server) return Promise.resolve(); - this.listener = undefined; - return new Promise((resolve) => { - server.close(() => { - resolve(); - }); - }); - } - - // ----------------------------------------------------------------------- - // Shutdown - // ----------------------------------------------------------------------- - - async shutdown(): Promise { - this.shutDown = true; - await this.stopListening(); - for (const linkId of [...this.links.keys()]) { - await this.disconnect(linkId); - } - } - - // ----------------------------------------------------------------------- - // Internal — TLS connection - // ----------------------------------------------------------------------- - - private async tlsConnect(host: string, port: number): Promise { - return new Promise((resolve, reject) => { - const socket = tls.connect( - { - key: this.identity.privateKey, - cert: this.identity.certificate, - host, - port, - rejectUnauthorized: false, - requestCert: true, - }, - () => { - resolve(socket); - }, - ); - - const timer = setTimeout(() => { - socket.destroy(); - reject( - new Error(`Federation connection timeout to ${host}:${String(port)}`), - ); - }, FED_CONNECT_TIMEOUT_MS); - - socket.on("error", (err) => { - clearTimeout(timer); - reject(err); - }); - }); - } - - // ----------------------------------------------------------------------- - // Internal — socket wiring - // ----------------------------------------------------------------------- - - private wireSocket(linkId: string, socket: tls.TLSSocket): void { - const buffer = new MessageBuffer(); - - socket.on("data", (data) => { - const items = buffer.append(data.toString()); - for (const item of items) { - if (isMeshMessage(item)) { - void this.handleMessage(linkId, item); - } - } - }); - - socket.on("error", () => { - void this.disconnect(linkId); - }); - - socket.on("close", () => { - void this.disconnect(linkId); - }); - } - - // ----------------------------------------------------------------------- - // Internal — message handling - // ----------------------------------------------------------------------- - - private async handleMessage(linkId: string, msg: MeshMessage): Promise { - const link = this.links.get(linkId); - if (!link || this.shutDown) return; - - switch (msg.method) { - case "fed_handshake": { - // Inbound connection sending its handshake - link.remoteMeshId = msg.meshId; - link.remoteName = msg.name; - link.ready = true; - - // Respond with ack - const ack: MeshMessage = { - method: "fed_ack", - meshId: this.meshId, - name: this.meshName, - version: FED_VERSION, - }; - await this.writeToSocket(link.socket, ack); - - // Sync local state to the new link - await this.syncStateToLink(linkId); - - // Start ping for this link - this.startPing(linkId); - break; - } - case "fed_ack": { - link.remoteMeshId = msg.meshId; - link.remoteName = msg.name; - link.ready = true; - - // Sync local state to the new link - await this.syncStateToLink(linkId); - - // Start ping for this link - this.startPing(linkId); - break; - } - case "fed_agent_visible": { - await this.callbacks.onAgentVisible(msg.agent); - break; - } - case "fed_agent_gone": { - await this.callbacks.onAgentGone(msg.agentId); - break; - } - case "fed_room_message": { - await this.callbacks.onRoomMessage(msg.roomId, msg.message); - break; - } - case "fed_room_join": { - await this.callbacks.onRoomJoin(msg.roomId, msg.agentId, msg.agentName); - break; - } - case "fed_room_leave": { - await this.callbacks.onRoomLeave(msg.roomId, msg.agentId); - break; - } - case "fed_ping": { - const pong: MeshMessage = { method: "fed_pong" }; - await this.writeToSocket(link.socket, pong); - break; - } - case "fed_pong": { - const pending = this.pendingPongs.get(linkId); - if (pending) { - clearTimeout(pending); - this.pendingPongs.delete(linkId); - } - break; - } - case "state_sync": - case "state_update": - case "introduce": - case "connect_request": - case "peer_list": - case "peer_joined": - case "peer_left": - case "become_coordinator": { - // Local-mesh-only message kinds — never sent over a federation link; ignore rather than throw so an unexpected wire message can't tear down the link. - break; - } - } - } - - // ----------------------------------------------------------------------- - // Internal — state sync - // ----------------------------------------------------------------------- - - /** - * After handshake, push our visible agents and federated room memberships - * to the newly connected link. - */ - private async syncStateToLink(linkId: string): Promise { - const link = this.links.get(linkId); - if (link?.ready !== true) return; - - // Sync visible agents - const agents = this.callbacks.getVisibleAgents(); - for (const agent of agents) { - const msg: MeshMessage = { method: "fed_agent_visible", agent }; - await this.writeToSocket(link.socket, msg); - } - - // Sync federated room memberships - const rooms = this.callbacks.getFederatedRoomMemberships(); - for (const [roomId, memberIds] of rooms) { - for (const memberId of memberIds) { - const agent = this.callbacks - .getVisibleAgents() - .find((a) => a.id === memberId); - const agentName = agent?.name ?? memberId; - const msg: MeshMessage = { - method: "fed_room_join", - roomId, - agentId: memberId, - agentName, - }; - await this.writeToSocket(link.socket, msg); - } - } - } - - // ----------------------------------------------------------------------- - // Internal — handshake waiting - // ----------------------------------------------------------------------- - - private async waitForHandshake( - linkId: string, - timeoutMs: number, - ): Promise { - return new Promise((resolve, reject) => { - const startTime = Date.now(); - - const check = (): void => { - const link = this.links.get(linkId); - if (link?.ready === true) { - resolve(); - return; - } - if (Date.now() - startTime > timeoutMs) { - void this.disconnect(linkId); - reject(new Error(`Federation handshake timeout for link ${linkId}`)); - return; - } - setTimeout(check, FED_HANDSHAKE_POLL_INTERVAL_MS); - }; - - check(); - }); - } - - // ----------------------------------------------------------------------- - // Internal — ping/pong health monitoring - // ----------------------------------------------------------------------- - - private startPing(linkId: string): void { - if (this.pingTimers.has(linkId)) return; - - const timer = setInterval(() => { - void this.sendPing(linkId); - }, FED_PING_INTERVAL_MS); - - this.pingTimers.set(linkId, timer); - } - - private async sendPing(linkId: string): Promise { - const link = this.links.get(linkId); - if (link?.ready !== true) return; - - const msg: MeshMessage = { method: "fed_ping" }; - await this.writeToSocket(link.socket, msg); - - // Set pong timeout - const timer = setTimeout(() => { - // No pong received — disconnect the link - void this.disconnect(linkId); - }, FED_PING_TIMEOUT_MS); - - this.pendingPongs.set(linkId, timer); - } - - // ----------------------------------------------------------------------- - // Internal — helpers - // ----------------------------------------------------------------------- - - private async writeToSocket( - socket: tls.TLSSocket, - msg: MeshMessage, - ): Promise { - if (socket.destroyed) return; - const data = encode(msg); - await new Promise((resolve, reject) => { - socket.write(data, "utf-8", (err) => { - if (err) reject(err); - else resolve(); - }); - }); - } - - private async broadcastToReady(msg: MeshMessage): Promise { - const data = encode(msg); - const writes: Promise[] = []; - for (const [, link] of this.links) { - if (!link.ready || link.socket.destroyed) continue; - writes.push( - new Promise((resolve) => { - link.socket.write(data, "utf-8", (err) => { - if (err) { - void this.disconnect(link.id); - } - resolve(); - }); - }), - ); - } - await Promise.all(writes); - } - - private clearLinkTimers(linkId: string): void { - const pingTimer = this.pingTimers.get(linkId); - if (pingTimer !== undefined) { - clearInterval(pingTimer); - this.pingTimers.delete(linkId); - } - const pongTimer = this.pendingPongs.get(linkId); - if (pongTimer !== undefined) { - clearTimeout(pongTimer); - this.pendingPongs.delete(linkId); - } - } -} diff --git a/src/core/index.ts b/src/core/index.ts index 9a2328c8..708de2eb 100644 --- a/src/core/index.ts +++ b/src/core/index.ts @@ -41,5 +41,3 @@ export type { ListenerPolicy, ConnectionHandle, } from "./transport.js"; -export { FederationManager } from "./federation.js"; -export type { FedLink, FedCallbacks } from "./federation.js"; diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index dc157150..2fdb52c8 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -5,7 +5,7 @@ * * Transport is set via setTransport() (e.g. WireMeshTransport for encrypted connections) before init() or any other transport-using method is called -- there is no default, since every real bridge builds its own transport from this store's own events getter, which needs the store to already exist. * - * MeshStore itself is an orchestrator: it owns the shared state (the core agents/rooms/messages/dms/deliveryQueues Maps and a handful of smaller fields) and constructs the collaborators that implement almost every behaviour against direct references into that state -- DeliveryEngine, FederationBridge, RoomProtocol, RoomMessaging, RoomLifecycle, AgentRegistry, ConnectionApproval, StaleAgentChecker, and PeerLifecycle. Every public method below that isn't inherently a MeshStore-level concern (transport/identity wiring, init/shutdown lifecycle, the events getter, mesh visibility, listener management, federation-adapter passthroughs) is a thin delegating wrapper to whichever collaborator now owns the real implementation, kept here only because CommsStore/MeshOnlyFeatures and a handful of concrete-only call sites (tests, bridge-mesh.ts, the web server, etc.) reach these names directly on a MeshStore-typed value. + * MeshStore itself is an orchestrator: it owns the shared state (the core agents/rooms/messages/dms/deliveryQueues Maps and a handful of smaller fields) and constructs the collaborators that implement almost every behaviour against direct references into that state -- DeliveryEngine, RoomProtocol, RoomMessaging, RoomLifecycle, AgentRegistry, ConnectionApproval, StaleAgentChecker, and PeerLifecycle. Every public method below that isn't inherently a MeshStore-level concern (transport/identity wiring, init/shutdown lifecycle, the events getter, mesh visibility, listener management) is a thin delegating wrapper to whichever collaborator now owns the real implementation, kept here only because CommsStore/MeshOnlyFeatures and a handful of concrete-only call sites (tests, bridge-mesh.ts, the web server, etc.) reach these names directly on a MeshStore-typed value. */ import * as os from "node:os"; @@ -14,13 +14,9 @@ import { CommsError } from "./store.js"; import { DiscoveryManager } from "./discovery.js"; import { MdnsDiscoveryBackend } from "./discovery-mdns.js"; import { TailscaleDiscoveryBackend } from "./discovery-tailscale.js"; -import { FederationManager } from "./federation.js"; -import type { FedLink } from "./federation.js"; -import { getCertificateFingerprint } from "./identity.js"; import { COORDINATOR_HOST } from "./mesh-store-shared.js"; import type { MeshStoreIdentity } from "./mesh-store-shared.js"; import { DeliveryEngine } from "./delivery-engine.js"; -import { FederationBridge } from "./federation-bridge.js"; import { RoomProtocol } from "./room-protocol.js"; import { RoomMessaging } from "./room-messaging.js"; import { RoomLifecycle } from "./room-lifecycle.js"; @@ -94,10 +90,8 @@ export class MeshStore implements CommsStore { private initialised = false; discovery: DiscoveryManager; - federation: FederationManager; private readonly deliveryEngine: DeliveryEngine; - private readonly federationBridge: FederationBridge; private readonly roomProtocol: RoomProtocol; private readonly roomMessaging: RoomMessaging; private readonly roomLifecycle: RoomLifecycle; @@ -197,7 +191,7 @@ export class MeshStore implements CommsStore { getOnDelivery: () => this.onDelivery, getOnPatch: () => this.onPatch, isShutDown: () => this.isShutDown, - // RoomProtocol doesn't exist yet at this point in the constructor -- this closure resolves `this.roomProtocol` lazily, only once markRead actually calls it at runtime, well after the constructor has finished. Mirrors the lazy-`this`-capture pattern FederationManager's own callbacks use below. + // RoomProtocol doesn't exist yet at this point in the constructor -- this closure resolves `this.roomProtocol` lazily, only once markRead actually calls it at runtime, well after the constructor has finished. sendRoomRequestToMember: async (memberId, roomPath, token, params) => this.roomProtocol.sendRoomRequestToMember( memberId, @@ -207,20 +201,6 @@ export class MeshStore implements CommsStore { ), }); - this.federationBridge = new FederationBridge({ - agents: this.agents, - rooms: this.rooms, - messages: this.messages, - deliveryEngine: this.deliveryEngine, - }); - - // Federation manager — coordinator-to-coordinator links - this.federation = new FederationManager( - this.peerId, // mesh ID is the coordinator's peer ID - `mesh-${this.peerId}`, - this.federationBridge, - ); - this.roomProtocol = new RoomProtocol({ rooms: this.rooms, messages: this.messages, @@ -243,7 +223,6 @@ export class MeshStore implements CommsStore { agents: this.agents, requireIdentity: () => this.requireIdentity(), roomProtocol: this.roomProtocol, - federation: this.federation, }); this.roomLifecycle = new RoomLifecycle({ @@ -255,7 +234,6 @@ export class MeshStore implements CommsStore { requireIdentity: () => this.requireIdentity(), requireTransport: () => this.requireTransport(), deliveryEngine: this.deliveryEngine, - federation: this.federation, }); this.agentRegistry = new AgentRegistry({ @@ -265,7 +243,6 @@ export class MeshStore implements CommsStore { getPeerId: () => this.peerId, requireTransport: () => this.requireTransport(), deliveryEngine: this.deliveryEngine, - federation: this.federation, }); this.connectionApproval = new ConnectionApproval({ @@ -505,7 +482,6 @@ export class MeshStore implements CommsStore { type: RoomType; owner: string; description: string; - federated?: boolean; }>, ): Promise { return this.roomLifecycle.createRoom(opts); @@ -748,48 +724,6 @@ export class MeshStore implements CommsStore { return result; } - // ----------------------------------------------------------------------- - // Federation (coordinator-to-coordinator) - // ----------------------------------------------------------------------- - - async fedConnect(host: string, port: number, name?: string): Promise { - return this.federation.connect(host, port, name); - } - - async fedDisconnect(linkId: string): Promise { - await this.federation.disconnect(linkId); - } - - fedLinks(): FedLink[] { - return this.federation.listLinks(); - } - - getFederationFingerprint(): string { - return getCertificateFingerprint(this.federation.tlsIdentity.certificate); - } - - async fedTrust(fingerprint: string): Promise { - this.federation.addTrustedFingerprint(fingerprint); - return Promise.resolve(); - } - - async fedUntrust(fingerprint: string): Promise { - this.federation.removeTrustedFingerprint(fingerprint); - return Promise.resolve(); - } - - fedTrustedFingerprints(): string[] { - return this.federation.listTrustedFingerprints(); - } - - async fedListen(host: string, port: number): Promise { - return this.federation.listen(host, port); - } - - async fedStopListening(): Promise { - return this.federation.stopListening(); - } - // ----------------------------------------------------------------------- // Shutdown // ----------------------------------------------------------------------- @@ -812,7 +746,6 @@ export class MeshStore implements CommsStore { } this.staleAgentChecker.stop(); - await this.federation.shutdown(); await this.requireTransport().shutdown(); } } diff --git a/src/core/room-lifecycle.ts b/src/core/room-lifecycle.ts index f9bb7005..71a0889c 100644 --- a/src/core/room-lifecycle.ts +++ b/src/core/room-lifecycle.ts @@ -39,7 +39,6 @@ import { roomStateExtension, } from "./room-wire-extensions.js"; import type { DeliveryEngine } from "./delivery-engine.js"; -import type { FederationManager } from "./federation.js"; import type { MeshTransport } from "./transport.js"; import type { HostedRoomAdvert } from "./wire-mesh-transport.js"; import type { @@ -65,7 +64,7 @@ function isHostedRoomAdvert(value: unknown): value is HostedRoomAdvert { return true; } -/** The state and collaborators RoomLifecycle needs from MeshStore. rooms/messages/agents/dmRequestsInitiatedByMe are direct references into MeshStore's own fields (dmRequestsInitiatedByMe shared with RoomProtocol, which reads what requestDmAccess writes here); deliveryEngine and federation are the already-constructed instances, narrowed to what room CRUD ever needs. */ +/** The state and collaborators RoomLifecycle needs from MeshStore. rooms/messages/agents/dmRequestsInitiatedByMe are direct references into MeshStore's own fields (dmRequestsInitiatedByMe shared with RoomProtocol, which reads what requestDmAccess writes here); deliveryEngine is the already-constructed instance, narrowed to what room CRUD ever needs. */ export interface RoomLifecycleDeps { rooms: Map; messages: Map; @@ -83,10 +82,6 @@ export interface RoomLifecycleDeps { | "deliverToRoom" | "deliverToMember" >; - federation: Pick< - FederationManager, - "broadcastRoomJoin" | "broadcastRoomLeave" - >; } export class RoomLifecycle { @@ -124,7 +119,6 @@ export class RoomLifecycle { type: RoomType; owner: string; description: string; - federated?: boolean; }>, ): Promise { // slugRoomName sanitises an arbitrary caller-supplied name (e.g. from a live create_room tool call, not just an internal cwd basename) into the room-path grammar's [A-Za-z0-9_-]+ charset -- createRoom is the one choke point every room creation goes through, so this is the right place to do it rather than trusting every caller to have pre-slugged, the way the old bare-name id never required at all. @@ -151,7 +145,6 @@ export class RoomLifecycle { memberLeaves: {}, invitedJoins: {}, invitedLeaves: {}, - federated: opts.federated ?? false, }; this.deps.rooms.set(id, room); @@ -275,7 +268,6 @@ export class RoomLifecycle { memberLeaves: {}, invitedJoins: {}, invitedLeaves: {}, - federated: false, }; this.deps.rooms.set(roomPath, room); this.deps.messages.set(roomPath, []); @@ -340,7 +332,6 @@ export class RoomLifecycle { memberLeaves: {}, invitedJoins: existing?.invitedJoins ?? {}, invitedLeaves: existing?.invitedLeaves ?? {}, - federated: existing?.federated ?? false, }; this.deps.rooms.set(roomPath, room); return room; @@ -456,12 +447,6 @@ export class RoomLifecycle { agentId, ); - // Notify federated links if the room is federated - if (room.federated === true) { - const agentName = agent?.name ?? agentId; - await this.deps.federation.broadcastRoomJoin(roomId, agentId, agentName); - } - return room; } @@ -505,11 +490,6 @@ export class RoomLifecycle { agent: agentId, }); - // Notify federated links if the room is federated - if (room.federated === true) { - await this.deps.federation.broadcastRoomLeave(roomId, agentId); - } - if (room.members.length === 0 && room.owner === agentId) { await this.destroyRoom(roomId, agentId); } diff --git a/src/core/room-messaging.ts b/src/core/room-messaging.ts index 493185c6..f2f6dff7 100644 --- a/src/core/room-messaging.ts +++ b/src/core/room-messaging.ts @@ -10,7 +10,6 @@ import { recordRoomSendNotice } from "./room-notice-log.js"; import { CommsError } from "./store.js"; import type { MeshStoreIdentity } from "./mesh-store-shared.js"; import type { RoomProtocol } from "./room-protocol.js"; -import type { FederationManager } from "./federation.js"; import type { AgentIdentity, DmMessage, @@ -19,7 +18,7 @@ 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 and federation are the already-constructed instances (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/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. */ export interface RoomMessagingDeps { rooms: Map; messages: Map; @@ -27,7 +26,6 @@ export interface RoomMessagingDeps { agents: Map; requireIdentity: () => MeshStoreIdentity; roomProtocol: Pick; - federation: Pick; } export class RoomMessaging { @@ -87,11 +85,6 @@ export class RoomMessaging { arr.push(message); this.deps.messages.set(roomId, arr); - // Forward to federated links if the room is federated - if (room.federated === true) { - await this.deps.federation.forwardRoomMessage(roomId, message); - } - const params: Record = { verb: "room.send", "message-id": messageId, diff --git a/src/core/room-router.ts b/src/core/room-router.ts index e0919df1..c98f07bb 100644 --- a/src/core/room-router.ts +++ b/src/core/room-router.ts @@ -58,19 +58,10 @@ function routeLegacyMessage( // Only ever reaches here if a session was somehow promoted without going through WireMeshTransport's own consumeQuarantined handling of it -- can't happen given every requiresApproval accept path routes through consumeQuarantined first, kept here only so an unrecognised-in-context method fails closed rather than falling to the onMessage case below. return; } - // Every other wire method -- mesh-state gossip (state_sync/state_update/peer_left) and the coordinator-to-coordinator federation methods -- has no dedicated TransportEvents callback and falls to the generic onMessage handler, exactly as the old unconditional default case did. + // Every other wire method -- mesh-state gossip (state_sync/state_update/peer_left) -- has no dedicated TransportEvents callback and falls to the generic onMessage handler, exactly as the old unconditional default case did. case "state_sync": case "state_update": - case "peer_left": - case "fed_handshake": - case "fed_ack": - case "fed_agent_visible": - case "fed_agent_gone": - case "fed_room_message": - case "fed_room_join": - case "fed_room_leave": - case "fed_ping": - case "fed_pong": { + case "peer_left": { events.onMessage(handle, message); return; } diff --git a/src/core/tool.ts b/src/core/tool.ts index a5ca7f24..71993a9a 100644 --- a/src/core/tool.ts +++ b/src/core/tool.ts @@ -19,7 +19,6 @@ import type { import type { ListenerInfo } from "./transport.js"; import type { CommsStore } from "./comms-store.js"; import type { DiscoveryManager } from "./discovery.js"; -import type { FedLink } from "./federation.js"; import { CommsError } from "./store.js"; /** Table column widths for the plain-text listing helpers below, chosen to line up with the existing aligned output. */ @@ -58,22 +57,13 @@ export interface CommsResult { } /** - * Listener management, federation, and connection approval: transport concerns CommsStore deliberately excludes (see comms-store.ts's own header) since only MeshStore, never FileStore, can support them. Every method here is optional for exactly that reason -- a CommsTool backed by a FileStore simply doesn't have them, and each call site below reports that as an ordinary CommsResult error rather than assuming they exist. + * Listener management and connection approval: transport concerns CommsStore deliberately excludes (see comms-store.ts's own header) since only MeshStore, never FileStore, can support them. Every method here is optional for exactly that reason -- a CommsTool backed by a FileStore simply doesn't have them, and each call site below reports that as an ordinary CommsResult error rather than assuming they exist. */ export interface MeshOnlyFeatures { addListener?: (host: string, port: number, policy: string) => Promise; removeListener?: (id: string) => Promise; listListeners?: () => ListenerInfo[]; getNetworkInterfaces?: () => NetworkInterface[]; - fedConnect?: (host: string, port: number, name?: string) => Promise; - fedDisconnect?: (linkId: string) => Promise; - fedLinks?: () => FedLink[]; - getFederationFingerprint?: () => string; - fedTrust?: (fingerprint: string) => Promise; - fedUntrust?: (fingerprint: string) => Promise; - fedTrustedFingerprints?: () => string[]; - fedListen?: (host: string, port: number) => Promise; - fedStopListening?: () => Promise; acceptConnection?: (connectionId: string) => Promise; rejectConnection?: (connectionId: string, reason: string) => Promise; listPendingConnections?: () => { @@ -103,7 +93,7 @@ function notMeshBacked(action: string): CommsResult { }; } -/** Runs a mesh/federation store call that may throw, converting a thrown error into a CommsResult instead of repeating the same try/catch at every call site. `action` performs the call and returns the success message directly. */ +/** Runs a mesh store call that may throw, converting a thrown error into a CommsResult instead of repeating the same try/catch at every call site. `action` performs the call and returns the success message directly. */ async function tryMeshAction( verb: string, action: () => Promise, @@ -204,24 +194,6 @@ export class CommsTool { return await this.meshSetVisibility(action); case "mesh_get_visibility": return this.meshGetVisibility(action); - case "mesh_fed_connect": - return await this.meshFedConnect(ctx, action); - case "mesh_fed_disconnect": - return await this.meshFedDisconnect(ctx, action); - case "mesh_fed_links": - return this.meshFedLinks(ctx); - case "mesh_fed_fingerprint": - return this.meshFedFingerprint(ctx); - case "mesh_fed_trust": - return await this.meshFedTrust(ctx, action); - case "mesh_fed_untrust": - return await this.meshFedUntrust(ctx, action); - case "mesh_fed_trusted": - return this.meshFedTrusted(ctx); - case "mesh_fed_listen": - return await this.meshFedListen(ctx, action); - case "mesh_fed_stop_listening": - return await this.meshFedStopListening(ctx); default: return { content: `Unknown action: ${JSON.stringify(action).slice(0, UNKNOWN_ACTION_PREVIEW_LENGTH)}`, @@ -623,18 +595,6 @@ export class CommsTool { }; } - private async meshFedConnect( - _ctx: Readonly, - action: CommsAction & { action: "mesh_fed_connect" }, - ): Promise { - if (!this.store.fedConnect) return notMeshBacked("mesh_fed_connect"); - const fedConnect = this.store.fedConnect.bind(this.store); - return tryMeshAction("establish federation link", async () => { - const linkId = await fedConnect(action.host, action.port, action.name); - return `Federation link established: ${linkId} to ${action.host}:${String(action.port)}`; - }); - } - private async meshConnect( _ctx: Readonly, action: CommsAction & { action: "mesh_connect" }, @@ -647,18 +607,6 @@ export class CommsTool { }); } - private async meshFedDisconnect( - _ctx: Readonly, - action: CommsAction & { action: "mesh_fed_disconnect" }, - ): Promise { - if (!this.store.fedDisconnect) return notMeshBacked("mesh_fed_disconnect"); - const fedDisconnect = this.store.fedDisconnect.bind(this.store); - return tryMeshAction("close federation link", async () => { - await fedDisconnect(action.linkId); - return `Federation link ${action.linkId} closed.`; - }); - } - private async meshAccept( _ctx: Readonly, action: CommsAction & { action: "mesh_accept" }, @@ -671,92 +619,6 @@ export class CommsTool { }); } - private meshFedLinks(_ctx: Readonly): CommsResult { - if (!this.store.fedLinks) return notMeshBacked("mesh_fed_links"); - const links = this.store.fedLinks(); - if (links.length === 0) - return { content: "No federation links.", isError: false }; - - const lines = links.map((l) => { - const dir = l.direction === "outbound" ? "→" : "←"; - return `${l.id} ${dir} ${l.remoteName} (${l.remoteMeshId})`; - }); - return { - content: `Federation links:\n${lines.map((l) => ` ${l}`).join("\n")}`, - isError: false, - }; - } - - private meshFedFingerprint(_ctx: Readonly): CommsResult { - if (!this.store.getFederationFingerprint) - return notMeshBacked("mesh_fed_fingerprint"); - const fingerprint = this.store.getFederationFingerprint(); - return { - content: `This mesh's federation fingerprint: ${fingerprint}\nHand this to the operator on the other side so they can run mesh_fed_trust with it — and do the same in reverse before either side connects.`, - isError: false, - }; - } - - private async meshFedTrust( - _ctx: Readonly, - action: CommsAction & { action: "mesh_fed_trust" }, - ): Promise { - if (!this.store.fedTrust) return notMeshBacked("mesh_fed_trust"); - const fedTrust = this.store.fedTrust.bind(this.store); - return tryMeshAction("trust fingerprint", async () => { - await fedTrust(action.fingerprint); - return `Trusted federation fingerprint: ${action.fingerprint}`; - }); - } - - private async meshFedUntrust( - _ctx: Readonly, - action: CommsAction & { action: "mesh_fed_untrust" }, - ): Promise { - if (!this.store.fedUntrust) return notMeshBacked("mesh_fed_untrust"); - const fedUntrust = this.store.fedUntrust.bind(this.store); - return tryMeshAction("untrust fingerprint", async () => { - await fedUntrust(action.fingerprint); - return `Untrusted federation fingerprint: ${action.fingerprint}`; - }); - } - - private meshFedTrusted(_ctx: Readonly): CommsResult { - if (!this.store.fedTrustedFingerprints) - return notMeshBacked("mesh_fed_trusted"); - const fingerprints = this.store.fedTrustedFingerprints(); - if (fingerprints.length === 0) - return { content: "No trusted federation fingerprints.", isError: false }; - return { - content: `Trusted federation fingerprints:\n${fingerprints.map((f) => ` ${f}`).join("\n")}`, - isError: false, - }; - } - - private async meshFedListen( - _ctx: Readonly, - action: CommsAction & { action: "mesh_fed_listen" }, - ): Promise { - if (!this.store.fedListen) return notMeshBacked("mesh_fed_listen"); - const fedListen = this.store.fedListen.bind(this.store); - return tryMeshAction("start federation listener", async () => { - await fedListen(action.host, action.port); - return `Listening for inbound federation links on ${action.host}:${String(action.port)}. Only connections presenting a trusted fingerprint (mesh_fed_trust) will be accepted.`; - }); - } - - private async meshFedStopListening( - _ctx: Readonly, - ): Promise { - if (!this.store.fedStopListening) - return notMeshBacked("mesh_fed_stop_listening"); - const fedStopListening = this.store.fedStopListening.bind(this.store); - return tryMeshAction("stop federation listener", async () => { - await fedStopListening(); - return "Stopped accepting inbound federation connections."; - }); - } - private async meshReject( _ctx: Readonly, action: CommsAction & { action: "mesh_reject" }, diff --git a/src/core/types.ts b/src/core/types.ts index 724dd7d3..be176812 100644 --- a/src/core/types.ts +++ b/src/core/types.ts @@ -135,7 +135,6 @@ export const RoomSchema = defineSchema( memberLeaves: z.record(z.string(), z.number()), invitedJoins: z.record(z.string(), z.number()), invitedLeaves: z.record(z.string(), z.number()), - federated: z.boolean().optional(), }), ); export type Room = z.infer; @@ -406,34 +405,6 @@ export const CommsActionSchema = defineSchema( adapter: z.string().optional(), }), z.object({ action: z.literal("mesh_get_visibility") }), - // Federation actions (coordinator-to-coordinator) - z.object({ - action: z.literal("mesh_fed_connect"), - host: z.string(), - port: z.number(), - name: z.string().optional(), - }), - z.object({ - action: z.literal("mesh_fed_disconnect"), - linkId: z.string(), - }), - z.object({ action: z.literal("mesh_fed_links") }), - z.object({ action: z.literal("mesh_fed_fingerprint") }), - z.object({ - action: z.literal("mesh_fed_trust"), - fingerprint: z.string(), - }), - z.object({ - action: z.literal("mesh_fed_untrust"), - fingerprint: z.string(), - }), - z.object({ action: z.literal("mesh_fed_trusted") }), - z.object({ - action: z.literal("mesh_fed_listen"), - host: z.string(), - port: z.number(), - }), - z.object({ action: z.literal("mesh_fed_stop_listening") }), ]), ); export type CommsAction = z.infer; diff --git a/src/core/wire-protocol.ts b/src/core/wire-protocol.ts index 754db025..056f816d 100644 --- a/src/core/wire-protocol.ts +++ b/src/core/wire-protocol.ts @@ -68,22 +68,7 @@ export type MeshMessage = | { method: "peer_list"; peers: PeerInfo[] } | { method: "peer_joined"; peer: PeerInfo } | { method: "peer_left"; peerId: string } - | { method: "become_coordinator"; peerList: PeerInfo[] } - // Federation wire messages (coordinator-to-coordinator only) - | { method: "fed_handshake"; meshId: string; name: string; version: string } - | { method: "fed_ack"; meshId: string; name: string; version: string } - | { method: "fed_agent_visible"; agent: AgentIdentity } - | { method: "fed_agent_gone"; agentId: string } - | { method: "fed_room_message"; roomId: string; message: RoomMessage } - | { - method: "fed_room_join"; - roomId: string; - agentId: string; - agentName: string; - } - | { method: "fed_room_leave"; roomId: string; agentId: string } - | { method: "fed_ping" } - | { method: "fed_pong" }; + | { method: "become_coordinator"; peerList: PeerInfo[] }; // --------------------------------------------------------------------------- // Framing — newline-delimited JSON diff --git a/src/test/agent-registry.test.ts b/src/test/agent-registry.test.ts index 0823f41d..cdd58c17 100644 --- a/src/test/agent-registry.test.ts +++ b/src/test/agent-registry.test.ts @@ -41,8 +41,6 @@ interface Harness { broadcastPatch: ReturnType; notifyRoomsOfStatus: ReturnType; notifyRoomsOfNameChange: ReturnType; - broadcastAgentVisible: ReturnType; - broadcastAgentGone: ReturnType; } function makeHarness(peerId = OWNER_ID): Harness { @@ -50,8 +48,6 @@ function makeHarness(peerId = OWNER_ID): Harness { const broadcastPatch = vi.fn().mockResolvedValue(undefined); const notifyRoomsOfStatus = vi.fn().mockResolvedValue(undefined); const notifyRoomsOfNameChange = vi.fn().mockResolvedValue(undefined); - const broadcastAgentVisible = vi.fn().mockResolvedValue(undefined); - const broadcastAgentGone = vi.fn().mockResolvedValue(undefined); const deps: AgentRegistryDeps = { agents: new Map(), identityCache: new Map(), @@ -67,7 +63,6 @@ function makeHarness(peerId = OWNER_ID): Harness { notifyRoomsOfStatus, notifyRoomsOfNameChange, }, - federation: { broadcastAgentVisible, broadcastAgentGone }, }; return { deps, @@ -76,8 +71,6 @@ function makeHarness(peerId = OWNER_ID): Harness { broadcastPatch, notifyRoomsOfStatus, notifyRoomsOfNameChange, - broadcastAgentVisible, - broadcastAgentGone, }; } @@ -162,41 +155,6 @@ describe("AgentRegistry — registerAgent", () => { expect(second.pid).toBe(2); expect(second.tags).toEqual(["updated"]); }); - - it("broadcasts presence to federated links only for a visible agent, not hidden or ghost", async () => { - const visible = makeHarness(); - await visible.registry.registerAgent({ - name: "v", - harness: "pi", - cwd: "/tmp/v", - pid: 1, - visibility: "visible", - tags: [], - }); - expect(visible.broadcastAgentVisible).toHaveBeenCalledTimes(1); - - const hidden = makeHarness(); - await hidden.registry.registerAgent({ - name: "h", - harness: "pi", - cwd: "/tmp/h", - pid: 1, - visibility: "hidden", - tags: [], - }); - expect(hidden.broadcastAgentVisible).not.toHaveBeenCalled(); - - const ghost = makeHarness(); - await ghost.registry.registerAgent({ - name: "g", - harness: "pi", - cwd: "/tmp/g", - pid: 1, - visibility: "ghost", - tags: [], - }); - expect(ghost.broadcastAgentVisible).not.toHaveBeenCalled(); - }); }); describe("AgentRegistry — getAgent", () => { @@ -404,7 +362,7 @@ describe("AgentRegistry — setAgentOffline", () => { expect(h.notifyRoomsOfStatus).not.toHaveBeenCalled(); }); - it("for the owning peer's own agent, sets status offline, bumps, persists, and broadcasts to rooms/mesh/federation", async () => { + it("for the owning peer's own agent, sets status offline, bumps, persists, and broadcasts to rooms/mesh", async () => { const h = makeHarness(OWNER_ID); h.deps.agents.set(OWNER_ID, agent({ status: "active" })); @@ -417,7 +375,6 @@ describe("AgentRegistry — setAgentOffline", () => { type: "agent_offline", agentId: OWNER_ID, }); - expect(h.broadcastAgentGone).toHaveBeenCalledWith(OWNER_ID); }); it("for a non-owning peer's agent, updates local state but does not broadcast anything", async () => { @@ -430,6 +387,5 @@ describe("AgentRegistry — setAgentOffline", () => { expect(h.bump).toHaveBeenCalledTimes(1); expect(h.notifyRoomsOfStatus).not.toHaveBeenCalled(); expect(h.broadcastPatch).not.toHaveBeenCalled(); - expect(h.broadcastAgentGone).not.toHaveBeenCalled(); }); }); diff --git a/src/test/delivery-engine-directed-notify.test.ts b/src/test/delivery-engine-directed-notify.test.ts index 36ed6c3c..e39cd6dd 100644 --- a/src/test/delivery-engine-directed-notify.test.ts +++ b/src/test/delivery-engine-directed-notify.test.ts @@ -403,61 +403,3 @@ describe("DeliveryEngine — emitDeliveryStatus directed remote-sender delivery" expect(h.sendRoomRequestToMember).not.toHaveBeenCalled(); }); }); - -describe("DeliveryEngine — deliverRoomMessageToMember", () => { - it("sends a directed room.notify to a remote member and emits a delivered receipt back to the sender", async () => { - const h = makeHarness(); - const target = roomMessage({ - id: messageId(), - from: THIRD_ID, - room: "room-1", - }); - h.deps.messages.set("room-1", [target]); - vi.mocked(loadRoomTokens).mockReturnValue({ "room-1": FAKE_TOKEN }); - - await h.engine.deliverRoomMessageToMember("room-1", OTHER_ID, target); - - expect(h.sendRoomRequestToMember).toHaveBeenCalledWith( - OTHER_ID, - "room-1", - FAKE_TOKEN, - { verb: "room.notify", event: { type: "room_message", message: target } }, - ); - expect(h.sendRoomRequestToMember).toHaveBeenCalledWith( - THIRD_ID, - "room-1", - FAKE_TOKEN, - { - verb: "room.notify", - event: { - type: "delivery_status", - messageId: target.id, - agent: OTHER_ID, - status: "delivered", - room: "room-1", - }, - }, - ); - expect(h.transport.broadcast).not.toHaveBeenCalled(); - }); - - it("fires local delivery directly when the member is this store's own peer", async () => { - const h = makeHarness(); - const target = roomMessage({ - id: messageId(), - from: THIRD_ID, - room: "room-1", - }); - h.deps.messages.set("room-1", [target]); - vi.mocked(loadRoomTokens).mockReturnValue({}); - const onDelivery = vi.fn(); - h.setOnDelivery(onDelivery); - - await h.engine.deliverRoomMessageToMember("room-1", PEER_ID, target); - - expect(onDelivery).toHaveBeenCalledWith( - PEER_ID, - expect.objectContaining({ type: "room_message", message: target }), - ); - }); -}); diff --git a/src/test/delivery-engine.test.ts b/src/test/delivery-engine.test.ts index 7fe5bd38..d1e99331 100644 --- a/src/test/delivery-engine.test.ts +++ b/src/test/delivery-engine.test.ts @@ -325,25 +325,6 @@ describe("DeliveryEngine — mergeRoom via applyPatch(room_upsert)", () => { expect(h.deps.rooms.get("room-1")?.name).toBe("new-name"); }); - it("retains the existing federated flag when the incoming room leaves it undefined", async () => { - const h = makeHarness(); - h.deps.rooms.set("room-1", room({ version: 1, federated: true })); - const incoming = room({ version: 2 }); - delete incoming.federated; - await h.engine.applyPatch({ type: "room_upsert", room: incoming }); - expect(h.deps.rooms.get("room-1")?.federated).toBe(true); - }); - - it("overwrites the federated flag when the incoming room states it explicitly, including false", async () => { - const h = makeHarness(); - h.deps.rooms.set("room-1", room({ version: 1, federated: true })); - await h.engine.applyPatch({ - type: "room_upsert", - room: room({ version: 2, federated: false }), - }); - expect(h.deps.rooms.get("room-1")?.federated).toBe(false); - }); - it("merges concurrent memberJoins by keeping the higher revision per agent, never overwriting with a lower one", async () => { const h = makeHarness(); h.deps.rooms.set( @@ -707,7 +688,7 @@ describe("DeliveryEngine — applyPatch(agent_upsert)", () => { }); }); -// The `this.deps.agents.set(patch.agentId, agent)` call at the end of this branch has one provable equivalent mutant Stryker still raises: removing it entirely. `agent` here is fetched via `this.deps.agents.get(patch.agentId)`, the exact same object reference already stored in the Map, and `agent.status = "offline"` mutates that object in place -- so re-setting the map entry to the identical reference it already holds is a genuine no-op, the same Map.set-same-reference pattern already documented elsewhere in this codebase (agent-registry.ts's setAgentOffline, federation-bridge.ts's onAgentGone/onRoomLeave). The test below already proves the real, observable effect (status flips to "offline"). +// The `this.deps.agents.set(patch.agentId, agent)` call at the end of this branch has one provable equivalent mutant Stryker still raises: removing it entirely. `agent` here is fetched via `this.deps.agents.get(patch.agentId)`, the exact same object reference already stored in the Map, and `agent.status = "offline"` mutates that object in place -- so re-setting the map entry to the identical reference it already holds is a genuine no-op, the same Map.set-same-reference pattern already documented elsewhere in this codebase (agent-registry.ts's setAgentOffline). The test below already proves the real, observable effect (status flips to "offline"). describe("DeliveryEngine — applyPatch(agent_offline)", () => { it("marks an existing agent as offline", async () => { const h = makeHarness(); diff --git a/src/test/federation-bridge.test.ts b/src/test/federation-bridge.test.ts deleted file mode 100644 index cfa6d8cf..00000000 --- a/src/test/federation-bridge.test.ts +++ /dev/null @@ -1,396 +0,0 @@ -/** - * Direct, DI-based unit tests for FederationBridge -- it was previously exercised only indirectly through end-to-end federation.integration.test.ts scenarios, leaving many individual branches, the fed:-prefix startsWith-vs-endsWith distinction, and getVisibleAgents/getFederatedRoomMemberships' filtering unobserved. FederationBridgeDeps is a narrow, injectable surface built exactly for this: a fake deps object with vi.fn() collaborators lets every branch be asserted on directly. - * - * Three mutants Stryker raises are true equivalents, not gaps -- documented here rather than chased with a contrived test, matching stale-agent-checker.test.ts's and agent-registry.test.ts's own precedent for the identical pattern: `onAgentGone`'s `this.deps.agents.set(localId, agent)`, and `onRoomJoin`/`onRoomLeave`'s two `this.deps.rooms.set(roomId, room)` calls, each re-set the same key to the exact same object reference the map's own `.get()` already returned. `bump`/`recordMemberOp`/`refreshMembership` all mutate that object in place (bump does `entity.version += 1`; recordMemberOp writes into `room.memberJoins`/`memberLeaves`; refreshMembership reassigns `room.members`/`invited`), so the map already holds the fully up-to-date object before the redundant `.set()` call -- no test can observe removing it. - */ -import { describe, expect, it, vi } from "vitest"; -import { - FederationBridge, - type FederationBridgeDeps, -} from "../core/federation-bridge.js"; -import type { AgentIdentity, Room, RoomMessage } from "../core/types.js"; - -function agent(overrides: Partial = {}): AgentIdentity { - return { - id: "local-agent", - version: 1, - name: "agent-name", - harness: "pi", - cwd: "/tmp", - pid: 111, - startedAt: "2026-01-01T00:00:00.000Z", - visibility: "visible", - status: "active", - tags: [], - subscribedRooms: [], - ...overrides, - }; -} - -function room(overrides: Partial = {}): Room { - return { - id: "room-1", - version: 1, - name: "room-name", - type: "public", - owner: "owner-device", - createdAt: "2026-01-01T00:00:00.000Z", - description: "", - members: [], - invited: [], - memberJoins: {}, - memberLeaves: {}, - invitedJoins: {}, - invitedLeaves: {}, - ...overrides, - }; -} - -function message(overrides: Partial = {}): RoomMessage { - return { - id: "msg-1", - from: "sender", - room: "room-1", - content: "hi", - timestamp: "2026-01-01T00:00:00.000Z", - readBy: [], - ...overrides, - }; -} - -interface Harness { - deps: FederationBridgeDeps; - bridge: FederationBridge; - bump: ReturnType; - recordMemberOp: ReturnType; - refreshMembership: ReturnType; - broadcastPatch: ReturnType; - deliverToRoom: ReturnType; - deliverRoomMessageToMember: ReturnType; -} - -function makeHarness(): Harness { - const bump = vi.fn(); - const recordMemberOp = - vi.fn(); - const refreshMembership = - vi.fn(); - const broadcastPatch = vi.fn().mockResolvedValue(undefined); - const deliverToRoom = vi.fn().mockResolvedValue(undefined); - const deliverRoomMessageToMember = vi.fn().mockResolvedValue(undefined); - const deps: FederationBridgeDeps = { - agents: new Map(), - rooms: new Map(), - messages: new Map(), - deliveryEngine: { - bump, - recordMemberOp, - refreshMembership, - broadcastPatch, - deliverToRoom, - deliverRoomMessageToMember, - }, - }; - return { - deps, - bridge: new FederationBridge(deps), - bump, - recordMemberOp, - refreshMembership, - broadcastPatch, - deliverToRoom, - deliverRoomMessageToMember, - }; -} - -describe("FederationBridge — onAgentVisible", () => { - it("stores the remote agent under a fed:@ key, tagged federated, and broadcasts it", async () => { - const h = makeHarness(); - const remote = agent({ id: "remote-id", harness: "codex", tags: ["x"] }); - - await h.bridge.onAgentVisible(remote); - - const stored = h.deps.agents.get("fed:remote-id@codex"); - expect(stored).toBeDefined(); - expect(stored?.id).toBe("fed:remote-id@codex"); - expect(stored?.tags).toEqual(["x", "federated"]); - expect(h.broadcastPatch).toHaveBeenCalledWith({ - type: "agent_upsert", - agent: stored, - }); - }); -}); - -describe("FederationBridge — onAgentGone", () => { - it("marks the matching fed:-prefixed local agent offline and broadcasts it", async () => { - const h = makeHarness(); - h.deps.agents.set( - "fed:remote-id@codex", - agent({ id: "fed:remote-id@codex", status: "active" }), - ); - - await h.bridge.onAgentGone("remote-id"); - - expect(h.deps.agents.get("fed:remote-id@codex")?.status).toBe("offline"); - expect(h.broadcastPatch).toHaveBeenCalledWith({ - type: "agent_offline", - agentId: "fed:remote-id@codex", - }); - }); - - it("does nothing when no local agent matches the fed:@ prefix", async () => { - const h = makeHarness(); - await h.bridge.onAgentGone("no-such-remote"); - expect(h.broadcastPatch).not.toHaveBeenCalled(); - }); - - it("requires the prefix to actually match at the start, not merely appear at the end", async () => { - const h = makeHarness(); - // Ends with "fed:remote-id@" but does not start with it -- must not match. - h.deps.agents.set( - "xfed:remote-id@", - agent({ id: "xfed:remote-id@", status: "active" }), - ); - - await h.bridge.onAgentGone("remote-id"); - - expect(h.deps.agents.get("xfed:remote-id@")?.status).toBe("active"); - expect(h.broadcastPatch).not.toHaveBeenCalled(); - }); -}); - -describe("FederationBridge — onRoomMessage", () => { - it("stores the message and delivers it to every local room member", async () => { - const h = makeHarness(); - h.deps.rooms.set( - "room-1", - room({ id: "room-1", federated: true, members: ["a", "b"] }), - ); - const msg = message(); - - await h.bridge.onRoomMessage("room-1", msg); - - expect(h.deps.messages.get("room-1")).toEqual([msg]); - expect(h.deliverRoomMessageToMember).toHaveBeenCalledWith( - "room-1", - "a", - msg, - ); - expect(h.deliverRoomMessageToMember).toHaveBeenCalledWith( - "room-1", - "b", - msg, - ); - }); - - it("skips fed:-prefixed shadow members -- they have no addressable mesh device of their own, federation.ts's own link forwarding is what reaches the real remote participant", async () => { - const h = makeHarness(); - h.deps.rooms.set( - "room-1", - room({ - id: "room-1", - federated: true, - members: ["a", "fed:remote-agent"], - }), - ); - const msg = message(); - - await h.bridge.onRoomMessage("room-1", msg); - - expect(h.deliverRoomMessageToMember).toHaveBeenCalledWith( - "room-1", - "a", - msg, - ); - expect(h.deliverRoomMessageToMember).not.toHaveBeenCalledWith( - "room-1", - "fed:remote-agent", - msg, - ); - }); - - it("does nothing for a room that isn't federated", async () => { - const h = makeHarness(); - h.deps.rooms.set("room-1", room({ id: "room-1", federated: false })); - - await h.bridge.onRoomMessage("room-1", message()); - - expect(h.deps.messages.get("room-1")).toBeUndefined(); - expect(h.deliverRoomMessageToMember).not.toHaveBeenCalled(); - }); -}); - -describe("FederationBridge — onRoomJoin", () => { - it("adds a fed:-prefixed shadow member, records the join, and broadcasts the room when the member is new", async () => { - const h = makeHarness(); - h.deps.rooms.set( - "room-1", - room({ id: "room-1", federated: true, members: [] }), - ); - - await h.bridge.onRoomJoin("room-1", "remote-agent", "Remote Name"); - - expect(h.bump).toHaveBeenCalledTimes(1); - expect(h.recordMemberOp).toHaveBeenCalledWith( - expect.anything(), - "member", - "join", - "fed:remote-agent", - ); - expect(h.refreshMembership).toHaveBeenCalledTimes(1); - expect(h.broadcastPatch).toHaveBeenCalledWith({ - type: "room_upsert", - room: expect.anything(), - }); - }); - - it("skips the join-recording step entirely when the shadow member is already present", async () => { - const h = makeHarness(); - h.deps.rooms.set( - "room-1", - room({ id: "room-1", federated: true, members: ["fed:remote-agent"] }), - ); - - await h.bridge.onRoomJoin("room-1", "remote-agent", "Remote Name"); - - expect(h.bump).not.toHaveBeenCalled(); - expect(h.recordMemberOp).not.toHaveBeenCalled(); - expect(h.broadcastPatch).not.toHaveBeenCalled(); - }); - - it("always notifies local members of the join, even when the shadow member already existed", async () => { - const h = makeHarness(); - h.deps.rooms.set( - "room-1", - room({ id: "room-1", federated: true, members: ["fed:remote-agent"] }), - ); - - await h.bridge.onRoomJoin("room-1", "remote-agent", "Remote Name"); - - expect(h.deliverToRoom).toHaveBeenCalledWith( - "room-1", - { type: "member_joined", room: "room-1", agent: "fed:remote-agent" }, - "fed:remote-agent", - ); - }); - - it("does nothing at all for a room that isn't federated", async () => { - const h = makeHarness(); - h.deps.rooms.set("room-1", room({ id: "room-1", federated: false })); - - await h.bridge.onRoomJoin("room-1", "remote-agent", "Remote Name"); - - expect(h.bump).not.toHaveBeenCalled(); - expect(h.deliverToRoom).not.toHaveBeenCalled(); - }); -}); - -describe("FederationBridge — onRoomLeave", () => { - it("records the leave, bumps and broadcasts the room, and notifies local members", async () => { - const h = makeHarness(); - h.deps.rooms.set( - "room-1", - room({ id: "room-1", federated: true, members: ["fed:remote-agent"] }), - ); - - await h.bridge.onRoomLeave("room-1", "remote-agent"); - - expect(h.bump).toHaveBeenCalledTimes(1); - expect(h.recordMemberOp).toHaveBeenCalledWith( - expect.anything(), - "member", - "leave", - "fed:remote-agent", - ); - expect(h.refreshMembership).toHaveBeenCalledTimes(1); - expect(h.broadcastPatch).toHaveBeenCalledWith({ - type: "room_upsert", - room: expect.anything(), - }); - expect(h.deliverToRoom).toHaveBeenCalledWith("room-1", { - type: "member_left", - room: "room-1", - agent: "fed:remote-agent", - }); - }); - - it("does nothing at all for a room that isn't federated", async () => { - const h = makeHarness(); - h.deps.rooms.set("room-1", room({ id: "room-1", federated: false })); - - await h.bridge.onRoomLeave("room-1", "remote-agent"); - - expect(h.bump).not.toHaveBeenCalled(); - expect(h.deliverToRoom).not.toHaveBeenCalled(); - }); -}); - -describe("FederationBridge — getVisibleAgents", () => { - it("includes a visible, non-federated agent", () => { - const h = makeHarness(); - const a = agent({ id: "local-a", visibility: "visible" }); - h.deps.agents.set(a.id, a); - expect(h.bridge.getVisibleAgents()).toEqual([a]); - }); - - it("excludes an agent that isn't visible", () => { - const h = makeHarness(); - h.deps.agents.set( - "hidden-a", - agent({ id: "hidden-a", visibility: "hidden" }), - ); - expect(h.bridge.getVisibleAgents()).toEqual([]); - }); - - it("excludes an already-federated-in agent even when visible", () => { - const h = makeHarness(); - h.deps.agents.set( - "fed:remote@codex", - agent({ id: "fed:remote@codex", visibility: "visible" }), - ); - expect(h.bridge.getVisibleAgents()).toEqual([]); - }); - - it("requires the fed: prefix to be at the start, not merely present at the end", () => { - const h = makeHarness(); - // Ends with "fed:" but does not start with it -- must be included. - const a = agent({ id: "xxxfed:", visibility: "visible" }); - h.deps.agents.set(a.id, a); - expect(h.bridge.getVisibleAgents()).toEqual([a]); - }); -}); - -describe("FederationBridge — getFederatedRoomMemberships", () => { - it("includes only federated rooms, with fed:-prefixed members filtered out", () => { - const h = makeHarness(); - h.deps.rooms.set( - "fed-room", - room({ - id: "fed-room", - federated: true, - members: ["local-a", "fed:remote-b"], - }), - ); - h.deps.rooms.set( - "plain-room", - room({ id: "plain-room", federated: false, members: ["local-c"] }), - ); - - const result = h.bridge.getFederatedRoomMemberships(); - - expect(result.get("fed-room")).toEqual(["local-a"]); - expect(result.has("plain-room")).toBe(false); - }); - - it("requires the fed: prefix to be at the start, not merely present at the end, when filtering members", () => { - const h = makeHarness(); - h.deps.rooms.set( - "fed-room", - room({ id: "fed-room", federated: true, members: ["xxxfed:"] }), - ); - - const result = h.bridge.getFederatedRoomMemberships(); - - expect(result.get("fed-room")).toEqual(["xxxfed:"]); - }); -}); diff --git a/src/test/federation.integration.test.ts b/src/test/federation.integration.test.ts deleted file mode 100644 index 3cdd1d6c..00000000 --- a/src/test/federation.integration.test.ts +++ /dev/null @@ -1,300 +0,0 @@ -/** - * Federation integration test — verifies that two MeshStore instances on different "machines" (simulated via separate TCP meshes) can federate through coordinator-to-coordinator TLS links, and that an inbound link presenting an untrusted certificate is rejected outright. - * - * Tests: - * 0. An inbound connection with no pinned fingerprint is rejected - * 1. Establish federation link between two meshes once both fingerprints are trusted - * 2. Agent presence propagates across federation - * 3. Messages in federated rooms propagate across federation - * 4. Non-federated rooms are isolated (messages never cross) - * 5. Federation link listing - * 6. Disconnect federation link - * - * Run: node dist/test/federation.integration.test.js - */ - -import { MeshStore } from "../core/mesh-store.js"; -import type { DeliveryEvent, RoomMessage } from "../core/types.js"; -import * as net from "node:net"; -import { test, expect } from "vitest"; -import { wireTestTransport } from "./test-transport.js"; - -// Use high ports to avoid collisions with real meshes -const MESH_A_PORT = 28876; -const MESH_B_PORT = 28877; - -// --------------------------------------------------------------------------- -// Timing constants — settle windows for asynchronous mesh/federation propagation. There is no "operation complete" signal for these steps, so the test waits a fixed budget rather than polling. -// --------------------------------------------------------------------------- - -/** Milliseconds to wait after a mesh-setup step (creating a mesh, starting the federation listener) for local state to settle. */ -const MESH_SETUP_SETTLE_MS = 100; -/** Milliseconds to wait for a freshly established federation link to settle before exercising it. */ -const FED_LINK_SETTLE_MS = 200; -/** Milliseconds to wait for a federated room (or a join against its mirror) to propagate across the federation link. */ -const FED_ROOM_SETTLE_MS = 200; -/** Milliseconds to wait for a room message to propagate across an established federation link. */ -const FED_MESSAGE_PROPAGATION_MS = 500; -/** Milliseconds to wait for a non-federated (local-only) room to settle after creation. */ -const LOCAL_ROOM_SETTLE_MS = 100; -/** Milliseconds to wait for a local-only message to settle, confirming it does not leak across federation. */ -const LOCAL_MESSAGE_SETTLE_MS = 100; -/** Milliseconds to wait for a federation link disconnect to propagate. */ -const FED_DISCONNECT_SETTLE_MS = 200; - -async function sleep(ms: number): Promise { - await new Promise((resolve) => { - setTimeout(resolve, ms); - }); -} - -// --------------------------------------------------------------------------- -// Helpers -// --------------------------------------------------------------------------- - -async function createMesh( - name: string, - coordinatorPort: number, -): Promise<{ - store: MeshStore; - deliveries: DeliveryEvent[]; -}> { - const store = new MeshStore(coordinatorPort); - await wireTestTransport(store); - const deliveries: DeliveryEvent[] = []; - store.onDelivery = (_agentId: string, event: DeliveryEvent) => { - deliveries.push(event); - }; - - await store.init(); - await store.registerAgent({ - name, - harness: "test", - cwd: `/test/${name}`, - pid: process.pid, - visibility: "visible", - tags: [], - }); - - return { store, deliveries }; -} - -/** Find a free port on localhost. */ -async function findFreePort(): Promise { - return await new Promise((resolve, reject) => { - const server = net.createServer(); - server.listen(0, "127.0.0.1", () => { - const addr = server.address(); - if (typeof addr === "object" && addr !== null) { - const port = addr.port; - server.close(() => resolve(port)); - } else { - server.close(() => reject(new Error("Failed to get port"))); - } - }); - server.on("error", reject); - }); -} - -// --------------------------------------------------------------------------- -// Test runner -// --------------------------------------------------------------------------- - -async function main(): Promise { - console.log("=== Federation Integration Tests ===\n"); - - console.log("Creating mesh A (coordinator)..."); - const a = await createMesh("mesh-a-agent", MESH_A_PORT); - await sleep(MESH_SETUP_SETTLE_MS); - - console.log("Creating mesh B (coordinator)..."); - const b = await createMesh("mesh-b-agent", MESH_B_PORT); - await sleep(MESH_SETUP_SETTLE_MS); - - const fedPort = await findFreePort(); - console.log(`Using federation port ${String(fedPort)}`); - - // Start A's real production federation listener (fedListen -> FederationManager.listen -> handleInbound, the same path a deployed coordinator uses — not a hand-rolled test-only TLS server). - await a.store.fedListen("127.0.0.1", fedPort); - await sleep(MESH_SETUP_SETTLE_MS); - - // --- Test 0: untrusted inbound connection is rejected --- - console.log("\nTest 0: untrusted connection is rejected..."); - await expect( - b.store.fedConnect("127.0.0.1", fedPort), - "Connecting before either side has pinned the other's fingerprint should be rejected", - ).rejects.toThrow(/rejected|not in the trusted-fingerprint allowlist/i); - expect( - b.store.fedLinks().length, - "B should have no federation links after a rejected attempt", - ).toBe(0); - console.log(" Rejected as expected — no link was created."); - - // --- Pin fingerprints on both sides, mirroring what an operator does out of band --- - console.log("\nPinning fingerprints on both sides..."); - const fingerprintA = a.store.getFederationFingerprint(); - const fingerprintB = b.store.getFederationFingerprint(); - expect( - fingerprintA.length > 0, - "A should report its own fingerprint", - ).toBeTruthy(); - expect( - fingerprintB.length > 0, - "B should report its own fingerprint", - ).toBeTruthy(); - await a.store.fedTrust(fingerprintB); - await b.store.fedTrust(fingerprintA); - expect(a.store.fedTrustedFingerprints()).toEqual([fingerprintB]); - expect(b.store.fedTrustedFingerprints()).toEqual([fingerprintA]); - - // --- Test 1: Establish federation link now that both sides trust each other --- - console.log("\nTest 1: Establish federation link..."); - const linkId = await b.store.fedConnect("127.0.0.1", fedPort); - console.log(` Link established: ${linkId}`); - expect(linkId, "Should return a link ID").toBeTruthy(); - - const linksB = b.store.fedLinks(); - expect(linksB.length, "B should have 1 federation link").toBe(1); - const link = linksB[0]; - expect(link, "Link should exist").toBeTruthy(); - if (link === undefined) throw new Error("Link should exist"); - expect( - link.remoteMeshId.length > 0, - "Remote mesh ID should be present", - ).toBeTruthy(); - - await sleep(FED_LINK_SETTLE_MS); - - // --- Test 2: Agent presence propagates --- - console.log("Test 2: Agent presence propagates..."); - const agentsB = await b.store.listAgents(b.store.peerId); - console.log(` B sees ${String(agentsB.length)} agent(s)`); - const fedAgentsB = agentsB.filter((ag) => ag.tags.includes("federated")); - expect( - fedAgentsB.length >= 1, - "B should see at least 1 federated agent from A", - ).toBeTruthy(); - - const agentsA = await a.store.listAgents(a.store.peerId); - console.log(` A sees ${String(agentsA.length)} agent(s)`); - const fedAgentsA = agentsA.filter((ag) => ag.tags.includes("federated")); - expect( - fedAgentsA.length >= 1, - "A should see at least 1 federated agent from B", - ).toBeTruthy(); - - // --- Test 3: Federated room messages propagate --- - console.log("Test 3: Federated room messages propagate..."); - - const fedRoomId = `fed-room-${String(Date.now())}`; - const fedRoom = await a.store.createRoom({ - name: fedRoomId, - type: "public", - owner: a.store.peerId, - description: "Federated test room", - federated: true, - }); - console.log(` Created federated room: ${fedRoom.id}`); - await sleep(FED_ROOM_SETTLE_MS); - - // Federation matches a room across the two separate meshes by literal id equality (handleFedRoomMessage/Join/Leave all key off the incoming roomId string directly) -- an owner-rooted path only coincides on both sides when both sides construct it from the same owner, so B's mirror of A's room is created with A's own peerId as owner, not B's. - const fedRoomB = await b.store.createRoom({ - name: fedRoomId, - type: "public", - owner: a.store.peerId, - description: "Federated test room", - federated: true, - }); - console.log(` Created matching federated room on B: ${fedRoomB.id}`); - // createRoom's default members is [owner] -- since owner is A's peerId (to make the id match), B's own local agent must explicitly join its mirror for handleFedRoomMessage's local-delivery loop to reach it. - await b.store.joinRoom(fedRoomB.id, b.store.peerId); - await sleep(FED_ROOM_SETTLE_MS); - - a.deliveries.length = 0; - b.deliveries.length = 0; - - const msg = await a.store.sendRoomMessage( - fedRoom.id, - a.store.peerId, - "Hello from mesh A!", - ); - console.log(` A sent: "${msg.content}"`); - await sleep(FED_MESSAGE_PROPAGATION_MS); - - const fedMsgs = b.deliveries.filter( - (e) => - e.type === "room_message" && e.message.content === "Hello from mesh A!", - ); - console.log(` B received ${String(fedMsgs.length)} federated message(s)`); - expect( - fedMsgs.length >= 1, - "B should receive the federated room message", - ).toBeTruthy(); - - // --- Test 4: Non-federated rooms are isolated --- - console.log("Test 4: Non-federated rooms are isolated..."); - - const localRoomId = `local-room-${String(Date.now())}`; - console.log(` Creating non-federated room: ${localRoomId}`); - const localRoom = await a.store.createRoom({ - name: localRoomId, - type: "public", - owner: a.store.peerId, - description: "Local-only room", - // federated defaults to false - }); - console.log( - ` Created non-federated room: ${localRoom.id}, federated=${String(localRoom.federated)}`, - ); - await sleep(LOCAL_ROOM_SETTLE_MS); - - b.deliveries.length = 0; - - console.log(" Sending message in non-federated room..."); - await a.store.sendRoomMessage( - localRoom.id, - a.store.peerId, - "Secret local message", - ); - console.log(" Message sent."); - await sleep(LOCAL_MESSAGE_SETTLE_MS); - - const leakedMsgs = b.deliveries.filter( - (e) => - e.type === "room_message" && e.message.content === "Secret local message", - ); - console.log(` B received ${String(leakedMsgs.length)} leaked message(s)`); - expect( - leakedMsgs.length, - "B should NOT receive non-federated room messages", - ).toBe(0); - - // --- Test 5: Federation link listing --- - console.log("Test 5: Federation link listing..."); - const linksA = a.store.fedLinks(); - expect(linksA.length, "A should have 1 federation link").toBe(1); - console.log(` A links: ${linksA.map((l) => l.remoteName).join(", ")}`); - - // --- Test 6: Disconnect federation link --- - console.log("Test 6: Disconnect federation link..."); - await b.store.fedDisconnect(linkId); - await sleep(FED_DISCONNECT_SETTLE_MS); - - const linksAfter = b.store.fedLinks(); - expect( - linksAfter.length, - "B should have 0 federation links after disconnect", - ).toBe(0); - - // --- Cleanup --- - console.log("\nCleaning up..."); - await a.store.fedStopListening(); - await a.store.shutdown(); - await b.store.shutdown(); - - console.log("\n✓ All federation tests passed!"); -} - -test("federates two meshes over coordinator-to-coordinator TLS links, propagating presence and room messages while rejecting an untrusted inbound link", async () => { - await main(); -}); diff --git a/src/test/mesh-store-orchestration.test.ts b/src/test/mesh-store-orchestration.test.ts index 936a80da..e1f8a058 100644 --- a/src/test/mesh-store-orchestration.test.ts +++ b/src/test/mesh-store-orchestration.test.ts @@ -1,5 +1,5 @@ /** - * Direct unit tests for MeshStore's own orchestration logic -- requireTransport/requireIdentity's guard errors, the connected getter, init()'s connect-vs-becomeCoordinator-vs-EADDRINUSE branching, the events getter's dispatch table, federation/listener/room-join-approval passthroughs, and shutdown() -- as opposed to the collaborator-owned behaviour wireTestTransport-based integration tests already cover end-to-end. MeshStore's constructor takes no injectable deps (unlike its collaborators), so these tests use a hand-built fake MeshTransport passed to the real setTransport(), and reach the private roomProtocol/connectionApproval/peerLifecycle/deliveryEngine/agentRegistry collaborators via a narrow, explicitly-justified cast -- TypeScript's `private` is compile-time only, and asserting a delegating wrapper actually calls through to the collaborator that owns the real implementation is exactly the kind of whitebox check no public-API-only test can express for a one-line pass-through. + * Direct unit tests for MeshStore's own orchestration logic -- requireTransport/requireIdentity's guard errors, the connected getter, init()'s connect-vs-becomeCoordinator-vs-EADDRINUSE branching, the events getter's dispatch table, listener/room-join-approval passthroughs, and shutdown() -- as opposed to the collaborator-owned behaviour wireTestTransport-based integration tests already cover end-to-end. MeshStore's constructor takes no injectable deps (unlike its collaborators), so these tests use a hand-built fake MeshTransport passed to the real setTransport(), and reach the private roomProtocol/connectionApproval/peerLifecycle/deliveryEngine/agentRegistry collaborators via a narrow, explicitly-justified cast -- TypeScript's `private` is compile-time only, and asserting a delegating wrapper actually calls through to the collaborator that owns the real implementation is exactly the kind of whitebox check no public-API-only test can express for a one-line pass-through. */ import { beforeEach, describe, expect, it, vi } from "vitest"; import { MeshStore } from "../core/mesh-store.js"; @@ -573,61 +573,6 @@ describe("MeshStore — room-join approval passthroughs", () => { }); }); -describe("MeshStore — federation passthroughs", () => { - it("forwards fedConnect/fedDisconnect/fedLinks/fedTrust/fedUntrust/fedTrustedFingerprints/fedListen/fedStopListening to the real, public federation manager", async () => { - const store = new MeshStore(); - store.setTransport(fakeTransport()); - - const connectSpy = vi - .spyOn(store.federation, "connect") - .mockResolvedValue("link-1"); - await expect(store.fedConnect("host", 1, "name")).resolves.toBe("link-1"); - expect(connectSpy).toHaveBeenCalledWith("host", 1, "name"); - - const disconnectSpy = vi - .spyOn(store.federation, "disconnect") - .mockResolvedValue(undefined); - await store.fedDisconnect("link-1"); - expect(disconnectSpy).toHaveBeenCalledWith("link-1"); - - const linksSpy = vi - .spyOn(store.federation, "listLinks") - .mockReturnValue([]); - store.fedLinks(); - expect(linksSpy).toHaveBeenCalledTimes(1); - - const trustSpy = vi - .spyOn(store.federation, "addTrustedFingerprint") - .mockReturnValue(undefined); - await store.fedTrust("fingerprint-a"); - expect(trustSpy).toHaveBeenCalledWith("fingerprint-a"); - - const untrustSpy = vi - .spyOn(store.federation, "removeTrustedFingerprint") - .mockReturnValue(undefined); - await store.fedUntrust("fingerprint-a"); - expect(untrustSpy).toHaveBeenCalledWith("fingerprint-a"); - - const listTrustedSpy = vi - .spyOn(store.federation, "listTrustedFingerprints") - .mockReturnValue([]); - store.fedTrustedFingerprints(); - expect(listTrustedSpy).toHaveBeenCalledTimes(1); - - const listenSpy = vi - .spyOn(store.federation, "listen") - .mockResolvedValue(undefined); - await store.fedListen("host", 2); - expect(listenSpy).toHaveBeenCalledWith("host", 2); - - const stopListeningSpy = vi - .spyOn(store.federation, "stopListening") - .mockResolvedValue(undefined); - await store.fedStopListening(); - expect(stopListeningSpy).toHaveBeenCalledTimes(1); - }); -}); - describe("MeshStore — shutdown()", () => { let store: MeshStore; let transport: ReturnType; @@ -638,24 +583,19 @@ describe("MeshStore — shutdown()", () => { store.setTransport(transport); }); - it("stops the stale-agent checker, shuts down federation, and shuts down the transport", async () => { + it("stops the stale-agent checker and shuts down the transport", async () => { const staleAgentChecker = collaborator(store, "staleAgentChecker") as { stop: ReturnType; }; const stopSpy = vi.spyOn(staleAgentChecker, "stop"); - const federationShutdownSpy = vi - .spyOn(store.federation, "shutdown") - .mockResolvedValue(undefined); await store.shutdown(); expect(stopSpy).toHaveBeenCalledTimes(1); - expect(federationShutdownSpy).toHaveBeenCalledTimes(1); expect(transport.shutdown).toHaveBeenCalledTimes(1); }); it("broadcasts agent_offline for its own self agent when one is registered", async () => { - vi.spyOn(store.federation, "shutdown").mockResolvedValue(undefined); const agent = await store.registerAgent({ name: "self", harness: "pi", @@ -681,7 +621,6 @@ describe("MeshStore — shutdown()", () => { }); it("clears every pending markRead timer", async () => { - vi.spyOn(store.federation, "shutdown").mockResolvedValue(undefined); const pending = collaborator(store, "pendingMarkReadTimers") as ReturnType< typeof setTimeout >[]; @@ -703,7 +642,6 @@ describe("MeshStore — shutdown()", () => { it("actually sets isShutDown, observable via DeliveryEngine no longer scheduling markRead timers afterward", async () => { // fireLocalDelivery returns before ever reaching the isShutDown-guarded push unless onDelivery is set -- without this, the test would pass for both real code and a mutant, since it never reaches the line under test. store.onDelivery = vi.fn(); - vi.spyOn(store.federation, "shutdown").mockResolvedValue(undefined); await store.shutdown(); const pending = collaborator(store, "pendingMarkReadTimers") as unknown[]; @@ -723,7 +661,6 @@ describe("MeshStore — shutdown()", () => { }); it("does not broadcast agent_offline when no self agent was ever registered", async () => { - vi.spyOn(store.federation, "shutdown").mockResolvedValue(undefined); const deliveryEngine = collaborator(store, "deliveryEngine") as { broadcastPatch: ReturnType; }; diff --git a/src/test/room-lifecycle-membership.test.ts b/src/test/room-lifecycle-membership.test.ts index b1492c71..2104fcaa 100644 --- a/src/test/room-lifecycle-membership.test.ts +++ b/src/test/room-lifecycle-membership.test.ts @@ -114,8 +114,6 @@ interface Harness { broadcastPatch: ReturnType; deliverToRoom: ReturnType; deliverToMember: ReturnType; - broadcastRoomJoin: ReturnType; - broadcastRoomLeave: ReturnType; sendRoomRequest: ReturnType; broadcastRevocation: ReturnType; } @@ -155,8 +153,6 @@ async function makeHarness(): Promise { const broadcastPatch = vi.fn().mockResolvedValue(undefined); const deliverToRoom = vi.fn().mockResolvedValue(undefined); const deliverToMember = vi.fn().mockResolvedValue(undefined); - const broadcastRoomJoin = vi.fn().mockResolvedValue(undefined); - const broadcastRoomLeave = vi.fn().mockResolvedValue(undefined); const broadcastRevocation = vi.fn().mockResolvedValue(undefined); const sendRoomRequest = vi.fn().mockResolvedValue({ result: "error", @@ -188,7 +184,6 @@ async function makeHarness(): Promise { deliverToRoom, deliverToMember, }, - federation: { broadcastRoomJoin, broadcastRoomLeave }, }; return { @@ -202,8 +197,6 @@ async function makeHarness(): Promise { broadcastPatch, deliverToRoom, deliverToMember, - broadcastRoomJoin, - broadcastRoomLeave, sendRoomRequest, broadcastRevocation, }; @@ -701,42 +694,6 @@ describe("RoomLifecycle — leaveRoom / leaveRemoteRoom", () => { ); }); - it("notifies federated links only when the room is federated", async () => { - const federated = await makeHarness(); - federated.deps.rooms.set( - "room-1", - room({ - id: "room-1", - owner: federated.ids.ownerId, - members: [federated.ids.ownerId, federated.ids.memberId], - memberJoins: { - [federated.ids.ownerId]: 1, - [federated.ids.memberId]: 1, - }, - federated: true, - }), - ); - await federated.lifecycle.leaveRoom("room-1", federated.ids.memberId); - expect(federated.broadcastRoomLeave).toHaveBeenCalledWith( - "room-1", - federated.ids.memberId, - ); - - const plain = await makeHarness(); - plain.deps.rooms.set( - "room-1", - room({ - id: "room-1", - owner: plain.ids.ownerId, - members: [plain.ids.ownerId, plain.ids.memberId], - memberJoins: { [plain.ids.ownerId]: 1, [plain.ids.memberId]: 1 }, - federated: false, - }), - ); - await plain.lifecycle.leaveRoom("room-1", plain.ids.memberId); - expect(plain.broadcastRoomLeave).not.toHaveBeenCalled(); - }); - it("destroys the room when the last remaining member is also its own owner", async () => { const h = await makeHarness(); h.deps.rooms.set( diff --git a/src/test/room-lifecycle-remote.test.ts b/src/test/room-lifecycle-remote.test.ts index 940e0d3c..e2396064 100644 --- a/src/test/room-lifecycle-remote.test.ts +++ b/src/test/room-lifecycle-remote.test.ts @@ -114,8 +114,6 @@ interface Harness { broadcastPatch: ReturnType; deliverToRoom: ReturnType; deliverToMember: ReturnType; - broadcastRoomJoin: ReturnType; - broadcastRoomLeave: ReturnType; sendRoomRequest: ReturnType; broadcastRevocation: ReturnType; } @@ -153,8 +151,6 @@ async function makeHarness(): Promise { const broadcastPatch = vi.fn().mockResolvedValue(undefined); const deliverToRoom = vi.fn().mockResolvedValue(undefined); const deliverToMember = vi.fn().mockResolvedValue(undefined); - const broadcastRoomJoin = vi.fn().mockResolvedValue(undefined); - const broadcastRoomLeave = vi.fn().mockResolvedValue(undefined); const broadcastRevocation = vi.fn().mockResolvedValue(undefined); const sendRoomRequest = vi.fn().mockResolvedValue({ result: "error", @@ -186,7 +182,6 @@ async function makeHarness(): Promise { deliverToRoom, deliverToMember, }, - federation: { broadcastRoomJoin, broadcastRoomLeave }, }; return { @@ -200,8 +195,6 @@ async function makeHarness(): Promise { broadcastPatch, deliverToRoom, deliverToMember, - broadcastRoomJoin, - broadcastRoomLeave, sendRoomRequest, broadcastRevocation, }; @@ -314,25 +307,20 @@ describe("RoomLifecycle — refreshRoomMembers", () => { const refreshed = await h.lifecycle.refreshRoomMembers(roomPath); expect(refreshed.version).toBe(1); expect(refreshed.description).toBe(""); - expect(refreshed.federated).toBe(false); expect(refreshed.memberJoins).toEqual({ [h.ids.ownerId]: 1 }); expect(refreshed.invited).toEqual([]); }); - it("increments the version relative to the existing local copy, and preserves its federated flag", async () => { + it("increments the version relative to the existing local copy", async () => { const h = await makeHarness(); const roomPath = ownerNamedRoomPath(h.ids.memberId, "r"); - h.deps.rooms.set( - roomPath, - room({ id: roomPath, version: 1, federated: true }), - ); + h.deps.rooms.set(roomPath, room({ id: roomPath, version: 1 })); const { slot } = h.deps.requireIdentity(); const token = await mintRoomToken(h.ids.ownerPort, h.ids.ownerId, roomPath); saveRoomToken(slot, roomPath, token); h.sendRoomRequest.mockResolvedValue(roomMembersOkOutcome([h.ids.ownerId])); const refreshed = await h.lifecycle.refreshRoomMembers(roomPath); expect(refreshed.version).toBe(2); - expect(refreshed.federated).toBe(true); }); it("prefers the room-state extension's own fields over the existing local copy's", async () => { diff --git a/src/test/room-lifecycle.test.ts b/src/test/room-lifecycle.test.ts index 949ac759..661596b5 100644 --- a/src/test/room-lifecycle.test.ts +++ b/src/test/room-lifecycle.test.ts @@ -112,8 +112,6 @@ interface Harness { broadcastPatch: ReturnType; deliverToRoom: ReturnType; deliverToMember: ReturnType; - broadcastRoomJoin: ReturnType; - broadcastRoomLeave: ReturnType; sendRoomRequest: ReturnType; broadcastRevocation: ReturnType; } @@ -151,8 +149,6 @@ async function makeHarness(): Promise { const broadcastPatch = vi.fn().mockResolvedValue(undefined); const deliverToRoom = vi.fn().mockResolvedValue(undefined); const deliverToMember = vi.fn().mockResolvedValue(undefined); - const broadcastRoomJoin = vi.fn().mockResolvedValue(undefined); - const broadcastRoomLeave = vi.fn().mockResolvedValue(undefined); const broadcastRevocation = vi.fn().mockResolvedValue(undefined); const sendRoomRequest = vi.fn().mockResolvedValue({ result: "error", @@ -184,7 +180,6 @@ async function makeHarness(): Promise { deliverToRoom, deliverToMember, }, - federation: { broadcastRoomJoin, broadcastRoomLeave }, }; return { @@ -198,8 +193,6 @@ async function makeHarness(): Promise { broadcastPatch, deliverToRoom, deliverToMember, - broadcastRoomJoin, - broadcastRoomLeave, sendRoomRequest, broadcastRevocation, }; @@ -287,29 +280,6 @@ describe("RoomLifecycle — createRoom", () => { }); }); - it("defaults federated to false when omitted", async () => { - const h = await makeHarness(); - const created = await h.lifecycle.createRoom({ - name: "r", - type: "public", - owner: h.ids.ownerId, - description: "", - }); - expect(created.federated).toBe(false); - }); - - it("honours an explicit federated: true", async () => { - const h = await makeHarness(); - const created = await h.lifecycle.createRoom({ - name: "r", - type: "public", - owner: h.ids.ownerId, - description: "", - federated: true, - }); - expect(created.federated).toBe(true); - }); - it("seeds an empty message history and broadcasts a room_upsert", async () => { const h = await makeHarness(); const created = await h.lifecycle.createRoom({ @@ -580,7 +550,6 @@ describe("RoomLifecycle — joinRoom / joinRemoteRoom", () => { [h.ids.ownerId]: 1, [h.ids.memberId]: 1, }); - expect(stored?.federated).toBe(false); const { slot } = h.deps.requireIdentity(); expect(loadRoomTokens(slot)[roomPath]).toBeDefined(); }); @@ -755,34 +724,4 @@ describe("RoomLifecycle — joinRoom / joinRemoteRoom", () => { h.ids.memberId, ); }); - - it("notifies federated links only when the room is federated", async () => { - const federated = await makeHarness(); - federated.deps.rooms.set( - "room-1", - room({ - id: "room-1", - type: "public", - owner: federated.ids.ownerId, - members: [], - federated: true, - }), - ); - await federated.lifecycle.joinRoom("room-1", federated.ids.memberId); - expect(federated.broadcastRoomJoin).toHaveBeenCalledTimes(1); - - const plain = await makeHarness(); - plain.deps.rooms.set( - "room-1", - room({ - id: "room-1", - type: "public", - owner: plain.ids.ownerId, - members: [], - federated: false, - }), - ); - await plain.lifecycle.joinRoom("room-1", plain.ids.memberId); - expect(plain.broadcastRoomJoin).not.toHaveBeenCalled(); - }); }); diff --git a/src/test/room-messaging-durable-send.test.ts b/src/test/room-messaging-durable-send.test.ts index 765c0506..415bea96 100644 --- a/src/test/room-messaging-durable-send.test.ts +++ b/src/test/room-messaging-durable-send.test.ts @@ -88,7 +88,6 @@ async function makeHarness() { dataStorage, }), roomProtocol: { sendRoomRequestToMember: async () => undefined }, - federation: { forwardRoomMessage: async () => undefined }, }; return { diff --git a/src/test/room-messaging.test.ts b/src/test/room-messaging.test.ts index c6393851..0aab88ba 100644 --- a/src/test/room-messaging.test.ts +++ b/src/test/room-messaging.test.ts @@ -66,7 +66,6 @@ function agent(overrides: Partial = {}): AgentIdentity { function makeHarness() { const sendRoomRequestToMember = vi.fn().mockResolvedValue(undefined); - const forwardRoomMessage = vi.fn().mockResolvedValue(undefined); const deps: RoomMessagingDeps = { rooms: new Map(), messages: new Map(), @@ -80,13 +79,11 @@ function makeHarness() { dataStorage: {} as never, }), roomProtocol: { sendRoomRequestToMember }, - federation: { forwardRoomMessage }, }; return { deps, messaging: new RoomMessaging(deps), sendRoomRequestToMember, - forwardRoomMessage, }; } @@ -185,18 +182,6 @@ describe("RoomMessaging — sendRoomMessage", () => { expect(withoutBehavior).not.toHaveProperty("streamingBehavior"); }); - it("forwards to federated links only when the room is federated", async () => { - const federated = makeHarness(); - federated.deps.rooms.set("room-1", room({ federated: true })); - await federated.messaging.sendRoomMessage("room-1", FROM_DEVICE_ID, "hi"); - expect(federated.forwardRoomMessage).toHaveBeenCalledTimes(1); - - const plain = makeHarness(); - plain.deps.rooms.set("room-1", room({ federated: false })); - await plain.messaging.sendRoomMessage("room-1", FROM_DEVICE_ID, "hi"); - expect(plain.forwardRoomMessage).not.toHaveBeenCalled(); - }); - it("sends a directed request to every other member, never to the sender itself", async () => { const h = makeHarness(); h.deps.rooms.set(