feat: wire agent advertisement into p2p capability gossip

Agents with identities now gossip their 0x addresses on the network
via NodeCapability advertisements. The autonet service subscribes to
agent lifecycle events and refreshes p2p capabilities on changes.

Author: EightRice

atn/autonet_service.py

@@ -18,6 +18,7 @@
 import hashlib
 import json
 import logging
+import threading
 from dataclasses import dataclass, field
 from enum import Enum
 from typing import Any, TYPE_CHECKING
@@ -140,6 +141,17 @@ def __init__(self, config: AutonetConfig, event_bus: EventBus | None = None,
                 self._on_execution_completed,
             )
 
+        # P2P agent advertisement
+        self._p2p_host = None      # AutonetHost (lazy, trio-based)
+        self._p2p_thread: threading.Thread | None = None
+        self._p2p_stop = threading.Event()
+        self._agent_registry = None  # Set by Runtime after init
+        # Subscribe to agent lifecycle events for p2p advertisement refresh
+        if self._events:
+            for evt in (EventType.AGENT_REGISTERED, EventType.AGENT_UNREGISTERED,
+                        EventType.AGENT_ACTIVATED, EventType.AGENT_DEACTIVATED):
+                self._events.subscribe(evt, self._on_agent_changed)
+
         # Constitution CID loaded lazily from on-chain Registry
         self._constitution_loaded = False
         # Raw constitution text (loaded once, cached for prompt injection)
@@ -177,6 +189,107 @@ def _discover_jurisdiction(self) -> None:
             log.warning("Failed to discover jurisdiction from %s: %s",
                         self.config.dao_address, e)
 
+    # ------------------------------------------------------------------
+    # P2P agent advertisement
+    # ------------------------------------------------------------------
+
+    def set_agent_registry(self, registry: Any) -> None:
+        """Called by Runtime after init to provide agent registry reference."""
+        self._agent_registry = registry
+
+    async def _on_agent_changed(self, event: Any) -> None:
+        """Refresh p2p capability when agents are registered/unregistered."""
+        self._refresh_p2p_agents()
+
+    def _refresh_p2p_agents(self) -> None:
+        """Rebuild agent advertisements from registry and push to p2p host."""
+        if not self._p2p_host or not self._agent_registry:
+            return
+        try:
+            ads = self._agent_registry.build_agent_advertisements()
+            # Also include the connected wallet as a root agent if present
+            if self.state.wallet_address and not any(
+                a["address"].lower() == self.state.wallet_address.lower() for a in ads
+            ):
+                ads.insert(0, {
+                    "address": self.state.wallet_address,
+                    "name": "root",
+                    "description": "",
+                    "agent_type": "orchestrator",
+                    "model": "",
+                    "is_root": True,
+                    "parent_address": "",
+                    "registered_on_chain": False,
+                })
+            self._p2p_host.update_capability(agents=ads)
+            log.debug("P2P capability updated with %d agent(s)", len(ads))
+        except Exception:
+            log.debug("Failed to refresh p2p agents", exc_info=True)
+
+    def start_p2p(self) -> None:
+        """Start the p2p host in a background thread for agent advertisement."""
+        if self._p2p_thread and self._p2p_thread.is_alive():
+            return
+        try:
+            from nodes.common.p2p import AutonetHost, NodeCapability
+            from nodes.common.config import load_config as load_autonet_config
+        except Exception:
+            log.debug("P2P not available (nodes package not installed or import error)")
+            return
+
+        config_path = self.config.config_path or None
+        try:
+            cfg = load_autonet_config(config_path)
+        except Exception:
+            cfg = None
+
+        listen_port = cfg.p2p.listen_port if cfg else 0
+        listen_host = cfg.p2p.listen_host if cfg else "0.0.0.0"
+        bootstrap = cfg.p2p.bootstrap_peers if cfg else []
+        advertise_interval = cfg.p2p.capability_advertise_interval if cfg else 60
+
+        node_id = f"atn-{self.state.wallet_address[:8]}" if self.state.wallet_address else "atn-daemon"
+        cap = NodeCapability(peer_id="", node_id=node_id)
+
+        host = AutonetHost(
+            node_id=node_id,
+            listen_port=listen_port,
+            listen_host=listen_host,
+            bootstrap_peers=bootstrap,
+            capability=cap,
+        )
+        self._p2p_host = host
+        self._p2p_stop.clear()
+
+        def _run():
+            import trio
+            async def _main():
+                async with host.run():
+                    self._refresh_p2p_agents()
+                    await host.advertise_capability()
+                    log.info("P2P host running, advertising %d agent(s)",
+                             len(host._capability.agents))
+                    while not self._p2p_stop.is_set():
+                        await trio.sleep(advertise_interval)
+                        self._refresh_p2p_agents()
+                        await host.advertise_capability()
+            try:
+                trio.run(_main)
+            except Exception:
+                log.debug("P2P host stopped", exc_info=True)
+
+        self._p2p_thread = threading.Thread(target=_run, name="p2p-host", daemon=True)
+        self._p2p_thread.start()
+        log.info("P2P agent advertisement started")
+
+    def stop_p2p(self) -> None:
+        """Stop the p2p background thread."""
+        self._p2p_stop.set()
+        if self._p2p_thread:
+            self._p2p_thread.join(timeout=5)
+            self._p2p_thread = None
+        self._p2p_host = None
+
     async def _emit(self, event_type_name: str, data: dict[str, Any] | None = None) -> None:
         """Emit an event if the event bus is available."""
         if not self._events:
@@ -295,6 +408,8 @@ async def start(self) -> dict[str, Any]:
 
             self.state.status = AutonetStatus.RUNNING
             log.info("Autonet service started")
+            # Start p2p agent advertisement alongside the training service
+            self.start_p2p()
             await self._emit("AUTONET_STARTED")
             return {"status": "started"}
 
@@ -338,6 +453,7 @@ async def stop(self) -> dict[str, Any]:
                 except asyncio.CancelledError:
                     pass
 
+            self.stop_p2p()
             self.state.status = AutonetStatus.STOPPED
             log.info("Autonet service stopped")
             await self._emit("AUTONET_STOPPED")

atn/runtime/__init__.py

@@ -203,6 +203,8 @@ def __init__(self, event_bus: EventBus, data_dir: Path | None = None, config: AT
         )
         # Give engine access to the autonet bridge for constitutional injection
         self.engine._autonet_bridge = self.autonet
+        # Wire agent registry into autonet bridge for p2p advertisement
+        self.autonet.set_agent_registry(self.registry)
 
         # Trace logger (Phase B Step 1 — Agent Trace Collection)
         from ..trace_logger import TraceLogger, TraceLoggingConfig as _TLConfig
@@ -281,6 +283,8 @@ async def start(self) -> None:
         # Start autonet service if configured
         if self._config.autonet.enabled:
             await self.autonet.start()
+        # Always start p2p agent advertisement (even without training service)
+        self.autonet.start_p2p()
 
         # Load constitution text (needed for prompt injection on registered agents)
         await self.autonet.load_constitution()

atn/runtime/agent_registry.py

@@ -240,6 +240,30 @@ def list_agents(self) -> list[tuple[AgentDefinition, AgentStatus]]:
             for aid, defn in self._agents.items()
         ]
 
+    def build_agent_advertisements(self) -> list[dict]:
+        """Build p2p AgentAdvertisement dicts for all agents with identity."""
+        ads = []
+        for aid, defn in self._agents.items():
+            identity = defn.identity
+            if not identity or not identity.address:
+                continue
+            parent_addr = ""
+            if defn.parent_id:
+                parent = self._agents.get(defn.parent_id)
+                if parent and parent.identity:
+                    parent_addr = parent.identity.address
+            ads.append({
+                "address": identity.address,
+                "name": defn.name,
+                "description": defn.description or "",
+                "agent_type": defn.agent_type or "",
+                "model": defn.model or "",
+                "is_root": defn.parent_id is None or defn.parent_id == "",
+                "parent_address": parent_addr,
+                "registered_on_chain": identity.registered_on_chain,
+            })
+        return ads
+
     def get_agent_key(self, agent_id: str) -> str | None:
         """Get the private key for an agent (parent holds child's key)."""
         return self._agent_keys.get(agent_id)