phase 1.0b: lightweight read-only federation (one-hop)

Zawwarsami16 · claude · Zawwarsami16 · commit dadd7a192a33 · 2026-05-09T03:42:22.000Z
a hub can be configured with peer hub URLs. /registry/global aggregates
local + peer listings, annotated with origin. one-hop discovery only —
no cross-hub call routing, no shared state, no signed peer relationships
yet (those are 1.1+).

components:
- zhub/federation.py: PeerRegistry — async httpx-backed cache of peer
  /registry endpoints with refresh interval. concurrent fetch via
  asyncio.gather so slow peers don't block fast ones. failures become
  empty lists silently — never block the local response.
- zhub/server.py: ZHUB_PEERS env (or new --peers CLI flag, comma-
  separated URLs) configures peer set. new GET /registry/global
  endpoint serves local + peer entries (each peer entry stamped with
  its origin URL). local entries get origin: "self".

tests (TDD — RED first, then GREEN):
- tests/test_federation.py: 3 cases
  - test_global_registry_aggregates_peer: hub A peers hub B, B
    publishes 'onb', A's /registry/global includes 'onb' with non-self
    origin
  - test_global_registry_without_peers_returns_only_local: peerless hub
    serves only its own publishers, all origin: 'self'
  - test_global_registry_offline_peer_omitted_gracefully: bogus peer
    URL doesn't 502 the local response — silently skipped

result: 32/32 pytest passing across all phases (signing, federation,
council, persistence, streaming, e2e, manifest, protocol, invoke shape,
generator non-streaming).

Co-Authored-By: Claude Opus 4.7 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/tests/test_federation.py b/tests/test_federation.py
@@ -0,0 +1,137 @@
+"""Two hubs in-process; one peers the other; /registry/global aggregates."""
+
+import asyncio
+import os
+import socket
+import threading
+import time
+
+import pytest
+
+try:
+    import fastapi  # noqa
+    import uvicorn  # noqa
+    import httpx  # noqa
+    DEPS_AVAILABLE = True
+except ImportError:
+    DEPS_AVAILABLE = False
+
+if DEPS_AVAILABLE:
+    from zhub.server import create_app
+from zhub import publish
+
+
+def _free_port() -> int:
+    with socket.socket() as s:
+        s.bind(("", 0))
+        return s.getsockname()[1]
+
+
+def _start_hub(port: int, peers_env: str = "") -> None:
+    """Start a hub in this thread. Reads peers from process env at start."""
+    if peers_env:
+        os.environ["ZHUB_PEERS"] = peers_env
+    else:
+        os.environ.pop("ZHUB_PEERS", None)
+    config = uvicorn.Config(create_app(), host="127.0.0.1", port=port, log_level="warning")
+    asyncio.run(uvicorn.Server(config).serve())
+
+
+def _wait(port: int) -> None:
+    for _ in range(30):
+        try:
+            with socket.create_connection(("127.0.0.1", port), timeout=0.1):
+                return
+        except OSError:
+            time.sleep(0.1)
+
+
+@pytest.mark.asyncio
+async def test_global_registry_aggregates_peer():
+    """Hub A peers hub B. A publisher registered on B is visible via A's
+    /registry/global endpoint, annotated with origin."""
+    if not DEPS_AVAILABLE:
+        pytest.skip("fastapi/uvicorn/httpx not installed")
+    port_a = _free_port()
+    port_b = _free_port()
+
+    # Start B first (no peers), then A peering B.
+    threading.Thread(target=_start_hub, args=(port_b, ""), daemon=True).start()
+    _wait(port_b)
+    threading.Thread(target=_start_hub, args=(port_a, f"http://127.0.0.1:{port_b}"), daemon=True).start()
+    _wait(port_a)
+
+    # Publish on B
+    pub = publish(
+        name="onb",
+        description="lives on B",
+        chat_handler=lambda m, o: "ok",
+        hub_url=f"ws://127.0.0.1:{port_b}",
+        public=True,
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.1)
+    assert pub.api_key
+
+    # Fetch hub A's global registry
+    async with httpx.AsyncClient() as c:
+        resp = await c.get(f"http://127.0.0.1:{port_a}/registry/global")
+    assert resp.status_code == 200
+    data = resp.json()
+
+    names = {e["name"] for e in data}
+    assert "onb" in names, f"expected 'onb' in {names}"
+    origins = {e.get("origin") for e in data if e["name"] == "onb"}
+    assert any(o and o != "self" for o in origins), \
+        f"expected at least one non-self origin, got {origins}"
+
+
+@pytest.mark.asyncio
+async def test_global_registry_without_peers_returns_only_local():
+    """A hub with no peers configured returns only its own local listings
+    from /registry/global (degenerate but correct)."""
+    if not DEPS_AVAILABLE:
+        pytest.skip("fastapi/uvicorn/httpx not installed")
+    port = _free_port()
+    threading.Thread(target=_start_hub, args=(port, ""), daemon=True).start()
+    _wait(port)
+
+    pub = publish(
+        name="solo",
+        description="no peers",
+        chat_handler=lambda m, o: "ok",
+        hub_url=f"ws://127.0.0.1:{port}",
+        public=True,
+    )
+    for _ in range(50):
+        if pub.api_key:
+            break
+        await asyncio.sleep(0.1)
+
+    async with httpx.AsyncClient() as c:
+        resp = await c.get(f"http://127.0.0.1:{port}/registry/global")
+    assert resp.status_code == 200
+    data = resp.json()
+    assert len(data) >= 1
+    assert all(e.get("origin") == "self" for e in data)
+
+
+@pytest.mark.asyncio
+async def test_global_registry_offline_peer_omitted_gracefully():
+    """If a peer is unreachable, /registry/global still returns local
+    listings — the offline peer is silently skipped."""
+    if not DEPS_AVAILABLE:
+        pytest.skip("fastapi/uvicorn/httpx not installed")
+    bogus_port = _free_port()  # nothing listens here
+    port = _free_port()
+    threading.Thread(target=_start_hub,
+                     args=(port, f"http://127.0.0.1:{bogus_port}"),
+                     daemon=True).start()
+    _wait(port)
+
+    async with httpx.AsyncClient() as c:
+        resp = await c.get(f"http://127.0.0.1:{port}/registry/global")
+    assert resp.status_code == 200  # not 502
+    # Body may be empty if no local publishers — that's fine.
diff --git a/zhub/federation.py b/zhub/federation.py
@@ -0,0 +1,84 @@
+"""Lightweight, read-only federation between hubs.
+
+A hub can be configured with a list of peer hub URLs. Periodically, each
+peer's `/registry` is fetched and cached. The aggregator endpoint
+`/registry/global` returns local listings + peer listings, annotated with
+their origin URL.
+
+This is one-hop discovery only:
+  - no cross-hub call routing (clients still go through their hub's own
+    publishers)
+  - no shared registry state
+  - offline peers are skipped silently — never block the local response
+
+Anything richer (cross-hub call routing, signed peer relationships, etc.)
+is Phase 1.1+.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import time
+from typing import Any
+
+try:
+    import httpx
+except ImportError as e:
+    raise SystemExit(
+        "zhub.federation requires httpx. install:\n"
+        "    pip install httpx"
+    ) from e
+
+
+log = logging.getLogger("zhub.federation")
+
+
+class PeerRegistry:
+    """Caches peer hub registries with a refresh interval."""
+
+    def __init__(self, peers: list[str], refresh_seconds: float = 60.0,
+                 timeout_seconds: float = 5.0) -> None:
+        self.peers = peers
+        self.refresh_seconds = refresh_seconds
+        self._cache: dict[str, tuple[list[dict[str, Any]], float]] = {}
+        self._http = httpx.AsyncClient(timeout=timeout_seconds)
+
+    async def get(self, peer_url: str) -> list[dict[str, Any]]:
+        """Return cached peer registry if fresh, otherwise re-fetch.
+        Empty list on any failure — callers must not block on a dead peer."""
+        cached = self._cache.get(peer_url)
+        if cached and time.time() - cached[1] < self.refresh_seconds:
+            return cached[0]
+        try:
+            resp = await self._http.get(peer_url.rstrip("/") + "/registry")
+            resp.raise_for_status()
+            data = resp.json()
+            if not isinstance(data, list):
+                log.warning("peer %s returned non-list registry; treating as empty", peer_url)
+                data = []
+        except Exception as e:
+            log.warning("peer %s unreachable: %s", peer_url, e)
+            return []
+        self._cache[peer_url] = (data, time.time())
+        return data
+
+    async def aggregate(self) -> list[dict[str, Any]]:
+        """Return all peer entries annotated with their origin URL.
+        Concurrent fetch — slow peers don't block fast ones."""
+        if not self.peers:
+            return []
+        results = await asyncio.gather(
+            *(self.get(peer) for peer in self.peers),
+            return_exceptions=False,
+        )
+        out: list[dict[str, Any]] = []
+        for peer, entries in zip(self.peers, results):
+            for e in entries:
+                e2 = dict(e)
+                e2["origin"] = peer
+                out.append(e2)
+        return out
+
+    async def close(self) -> None:
+        await self._http.aclose()
diff --git a/zhub/server.py b/zhub/server.py
@@ -18,6 +18,7 @@
 from __future__ import annotations
 
 import argparse
+import os
 import asyncio
 import json  # noqa: F401  -- used in inline SSE serialization
 import logging
@@ -323,6 +324,33 @@ async def registry() -> JSONResponse:
                 })
         return JSONResponse(listings)
 
+    @app.get("/registry/global")
+    async def registry_global() -> JSONResponse:
+        """Local listings + peer-hub listings, annotated with origin.
+        Peers come from ZHUB_PEERS env var (comma-separated URLs).
+        Offline peers are silently skipped — never block the local response."""
+        local: list[dict[str, Any]] = []
+        for name, p in hub.publishers.items():
+            if p.manifest.get("public"):
+                local.append({
+                    "name": name,
+                    "description": p.manifest.get("description", ""),
+                    "capabilities": [c.get("name") for c in p.manifest.get("capabilities", [])],
+                    "manifest_url": f"/{name}/manifest.json",
+                    "origin": "self",
+                })
+        peers_env = os.environ.get("ZHUB_PEERS", "")
+        peers = [p.strip() for p in peers_env.split(",") if p.strip()]
+        if peers:
+            from .federation import PeerRegistry
+            pr = PeerRegistry(peers)
+            try:
+                peer_entries = await pr.aggregate()
+            finally:
+                await pr.close()
+            return JSONResponse(local + peer_entries)
+        return JSONResponse(local)
+
     @app.get("/{ai_name}/manifest.json")
     async def manifest(ai_name: str) -> JSONResponse:
         publisher = hub.publishers.get(ai_name)
@@ -673,8 +701,16 @@ def main() -> None:
         default="zhub.db",
         help="SQLite path for persistent publisher registry. Pass empty string to disable.",
     )
+    parser.add_argument(
+        "--peers",
+        default="",
+        help="Comma-separated peer hub URLs for read-only federation. "
+             "Peer registries surface at /registry/global with origin annotation.",
+    )
     args = parser.parse_args()
     db_path: Optional[str] = args.db if args.db else None
+    if args.peers:
+        os.environ["ZHUB_PEERS"] = args.peers
 
     logging.basicConfig(
         level=getattr(logging, args.log_level.upper(), logging.INFO),
