From 8b23de21e29b2416591a814c01555948dc16edeb Mon Sep 17 00:00:00 2001 From: Joseph Mearman Date: Wed, 16 Sep 2026 04:33:17 +0100 Subject: [PATCH] feat(core): gossip this side's own agent identity and merge discoveries into listAgents WireMeshTransport's gossip tick gains a third optional source (getSelfAgentAdvert), gossiping this side's own name/harness/cwd/pid/ startedAt/tags/subscribedRooms under agent/self -- the write half of P3.8's eventual agent register/update/offline retirement. Deliberately excludes status (already carried separately under presence/status) and visibility (only ever populated for a "visible" agent in the first place, so a discovered entry's visibility is always exactly "visible" by construction). MeshStore.selfAgentAdvert reads this store's own agent record and withholds the advert entirely for anything other than "visible" -- gossip reaches every connected peer regardless of mesh-approval status, so advertising a hidden or ghost agent's identity this way would leak exactly what those visibility levels exist to withhold. AgentRegistry.listAgents merges in a gossip-discovered agent not otherwise locally known, mirroring listRooms' own room-discovery merge (#138): a display-only placeholder AgentIdentity, real facts from the gossiped advert, status read from the existing presence/ status key, never merged into the local agents map, and always shadowed by a real local record when one exists. --- src/core/agent-registry.ts | 63 ++++++++ src/core/bridge-mesh.ts | 1 + src/core/mesh-store.ts | 21 ++- src/core/wire-mesh-transport.ts | 29 +++- src/test/agent-registry.test.ts | 86 +++++++++++ .../agent-self-gossip.integration.test.ts | 141 ++++++++++++++++++ 6 files changed, 339 insertions(+), 2 deletions(-) create mode 100644 src/test/agent-self-gossip.integration.test.ts diff --git a/src/core/agent-registry.ts b/src/core/agent-registry.ts index 41978152..a24635ab 100644 --- a/src/core/agent-registry.ts +++ b/src/core/agent-registry.ts @@ -5,6 +5,9 @@ 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. */ @@ -13,6 +16,7 @@ export interface AgentRegistryDeps { identityCache: Map; startedAt: string; getPeerId: () => string; + requireTransport: () => MeshTransport; deliveryEngine: Pick< DeliveryEngine, | "bump" @@ -26,6 +30,21 @@ export interface AgentRegistryDeps { >; } +/** 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. */ +function isAgentSelfAdvert(value: unknown): value is AgentSelfAdvert { + if (typeof value !== "object" || value === null) return false; + if (!("name" in value) || typeof value.name !== "string") return false; + if (!("harness" in value) || typeof value.harness !== "string") return false; + if (!("cwd" in value) || typeof value.cwd !== "string") return false; + if (!("pid" in value) || typeof value.pid !== "number") return false; + if (!("startedAt" in value) || typeof value.startedAt !== "string") + return false; + if (!("tags" in value) || !Array.isArray(value.tags)) return false; + if (!("subscribedRooms" in value) || !Array.isArray(value.subscribedRooms)) + return false; + return true; +} + export class AgentRegistry { constructor(private readonly deps: AgentRegistryDeps) {} @@ -144,6 +163,50 @@ export class AgentRegistry { if (agent.visibility === "ghost" && agent.id !== requesterId) continue; result.push(agent); } + for (const discovered of this.listDiscoverableAgents()) { + if (this.deps.agents.has(discovered.deviceId)) continue; + result.push({ + id: discovered.deviceId, + version: 0, + name: discovered.advert.name, + harness: discovered.advert.harness, + cwd: discovered.advert.cwd, + pid: discovered.advert.pid, + startedAt: discovered.advert.startedAt, + visibility: "visible", + status: discovered.status ?? "active", + tags: discovered.advert.tags, + subscribedRooms: discovered.advert.subscribedRooms, + }); + } + return result; + } + + /** + * Every agent this store has heard gossiped by another device but never registered or otherwise locally recorded -- the read half of P3.8's eventual agent register/update/offline retirement (agent-comms#48), mirroring listRooms' own room-discovery merge (#138). Never merged into this.deps.agents: a gossip hint is not the same as a real registration, and this store has nothing else authoritative to report for it. Only ever an agent that gossiped itself as "visible" (MeshStore's own selfAgentAdvert getter never advertises a hidden or ghost agent this way), so no ghost-filtering is needed here the way listAgents' own local-agent check needs. + */ + private listDiscoverableAgents(): readonly { + deviceId: string; + advert: AgentSelfAdvert; + status: AgentStatus | undefined; + }[] { + const transport = this.deps.requireTransport(); + if (transport.listKnownDevices === undefined) return []; + const result: { + deviceId: string; + advert: AgentSelfAdvert; + status: AgentStatus | undefined; + }[] = []; + for (const { deviceId, advert } of transport.listKnownDevices()) { + const candidate = advert["agent/self"]; + if (!isAgentSelfAdvert(candidate)) continue; + const status = advert["presence/status"]; + result.push({ + deviceId, + advert: candidate, + status: AgentStatus.is(status) ? status : undefined, + }); + } return result; } diff --git a/src/core/bridge-mesh.ts b/src/core/bridge-mesh.ts index 27bf8b70..f336a847 100644 --- a/src/core/bridge-mesh.ts +++ b/src/core/bridge-mesh.ts @@ -47,6 +47,7 @@ export function createBridgeMeshSync( undefined, () => store.hostedRooms, dataStorage, + () => store.selfAgentAdvert, ), ); const tool = new CommsTool(store, store.discovery); diff --git a/src/core/mesh-store.ts b/src/core/mesh-store.ts index 0b30db84..dc157150 100644 --- a/src/core/mesh-store.ts +++ b/src/core/mesh-store.ts @@ -29,7 +29,10 @@ import { ConnectionApproval } from "./connection-approval.js"; import { StaleAgentChecker } from "./stale-agent-checker.js"; import { PeerLifecycle } from "./peer-lifecycle.js"; import type { RoomVerbHandler } from "./room-router.js"; -import type { HostedRoomAdvert } from "./wire-mesh-transport.js"; +import type { + AgentSelfAdvert, + HostedRoomAdvert, +} from "./wire-mesh-transport.js"; import type { MeshStatePatch, PeerInfo, @@ -124,6 +127,21 @@ export class MeshStore implements CommsStore { return result; } + /** This store's own gossip-safe agent-identity advert, synchronously, in the shape WireMeshTransport's own gossip re-advertisement timer reads on every tick -- the write half of P3.8's eventual agent register/update/offline retirement (agent-comms#48). undefined before registerAgent has ever run (nothing to advertise yet), or when this agent's own visibility isn't "visible" -- gossip already reaches every connected peer regardless of mesh-approval status (see wire-mesh-transport.ts's own allSessions/quarantine comments), so advertising a hidden or ghost agent's identity this way would leak exactly what those visibility levels exist to withhold. */ + get selfAgentAdvert(): AgentSelfAdvert | undefined { + const agent = this.agents.get(this.peerId); + if (agent?.visibility !== "visible") return undefined; + return { + name: agent.name, + harness: agent.harness, + cwd: agent.cwd, + pid: agent.pid, + startedAt: agent.startedAt, + tags: agent.tags, + subscribedRooms: agent.subscribedRooms, + }; + } + /** Whether the mesh has a live coordinator connection. */ get connected(): boolean { return ( @@ -245,6 +263,7 @@ export class MeshStore implements CommsStore { identityCache: this.identityCache, startedAt: this.startedAt, getPeerId: () => this.peerId, + requireTransport: () => this.requireTransport(), deliveryEngine: this.deliveryEngine, federation: this.federation, }); diff --git a/src/core/wire-mesh-transport.ts b/src/core/wire-mesh-transport.ts index 82dae42e..56991acb 100644 --- a/src/core/wire-mesh-transport.ts +++ b/src/core/wire-mesh-transport.ts @@ -92,6 +92,20 @@ export interface HostedRoomAdvert { description: string; } +/** The domain-qualified gossip extension key this transport writes this side's own agent identity facts under -- the write half of P3.8's eventual agent register/update/offline retirement (agent-comms#48). Same namespacing convention as PRESENCE_GOSSIP_KEY/HOSTED_ROOMS_GOSSIP_KEY. */ +const AGENT_SELF_GOSSIP_KEY = "agent/self"; + +/** The lightweight, gossip-safe shape an agent advertises itself under: enough for a peer with no prior local record of this device to construct a real AgentIdentity-shaped discovery entry. Deliberately excludes status (already carried separately under presence/status, no need to duplicate it here) and visibility (this field is only ever populated for a "visible" agent in the first place -- see MeshStore's own selfAgentAdvert getter -- so a discovered entry's visibility is always exactly "visible" by construction, never something this advert needs to assert itself). */ +export interface AgentSelfAdvert { + name: string; + harness: string; + cwd: string; + pid: number; + startedAt: string; + tags: string[]; + subscribedRooms: string[]; +} + /** Upper bound on the number of oplog entries handleDataRequest returns in a single data-entries response -- generous for the small, chat-sized messages this domain carries today, while still bounding one peer's worst-case memory/frame size when answering a request for a large catch-up gap. A requester short of this still gets everything up to its own current head; anything beyond it needs a follow-up data-request, exactly the same incremental-catch-up shape a data-have/data-request/data-entries cycle already has. */ const DATA_ENTRIES_RESPONSE_LIMIT = 100; @@ -241,6 +255,10 @@ export class WireMeshTransport implements MeshTransport { /** Backs this side's own responder for an incoming data-have/data-request/data-entries frame (agent-comms#50's P5 integration) -- undefined for every existing construction site that predates this feature, in which case handleDataFrame is a no-op. Deciding when to proactively call sendDataFrame at all (the catch-up policy: which peers' logs to track, when to send an initial data-have) stays entirely the caller's own business; this field only ever backs the mechanical parts (answering a have/request, storing entries). */ private readonly dataStorage: KeyValueStorage | undefined; + /** Reads this side's own gossip-safe agent-identity advert for the next gossip re-advertisement tick, the same pull-not-push shape getCurrentPresence/getHostedRooms already established. undefined when no agent-identity source was wired in, or when MeshStore's own getter decides this agent shouldn't advertise itself this way right now (e.g. not "visible", or no self-agent record yet). */ + private readonly getSelfAgentAdvert: + (() => AgentSelfAdvert | undefined) | undefined; + /** Every peer this side has ever received a frame from, keyed by device-id hex, tracking the raw wire-mesh-core Connection each frame arrived on -- what sendDataFrame needs, since neither AcceptedMeshSession nor MeshSession exposes a generic "send an arbitrary frame" method the way the raw Connection itself does. Registered eagerly on the very first frame from a connection (including one still in quarantine, e.g. before connect_request approval) so a later sendDataFrame call can reach it -- handleDataFrame's own trust gate (peerSessions.has) is what actually decides whether to act on anything received this way, not this map. */ private readonly connectionsByPeer = new Map(); @@ -253,6 +271,7 @@ export class WireMeshTransport implements MeshTransport { presenceReadvertiseIntervalMs: number = PRESENCE_READVERTISE_INTERVAL_MS, getHostedRooms?: () => readonly HostedRoomAdvert[], dataStorage?: KeyValueStorage, + getSelfAgentAdvert?: () => AgentSelfAdvert | undefined, ) { this.events = events; this.wireTransport = createTlsTransport({ @@ -268,7 +287,12 @@ export class WireMeshTransport implements MeshTransport { this.getCurrentPresence = getCurrentPresence; this.getHostedRooms = getHostedRooms; this.dataStorage = dataStorage; - if (getCurrentPresence !== undefined || getHostedRooms !== undefined) { + this.getSelfAgentAdvert = getSelfAgentAdvert; + if ( + getCurrentPresence !== undefined || + getHostedRooms !== undefined || + getSelfAgentAdvert !== undefined + ) { this.gossipInterval = setInterval(() => { this.readvertiseGossip(); }, presenceReadvertiseIntervalMs); @@ -330,6 +354,9 @@ export class WireMeshTransport implements MeshTransport { const hostedRooms = this.getHostedRooms?.(); if (hostedRooms !== undefined) extensions[HOSTED_ROOMS_GOSSIP_KEY] = hostedRooms; + const selfAgentAdvert = this.getSelfAgentAdvert?.(); + if (selfAgentAdvert !== undefined) + extensions[AGENT_SELF_GOSSIP_KEY] = selfAgentAdvert; if (Object.keys(extensions).length === 0) return; for (const session of this.allSessions) { session.sendGossipUpdate(extensions).catch((error: unknown) => { diff --git a/src/test/agent-registry.test.ts b/src/test/agent-registry.test.ts index 17c7af89..0823f41d 100644 --- a/src/test/agent-registry.test.ts +++ b/src/test/agent-registry.test.ts @@ -57,6 +57,10 @@ function makeHarness(peerId = OWNER_ID): Harness { identityCache: new Map(), startedAt: STARTED_AT, getPeerId: () => peerId, + requireTransport: () => + ({ listKnownDevices: undefined }) as unknown as ReturnType< + AgentRegistryDeps["requireTransport"] + >, deliveryEngine: { bump, broadcastPatch, @@ -301,6 +305,88 @@ describe("AgentRegistry — listAgents", () => { const result = await registry.listAgents("anyone-else"); expect(result.map((a) => a.id)).toEqual(["visible-agent"]); }); + + it("merges in a gossip-discovered agent not otherwise locally known, as a placeholder-shaped AgentIdentity", async () => { + const { registry, deps } = makeHarness(); + deps.requireTransport = () => + ({ + listKnownDevices: () => [ + { + deviceId: "discovered-device", + advert: { + "agent/self": { + name: "discovered-agent", + harness: "codex", + cwd: "/tmp/discovered", + pid: 7, + startedAt: "2026-02-02T00:00:00.000Z", + tags: ["from-gossip"], + subscribedRooms: [], + }, + "presence/status": "busy", + }, + }, + ], + }) as unknown as ReturnType; + + const result = await registry.listAgents("some-requester"); + const discovered = result.find((a) => a.id === "discovered-device"); + expect(discovered).toMatchObject({ + id: "discovered-device", + name: "discovered-agent", + harness: "codex", + cwd: "/tmp/discovered", + pid: 7, + startedAt: "2026-02-02T00:00:00.000Z", + visibility: "visible", + status: "busy", + tags: ["from-gossip"], + subscribedRooms: [], + }); + }); + + it("never lets a gossip-discovered agent shadow an agent this store already knows locally", async () => { + const { registry, deps } = makeHarness(); + deps.agents.set( + "already-known", + agent({ id: "already-known", name: "real-agent" }), + ); + deps.requireTransport = () => + ({ + listKnownDevices: () => [ + { + deviceId: "already-known", + advert: { + "agent/self": { + name: "stale-gossip-copy", + harness: "codex", + cwd: "/tmp", + pid: 1, + startedAt: "2026-01-01T00:00:00.000Z", + tags: [], + subscribedRooms: [], + }, + }, + }, + ], + }) as unknown as ReturnType; + + const result = await registry.listAgents("some-requester"); + expect(result.filter((a) => a.id === "already-known")).toHaveLength(1); + expect(result.find((a) => a.id === "already-known")?.name).toBe( + "real-agent", + ); + }); + + it("ignores a transport with no listKnownDevices capability, matching every construction site that predates this feature", async () => { + const { registry, deps } = makeHarness(); + deps.agents.set( + "visible-agent", + agent({ id: "visible-agent", visibility: "visible" }), + ); + const result = await registry.listAgents("anyone-else"); + expect(result.map((a) => a.id)).toEqual(["visible-agent"]); + }); }); describe("AgentRegistry — setAgentOffline", () => { diff --git a/src/test/agent-self-gossip.integration.test.ts b/src/test/agent-self-gossip.integration.test.ts new file mode 100644 index 00000000..8cb0d1a7 --- /dev/null +++ b/src/test/agent-self-gossip.integration.test.ts @@ -0,0 +1,141 @@ +/** + * WireMeshTransport's periodic gossip tick also carries this side's own agent-identity self-advert (a `agent/self` extension on peer-advert's own open tail, the same convention presence/status and room/hosted already established), so a peer's `listKnownDevices()` can read another device's agent facts directly. This is the write half of P3.8's eventual agent register/update/offline retirement (agent-comms#48): the read side (merging a gossip-discovered agent into listAgents) is deliberately not built here -- it needs its own follow-up, per the room-discovery precedent (#131/#138) this mirrors. + */ + +import { test, describe, expect } from "vitest"; +import { generateIdentity } from "../core/identity.js"; +import { toIdentityPort } from "../core/wire-mesh-identity.js"; +import { deviceIdToHex } from "wire-mesh-core/domain/device-id"; +import { + WireMeshTransport, + type AgentSelfAdvert, +} from "../core/wire-mesh-transport.js"; +import type { TransportEvents } from "../core/transport.js"; +import { waitFor } from "./test-transport.js"; + +const SHORT_INTERVAL_MS = 50; + +function inertEvents(): TransportEvents { + return { + onMessage: () => undefined, + onPeerConnected: () => undefined, + onPeerDisconnected: () => undefined, + onIntroduction: () => undefined, + onConnectionRequest: () => undefined, + onPeerList: () => undefined, + onPeerJoined: () => undefined, + onBecomeCoordinator: () => undefined, + onRevocationAnnounce: () => undefined, + onPresenceAdvert: () => undefined, + }; +} + +describe("WireMeshTransport agent-self gossip", () => { + test("a peer's listKnownDevices reflects another device's advertised agent identity", async () => { + const identityA = generateIdentity(); + const identityB = generateIdentity(); + const peerIdA = deviceIdToHex( + await toIdentityPort(identityA).then((p) => p.deviceId), + ); + const peerIdB = deviceIdToHex( + await toIdentityPort(identityB).then((p) => p.deviceId), + ); + + const selfAdvert: AgentSelfAdvert = { + name: "agent-a", + harness: "pi", + cwd: "/tmp/a", + pid: 4242, + startedAt: "2026-01-01T00:00:00.000Z", + tags: ["from-a"], + subscribedRooms: [], + }; + + const transportA = new WireMeshTransport( + inertEvents(), + identityA, + undefined, + undefined, + undefined, + SHORT_INTERVAL_MS, + undefined, + undefined, + () => selfAdvert, + ); + const transportB = new WireMeshTransport(inertEvents(), identityB); + + try { + await transportA.startDataServer(); + await transportB.connectToPeer( + { + id: peerIdA, + port: transportA.dataPort, + startedAt: new Date().toISOString(), + }, + peerIdB, + ); + + await waitFor(() => { + const advert = transportB + .listKnownDevices() + .find((entry) => entry.deviceId === peerIdA)?.advert["agent/self"]; + return ( + typeof advert === "object" && + advert !== null && + "name" in advert && + advert.name === "agent-a" + ); + }, "B observes A's advertised agent identity"); + + const known = transportB + .listKnownDevices() + .find((entry) => entry.deviceId === peerIdA); + expect(known?.advert["agent/self"]).toEqual(selfAdvert); + } finally { + await transportB.shutdown(); + await transportA.shutdown(); + } + }); + + test("a session with no agent-self source configured never advertises the agent/self key", async () => { + const identityA = generateIdentity(); + const identityB = generateIdentity(); + const peerIdA = deviceIdToHex( + await toIdentityPort(identityA).then((p) => p.deviceId), + ); + const peerIdB = deviceIdToHex( + await toIdentityPort(identityB).then((p) => p.deviceId), + ); + + const transportA = new WireMeshTransport(inertEvents(), identityA); + const transportB = new WireMeshTransport(inertEvents(), identityB); + + try { + await transportA.startDataServer(); + await transportB.connectToPeer( + { + id: peerIdA, + port: transportA.dataPort, + startedAt: new Date().toISOString(), + }, + peerIdB, + ); + + await waitFor( + () => + transportB + .listKnownDevices() + .some((entry) => entry.deviceId === peerIdA), + "B's known-devices view includes A", + ); + + const known = transportB + .listKnownDevices() + .find((entry) => entry.deviceId === peerIdA); + expect(known?.advert["agent/self"]).toBeUndefined(); + } finally { + await transportB.shutdown(); + await transportA.shutdown(); + } + }); +});