chore: clear the ground for the wire-mesh migration (P0)

.github/workflows/ci.yml

@@ -40,6 +40,7 @@ jobs:
   test:
     name: Test
     runs-on: ubuntu-latest
+    timeout-minutes: 10
     steps:
       - uses: actions/checkout@v7
       - uses: pnpm/action-setup@v6

package.json

@@ -52,7 +52,7 @@
     "generate:server-json": "tsx scripts/sync-release-metadata.ts",
     "publish:mcp-registry": "tsx scripts/publish-mcp-registry.ts",
     "lint": "eslint .",
-    "test": "node --test --test-concurrency=1 dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js dist/test/identity-cert.test.js dist/test/broadcast-window.integration.test.js dist/test/identity-restart.integration.test.js dist/test/mesh-e2e.integration.test.js dist/test/ws-broadcast-window.integration.test.js dist/test/tls-transport.integration.test.js dist/test/peer-id-verification.integration.test.js dist/test/become-coordinator-actual-port.integration.test.js dist/test/state-sync-convergence.test.js dist/test/downtime-replay.test.js dist/test/downtime-replay.integration.test.js dist/test/filestore.test.js dist/test/handshake.test.js",
+    "test": "node --test --test-concurrency=1 dist/test/coordinator-socket-error.integration.test.js dist/test/identity-store.test.js dist/test/identity-cert.test.js dist/test/broadcast-window.integration.test.js dist/test/identity-restart.integration.test.js dist/test/mesh-e2e.integration.test.js dist/test/tls-transport.integration.test.js dist/test/peer-id-verification.integration.test.js dist/test/become-coordinator-actual-port.integration.test.js dist/test/state-sync-convergence.test.js dist/test/downtime-replay.test.js dist/test/downtime-replay.integration.test.js dist/test/filestore.test.js dist/test/handshake.test.js dist/test/approval.integration.test.js dist/test/listener-policy.integration.test.js dist/test/mesh-smoke.integration.test.js",
     "test:visibility": "node --test dist/test/visibility.integration.test.js",
     "test:delivery": "node dist/test/delivery-receipt.runner.js",
     "test:federation": "node dist/test/federation.integration.test.js",

src/core/comms-store.ts

@@ -1,26 +1,23 @@
 /**
  * CommsStore — abstract interface for the agent communication store.
  *
- * Two implementations:
- *   FileStore — filesystem-backed, no server process (fallback)
- *   MeshStore — TCP peer mesh, in-memory, real-time push (preferred)
+ * Two implementations: FileStore — filesystem-backed, no server process (fallback) MeshStore — TCP peer mesh, in-memory, real-time push (preferred)
  *
  * 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.
  */
 
 import type {
   AgentIdentity,
   DeliveryEvent,
   DmMessage,
-  NetworkInterface,
   Room,
   RoomMessage,
   RoomType,
   StreamingBehavior,
   Visibility,
 } from "./types.js";
-import type { ListenerInfo } from "./transport.js";
-import type { FedLink } from "./federation.js";
 
 export interface CommsStore {
   // -- Identity --
@@ -95,43 +92,6 @@ export interface CommsStore {
   deliver(agentId: string, event: DeliveryEvent): Promise<void>;
   drainDelivery(agentId: string): Promise<DeliveryEvent[]>;
 
-  // -- Listener management (coordinator only) --
-  addListener(host: string, port: number, policy: string): Promise<string>;
-  removeListener(id: string): Promise<void>;
-  listListeners(): ListenerInfo[];
-  getNetworkInterfaces(): NetworkInterface[];
-
-  // -- Federation (coordinator-to-coordinator) --
-  fedConnect(host: string, port: number, name?: string): Promise<string>;
-  fedDisconnect(linkId: string): Promise<void>;
-  fedLinks(): FedLink[];
-  /** This instance's own federation TLS fingerprint, to hand to an operator on the other side to pin. */
-  getFederationFingerprint(): string;
-  /** Pin a remote mesh's certificate fingerprint as trusted for federation, inbound or outbound. */
-  fedTrust(fingerprint: string): Promise<void>;
-  /** Remove a previously pinned federation fingerprint. */
-  fedUntrust(fingerprint: string): Promise<void>;
-  /** List currently trusted federation fingerprints. */
-  fedTrustedFingerprints(): string[];
-  /** Start accepting inbound federation links on host:port. Rejects any connection whose certificate isn't pinned via fedTrust(). */
-  fedListen(host: string, port: number): Promise<void>;
-  /** Stop accepting new inbound federation connections. Existing links are unaffected. */
-  fedStopListening(): Promise<void>;
-  // -- Connection approval --
-  acceptConnection(connectionId: string): Promise<void>;
-  rejectConnection(connectionId: string, reason: string): Promise<void>;
-  listPendingConnections(): {
-    connectionId: string;
-    peerId: string;
-    dataPort: number;
-    name: string;
-    fingerprint: string;
-  }[];
-  connectToRemote(host: string, port: number): Promise<void>;
-
-  /** Start only the data server without connecting to a coordinator. */
-  startDataServerOnly(): Promise<void>;
-
   // -- Lifecycle --
   init(): Promise<void>;
 }

src/core/mesh-store.ts

@@ -7,14 +7,16 @@
  * Delivery events are pushed directly over the transport — no polling,
  * no filesystem.
  *
- * Transport is injected via the constructor. TcpTransport for localhost
- * TCP, TlsTransport for encrypted connections, etc.
+ * Transport is set via setTransport() (e.g. TlsTransport 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.
  */
 
 import * as os from "node:os";
 import { nanoid } from "./nanoid.js";
 import { CommsError } from "./store.js";
-import { TcpTransport } from "./tcp-transport.js";
 import { dmKey, normaliseWireState } from "./wire-protocol.js";
 import type { SerialisedState } from "./wire-protocol.js";
 import { DiscoveryManager } from "./discovery.js";
@@ -99,7 +101,7 @@ export class MeshStore implements CommsStore {
   private deliveryQueues = new Map<string, DeliveryEvent[]>();
   private identityCache = new Map<string, { id: string }>();
 
-  private transport: MeshTransport;
+  private transport: MeshTransport | undefined;
   private peerInfo = new Map<string, PeerInfo>();
   private staleCheckTimer: ReturnType<typeof setInterval> | undefined;
   private isShutDown = false;
@@ -109,7 +111,8 @@ export class MeshStore implements CommsStore {
   /** Whether the mesh has a live coordinator connection. */
   get connected(): boolean {
     return (
-      this.transport.isCoordinator || this.transport.hasCoordinatorConnection
+      this.requireTransport().isCoordinator ||
+      this.requireTransport().hasCoordinatorConnection
     );
   }
 
@@ -126,6 +129,11 @@ export class MeshStore implements CommsStore {
   /** Fires for every state patch — both locally generated and remote. */
   onPatch: ((patch: MeshStatePatch) => void | Promise<void>) | undefined;
 
+  /**
+   * Fires for a transport-level error a caller may want to observe (a connection rejected for presenting a certificate that doesn't match its claimed peer ID, this store's own inability to join or create a mesh, etc). Previously wired to `this.events.onError` inside this class's own methods, but the `events` getter never actually implemented `onError` on the object it returns, so every one of those calls was silently a no-op -- this store had no way to observe its own transport failures at all. Left undefined by default (matching onDelivery/onPatch): a caller that wants visibility sets it, exactly like those two.
+   */
+  onError: ((error: Error) => void) | undefined;
+
   /** Serialise the full mesh state for state_sync messages. */
   serialise(): SerialisedState {
     return {
@@ -147,9 +155,6 @@ export class MeshStore implements CommsStore {
     this.peerId = nanoid(8);
     this.startedAt = new Date().toISOString();
     this.coordinatorPort = coordinatorPort;
-    // Wire up transport with this store's event handlers.
-    // TcpTransport is the default; callers can replace via setTransport().
-    this.transport = new TcpTransport(this.events);
 
     // Discovery manager — registers available backends
     this.discovery = new DiscoveryManager();
@@ -176,24 +181,34 @@ export class MeshStore implements CommsStore {
     );
   }
 
-  /** Replace the transport (e.g. with TlsTransport for encrypted connections). */
+  /** Sets the transport (e.g. TlsTransport for encrypted connections). Must be called before init() or any other transport-using method. */
   setTransport(transport: MeshTransport): void {
     this.transport = transport;
   }
 
+  /** The set transport, or throws if setTransport() hasn't been called yet -- the single place every transport-using method reads through, so the "must call setTransport() first" contract is enforced at one boundary rather than checked ad hoc at each call site. */
+  private requireTransport(): MeshTransport {
+    if (this.transport === undefined) {
+      throw new Error(
+        "MeshStore: no transport set; call setTransport() before using the store",
+      );
+    }
+    return this.transport;
+  }
+
   // -----------------------------------------------------------------------
   // Mesh lifecycle
   // -----------------------------------------------------------------------
 
   async init(): Promise<void> {
     if (this.initialised) return;
     this.initialised = true;
-    await this.transport.startDataServer();
+    await this.requireTransport().startDataServer();
 
     // Register our own peer info
     this.peerInfo.set(this.peerId, {
       id: this.peerId,
-      port: this.transport.dataPort,
+      port: this.requireTransport().dataPort,
       startedAt: this.startedAt,
     });
 
@@ -208,16 +223,16 @@ export class MeshStore implements CommsStore {
     let connected = false;
 
     try {
-      await this.transport.connectToCoordinator(
+      await this.requireTransport().connectToCoordinator(
         COORDINATOR_HOST,
         this.coordinatorPort,
         this.peerId,
-        this.transport.dataPort,
+        this.requireTransport().dataPort,
       );
       connected = true;
     } catch {
       try {
-        await this.transport.becomeCoordinator(
+        await this.requireTransport().becomeCoordinator(
           COORDINATOR_HOST,
           this.coordinatorPort,
         );
@@ -243,7 +258,7 @@ export class MeshStore implements CommsStore {
       return;
     }
 
-    this.transport.unref();
+    this.requireTransport().unref();
   }
 
   // -----------------------------------------------------------------------
@@ -253,13 +268,16 @@ export class MeshStore implements CommsStore {
   private handlePeerList(peers: PeerInfo[]): void {
     for (const peer of peers) {
       this.peerInfo.set(peer.id, peer);
-      void this.transport.connectToPeer(peer, this.peerId);
+      // The list always includes this store's own entry — dialling yourself is a wasted connection attempt (and, on some platforms, an immediate self-inflicted ECONNRESET) that never needs to happen.
+      if (peer.id === this.peerId) continue;
+      void this.requireTransport().connectToPeer(peer, this.peerId);
     }
   }
 
   private handlePeerJoined(peer: PeerInfo): void {
     this.peerInfo.set(peer.id, peer);
-    void this.transport.connectToPeer(peer, this.peerId);
+    if (peer.id === this.peerId) return;
+    void this.requireTransport().connectToPeer(peer, this.peerId);
   }
 
   private async handleIntroduction(
@@ -278,14 +296,14 @@ export class MeshStore implements CommsStore {
       method: "peer_list",
       peers: [...this.peerInfo.values()],
     };
-    await this.transport.send(handle, peerList);
+    await this.requireTransport().send(handle, peerList);
 
     // Broadcast arrival to all existing peers
     const joined: MeshMessage = { method: "peer_joined", peer: newPeer };
-    await this.transport.broadcast(joined);
+    await this.requireTransport().broadcast(joined);
 
     // Connect to the new peer's data server
-    void this.transport.connectToPeer(newPeer, this.peerId);
+    void this.requireTransport().connectToPeer(newPeer, this.peerId);
   }
 
   private async handlePeerConnected(
@@ -295,7 +313,7 @@ export class MeshStore implements CommsStore {
     // If we have state and the peer doesn't, send state sync
     if (this.agents.size > 0) {
       const state: SerialisedState = this.serialise();
-      await this.transport.send(handle, {
+      await this.requireTransport().send(handle, {
         method: "state_sync",
         state,
       });
@@ -402,14 +420,14 @@ export class MeshStore implements CommsStore {
 
   private async handleBecomeCoordinator(peerList: PeerInfo[]): Promise<void> {
     // Take over as coordinator using the data server we already have
-    await this.transport.becomeCoordinator(
+    await this.requireTransport().becomeCoordinator(
       COORDINATOR_HOST,
       this.coordinatorPort,
     );
     this.peerInfo.clear();
     for (const peer of peerList) {
       this.peerInfo.set(peer.id, peer);
-      void this.transport.connectToPeer(peer, this.peerId);
+      void this.requireTransport().connectToPeer(peer, this.peerId);
     }
     this.startStaleCheck();
   }
@@ -462,7 +480,7 @@ export class MeshStore implements CommsStore {
     }
     this.pendingInboundConnections.delete(connectionId);
     const handle: ConnectionHandle = { id: connectionId };
-    await this.transport.acceptConnection(handle);
+    await this.requireTransport().acceptConnection(handle);
   }
 
   /** Reject a pending inbound connection. */
@@ -473,7 +491,7 @@ export class MeshStore implements CommsStore {
     }
     this.pendingInboundConnections.delete(connectionId);
     const handle: ConnectionHandle = { id: connectionId };
-    await this.transport.rejectConnection(handle, reason);
+    await this.requireTransport().rejectConnection(handle, reason);
   }
 
   /** List all pending inbound connections awaiting approval. */
@@ -498,12 +516,12 @@ export class MeshStore implements CommsStore {
     // Fire-and-forget: don't await the full approval handshake.
     // The coordinator will either accept (triggering normal introduction flow)
     // or reject (closing the socket). Handle rejection to avoid unhandled rejection.
-    this.transport
+    this.requireTransport()
       .connectToRemote(
         host,
         port,
         this.peerId,
-        this.transport.dataPort,
+        this.requireTransport().dataPort,
         agent?.name ?? "",
         "",
       )
@@ -518,13 +536,13 @@ export class MeshStore implements CommsStore {
   /** Start only the data server without connecting to a coordinator.
    *  Used for testing scenarios where the peer connects via connectToRemote. */
   async startDataServerOnly(): Promise<void> {
-    await this.transport.startDataServer();
+    await this.requireTransport().startDataServer();
     this.peerInfo.set(this.peerId, {
       id: this.peerId,
-      port: this.transport.dataPort,
+      port: this.requireTransport().dataPort,
       startedAt: this.startedAt,
     });
-    this.transport.unref();
+    this.requireTransport().unref();
   }
 
   // -----------------------------------------------------------------------
@@ -558,6 +576,9 @@ export class MeshStore implements CommsStore {
       onConnectionRequest: (handle, request) => {
         this.handleConnectionRequest(handle, request);
       },
+      onError: (error) => {
+        this.onError?.(error);
+      },
     };
   }
 
@@ -766,7 +787,7 @@ export class MeshStore implements CommsStore {
   // -----------------------------------------------------------------------
 
   private async broadcastPatch(patch: MeshStatePatch): Promise<void> {
-    await this.transport.broadcast({ method: "state_update", patch });
+    await this.requireTransport().broadcast({ method: "state_update", patch });
     if (this.onPatch) {
       await this.onPatch(patch);
     }
@@ -1601,15 +1622,15 @@ export class MeshStore implements CommsStore {
     if (!isListenerPolicy(policy)) {
       throw new CommsError(`Invalid policy "${policy}"`, "INVALID_POLICY");
     }
-    return this.transport.addListener(host, port, policy);
+    return this.requireTransport().addListener(host, port, policy);
   }
 
   async removeListener(id: string): Promise<void> {
-    return this.transport.removeListener(id);
+    return this.requireTransport().removeListener(id);
   }
 
   listListeners(): ListenerInfo[] {
-    return this.transport.listListeners();
+    return this.requireTransport().listListeners();
   }
 
   getNetworkInterfaces(): NetworkInterface[] {
@@ -1822,6 +1843,6 @@ export class MeshStore implements CommsStore {
 
     this.stopStaleCheck();
     await this.federation.shutdown();
-    await this.transport.shutdown();
+    await this.requireTransport().shutdown();
   }
 }