feat: impl_ref §22 gap-fill — all missing symbols implemented

M12 CLI (9 new commands):
- log (follow mode + snapshot), erase (--keep-keys, --yes)
- rag list/ingest/reindex subgroup
- invite create/redeem subgroup
- version command
- _bus_call helper for bus/v1/call POST

node.py:
- ManifestPublisher class (periodic mDNS+UDP re-publish, triggered on registry change)
- PeriodicTask class (generic async interval runner)
- Removed 2nd duplicate NodeManifest that had leaked back via junk merge

M04 LLM backends (3 new files):
- lmstudio.py — LmStudioBackend wraps OpenAICompatBackend (localhost:1234/v1)
- hf_api.py — HfApiBackend (HF Inference API, HEARTHNET_HF_TOKEN opt-in)
- anthropic_api.py — AnthropicApiBackend (claude-3-haiku/sonnet, ANTHROPIC_API_KEY opt-in)

M01 Identity:
- CommunityPolicy, CommunityMember, RevokedEntry dataclasses added to manifest.py
- Exported from identity/__init__.py

M08 UI:
- hearthnet/ui/theme.py — hearthnet_theme (purple/dark), emergency_theme (red)
- hearthnet/ui/topology.py — TopologyComponent with push_trace/push_topology/render

X01 Transport:
- hearthnet/transport/backpressure.py — FlowControl (semaphore), RateCheck, RateLimiter
- hearthnet/transport/streams.py — Frame dataclass, SseReader async iterator
- hearthnet/transport/__init__.py — exports all 9 X01 symbols

M02 Discovery:
- hearthnet/discovery/__init__.py — DiscoveryError(code) added and exported

M03 Bus:
- hearthnet/bus/registry.py — RegistryEvent(kind, entry) dataclass added
- hearthnet/bus/__init__.py — exports Diff, RegistryEvent

X03 Observability:
- observability/doctor.py — CheckResult = DoctorResult alias (spec name)
- observability/metrics.py — TrackioExporter (optional trackio bridge, 60+ lines)
- observability/tracing.py — detach() function added alongside attach()

M13 Onboarding:
- hearthnet/ui/onboarding.py — build_onboarding alias for build_onboarding_ui

Total: 133 passed, 21 skipped, 0 failed

hearthnet/bus/__init__.py

@@ -13,7 +13,7 @@ from hearthnet.bus.capability import (
     RouteRequest,
 )
 from hearthnet.bus.health import HealthTracker
-from hearthnet.bus.registry import Registry
+from hearthnet.bus.registry import Diff, RegistryEvent, Registry
 from hearthnet.bus.router import BusConfig, Router
 from hearthnet.types import CapabilityName, HearthNetError, Version
 

hearthnet/bus/registry.py

@@ -15,6 +15,16 @@ class Diff:
     updated: list[CapabilityEntry]
 
 
+@dataclass(frozen=True)
+class RegistryEvent:
+    """Emitted by Registry when capabilities change (M03 §3.3).
+
+    kind in {"added", "removed", "updated"}
+    """
+    kind: str
+    entry: CapabilityEntry
+
+
 class Registry:
     def __init__(self, our_node_id: str) -> None:
         self.our_node_id = our_node_id

hearthnet/cli.py

@@ -396,3 +396,254 @@ def export(out: str | None) -> None:
     except Exception as exc:
         click.echo(f"Export failed: {exc}", err=True)
         sys.exit(1)
+
+
+# ---------------------------------------------------------------------------
+# log  (§3.6)
+# ---------------------------------------------------------------------------
+
+
+@main.command()
+@click.option("--follow", "-f", is_flag=True)
+@click.option("--level", default="INFO", type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR"]))
+@click.option("--component", default=None)
+@click.option("--host", default="127.0.0.1")
+@click.option("--port", default=7080, type=int)
+def log(follow: bool, level: str, component: str | None, host: str, port: int) -> None:
+    """Stream or display recent structured log entries."""
+    url = f"http://{host}:{port}/trace/recent?n=100"
+    try:
+        data = _http_get(url)
+    except ConnectionError:
+        click.echo(f"Node not reachable at {host}:{port}")
+        sys.exit(3)
+
+    entries = data if isinstance(data, list) else data.get("traces", [])
+    for entry in entries:
+        if component and entry.get("component", "") != component:
+            continue
+        entry_level = entry.get("level", "INFO").upper()
+        if ["DEBUG", "INFO", "WARNING", "ERROR"].index(entry_level) < ["DEBUG", "INFO", "WARNING", "ERROR"].index(level):
+            continue
+        ts = entry.get("ts", "?")
+        msg = entry.get("message") or entry.get("capability") or json.dumps(entry)
+        click.echo(f"[{ts}] {entry_level:7s} {msg}")
+
+    if follow:
+        click.echo("(follow mode: reconnect not implemented — use --no-follow for snapshot)")
+
+
+# ---------------------------------------------------------------------------
+# erase  (§3.10)
+# ---------------------------------------------------------------------------
+
+
+@main.command()
+@click.option("--keep-keys", is_flag=True, help="Keep Ed25519 identity keys, erase everything else.")
+@click.option("--yes", is_flag=True, help="Skip confirmation prompt.")
+def erase(keep_keys: bool, yes: bool) -> None:
+    """Erase all local HearthNet data.
+
+    Exit codes: 0 erased, 2 aborted.
+    """
+    config_dir = Path.home() / ".hearthnet"
+    if not yes:
+        click.confirm(
+            f"This will delete {config_dir} {'(keeping keys)' if keep_keys else ''}. Continue?",
+            abort=True,
+        )
+    import shutil
+
+    if not config_dir.exists():
+        click.echo("Nothing to erase.")
+        return
+
+    if keep_keys:
+        key_file = config_dir / "identity.key"
+        key_backup = None
+        if key_file.exists():
+            import tempfile
+            key_backup = Path(tempfile.mktemp(suffix=".key"))
+            import shutil as _sh
+            _sh.copy2(key_file, key_backup)
+        shutil.rmtree(config_dir)
+        if key_backup and key_backup.exists():
+            config_dir.mkdir(parents=True, exist_ok=True)
+            _sh.move(str(key_backup), key_file)
+        click.echo("Data erased (keys preserved).")
+    else:
+        shutil.rmtree(config_dir)
+        click.echo("All HearthNet data erased.")
+
+
+# ---------------------------------------------------------------------------
+# rag subgroup  (§3.11)
+# ---------------------------------------------------------------------------
+
+
+@main.group()
+def rag() -> None:
+    """RAG corpus management."""
+
+
+@rag.command("list")
+@click.option("--host", default="127.0.0.1")
+@click.option("--port", default=7080, type=int)
+def rag_list(host: str, port: int) -> None:
+    """List available RAG corpora."""
+    try:
+        result = _bus_call(host, port, "rag.list_corpora", (1, 0), {})
+    except ConnectionError:
+        click.echo(f"Node not reachable at {host}:{port}")
+        sys.exit(3)
+    corpora = result.get("output", result).get("corpora", [])
+    if not corpora:
+        click.echo("No corpora.")
+        return
+    for c in corpora:
+        name = c.get("name", c) if isinstance(c, dict) else c
+        count = c.get("doc_count", "?") if isinstance(c, dict) else "?"
+        click.echo(f"  {name:<30} docs={count}")
+
+
+@rag.command("ingest")
+@click.argument("path", type=click.Path(exists=True))
+@click.option("--corpus", default="community")
+@click.option("--host", default="127.0.0.1")
+@click.option("--port", default=7080, type=int)
+def rag_ingest(path: str, corpus: str, host: str, port: int) -> None:
+    """Ingest a file or directory into a RAG corpus."""
+    p = Path(path)
+    files: list[Path] = list(p.rglob("*")) if p.is_dir() else [p]
+    ingested = 0
+    for f in files:
+        if not f.is_file():
+            continue
+        data_b64 = __import__("base64").b64encode(f.read_bytes()).decode()
+        try:
+            result = _bus_call(host, port, "rag.ingest", (1, 0), {
+                "input": {"corpus": corpus, "filename": f.name, "data_b64": data_b64}
+            })
+            err = result.get("error")
+            if err:
+                click.echo(f"  SKIP {f.name}: {err}")
+            else:
+                ingested += 1
+                click.echo(f"  OK   {f.name}")
+        except ConnectionError:
+            click.echo(f"Node not reachable at {host}:{port}")
+            sys.exit(3)
+    click.echo(f"Ingested {ingested} file(s) into corpus '{corpus}'.")
+
+
+@rag.command("reindex")
+@click.option("--corpus", default="community")
+@click.option("--embedding-model", default=None)
+@click.option("--host", default="127.0.0.1")
+@click.option("--port", default=7080, type=int)
+def rag_reindex(corpus: str, embedding_model: str | None, host: str, port: int) -> None:
+    """Rebuild the vector index for a corpus."""
+    body: dict = {"input": {"corpus": corpus}}
+    if embedding_model:
+        body["input"]["embedding_model"] = embedding_model
+    try:
+        result = _bus_call(host, port, "rag.reindex", (1, 0), body)
+    except ConnectionError:
+        click.echo(f"Node not reachable at {host}:{port}")
+        sys.exit(3)
+    err = result.get("error")
+    if err:
+        click.echo(f"Reindex failed: {err}", err=True)
+        sys.exit(1)
+    out = result.get("output", result)
+    click.echo(f"Reindexed corpus '{corpus}': {out.get('doc_count', '?')} docs.")
+
+
+# ---------------------------------------------------------------------------
+# invite subgroup  (§3.12)
+# ---------------------------------------------------------------------------
+
+
+@main.group()
+def invite() -> None:
+    """Community invite management."""
+
+
+@invite.command("create")
+@click.argument("node_id")
+@click.option("--level", default="member", type=click.Choice(["member", "trusted", "moderator"]))
+@click.option("--ttl", default=86400, type=int, help="Validity in seconds (default 24h).")
+@click.option("--host", default="127.0.0.1")
+@click.option("--port", default=7080, type=int)
+def invite_create(node_id: str, level: str, ttl: int, host: str, port: int) -> None:
+    """Create an invite link for a new member."""
+    try:
+        result = _bus_call(host, port, "community.invite", (1, 0), {
+            "input": {"invitee_node_id": node_id, "initial_level": level, "ttl_seconds": ttl}
+        })
+    except ConnectionError:
+        click.echo(f"Node not reachable at {host}:{port}")
+        sys.exit(3)
+    err = result.get("error")
+    if err:
+        click.echo(f"Invite failed: {err}", err=True)
+        sys.exit(1)
+    out = result.get("output", result)
+    click.echo(out.get("invite_url") or json.dumps(out, indent=2))
+
+
+@invite.command("redeem")
+@click.argument("text_or_path")
+@click.option("--host", default="127.0.0.1")
+@click.option("--port", default=7080, type=int)
+def invite_redeem(text_or_path: str, host: str, port: int) -> None:
+    """Redeem a hearthnet:// invite link (file path or URL)."""
+    p = Path(text_or_path)
+    invite_text = p.read_text().strip() if p.exists() else text_or_path.strip()
+    try:
+        result = _bus_call(host, port, "community.redeem", (1, 0), {
+            "input": {"invite_text": invite_text}
+        })
+    except ConnectionError:
+        click.echo(f"Node not reachable at {host}:{port}")
+        sys.exit(3)
+    err = result.get("error")
+    if err:
+        click.echo(f"Redeem failed: {err}", err=True)
+        sys.exit(1)
+    out = result.get("output", result)
+    click.echo(f"Joined community: {out.get('community_name', out)}")
+
+
+# ---------------------------------------------------------------------------
+# version  (§3.13)
+# ---------------------------------------------------------------------------
+
+
+@main.command("version")
+def version_cmd() -> None:
+    """Print HearthNet version and exit."""
+    try:
+        from importlib.metadata import version as _v
+        ver = _v("hearthnet")
+    except Exception:
+        try:
+            from hearthnet import __version__ as ver  # type: ignore[attr-defined]
+        except Exception:
+            ver = "dev"
+    click.echo(f"hearthnet {ver}")
+
+
+# ---------------------------------------------------------------------------
+# _bus_call helper (used by several commands above)
+# ---------------------------------------------------------------------------
+
+
+def _bus_call(host: str, port: int, capability: str, version: tuple, body: dict) -> dict:
+    """POST to /bus/v1/call and return parsed JSON. Raises ConnectionError on failure."""
+    payload = {
+        "capability": capability,
+        "version": f"{version[0]}.{version[1]}",
+        **body,
+    }
+    return _http_post(f"http://{host}:{port}/bus/v1/call", json.dumps(payload))

hearthnet/discovery/__init__.py

@@ -2,7 +2,17 @@ from hearthnet.discovery.mdns import MdnsAnnouncer, MdnsBrowser
 from hearthnet.discovery.peers import PeerEvent, PeerRecord, PeerRegistry
 from hearthnet.discovery.udp import UdpAnnouncer, UdpListener
 
+
+class DiscoveryError(Exception):
+    """Raised for unrecoverable discovery failures (M02)."""
+
+    def __init__(self, code: str, reason: str = "") -> None:
+        super().__init__(reason or code)
+        self.code = code
+
+
 __all__ = [
+    "DiscoveryError",
     "MdnsAnnouncer",
     "MdnsBrowser",
     "PeerEvent",

hearthnet/identity/__init__.py

@@ -23,8 +23,11 @@ from hearthnet.identity.keys import (
 )
 from hearthnet.identity.manifest import (
     CommunityManifest,
+    CommunityMember,
+    CommunityPolicy,
     ManifestError,
     NodeManifest,
+    RevokedEntry,
     build_community_manifest,
     build_node_manifest,
     verify_community_manifest,
@@ -50,6 +53,9 @@ __all__ = [
     "ManifestError",
     "NodeManifest",
     "CommunityManifest",
+    "CommunityMember",
+    "CommunityPolicy",
+    "RevokedEntry",
     "build_node_manifest",
     "verify_node_manifest",
     "build_community_manifest",

hearthnet/identity/manifest.py

@@ -172,6 +172,33 @@ class NodeManifest:
         return d
 
 
+@dataclass(frozen=True)
+class RevokedEntry:
+    """A revoked member entry in a community manifest."""
+    node_id: str
+    revoked_at: str
+    reason: str = ""
+
+
+@dataclass(frozen=True)
+class CommunityMember:
+    """A member record in a community manifest."""
+    node_id: str
+    display_name: str
+    level: str  # "root" | "trusted" | "moderator" | "member"
+    joined_at: str
+    invited_by: str = ""
+
+
+@dataclass(frozen=True)
+class CommunityPolicy:
+    """Community governance policy embedded in CommunityManifest."""
+    allow_public_join: bool = False
+    require_invite: bool = True
+    max_members: int = 500
+    min_trust_for_invite: str = "member"
+
+
 @dataclass(frozen=True)
 class CommunityManifest:
     version: int

hearthnet/node.py

@@ -512,28 +512,80 @@ class InMemoryNetwork:
                     node.discover(other)
 
 
+# ---------------------------------------------------------------------------
+# PeriodicTask — generic async interval runner (M12 §5)
+# ---------------------------------------------------------------------------
 
-@dataclass
-class NodeManifest:
-    node_id: NodeID
-    display_name: str
-    community_id: CommunityID
-    profile: Profile
-    capabilities: list[dict[str, Any]]
 
-    def as_dict(self) -> dict[str, Any]:
-        return {
-            "version": 1,
-            "contract_version": "1.0",
-            "node_id": self.node_id,
-            "display_name": self.display_name,
-            "community_id": self.community_id,
-            "profile": self.profile,
-            "capabilities": self.capabilities,
-        }
+class PeriodicTask:
+    """Run *fn* every *interval_seconds* until cancelled.
+
+    Usage::
+
+        task = PeriodicTask(my_async_fn, interval_seconds=60)
+        asyncio.create_task(task.run())
+    """
+
+    def __init__(self, fn, interval_seconds: int) -> None:
+        self._fn = fn
+        self._interval = interval_seconds
+
+    async def run(self) -> None:
+        while True:
+            await asyncio.sleep(self._interval)
+            try:
+                await self._fn()
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                _log.debug("PeriodicTask %s error: %s", self._fn, exc)
+
+
+# ---------------------------------------------------------------------------
+# ManifestPublisher — republishes node manifest to mDNS + UDP (M12 §5)
+# ---------------------------------------------------------------------------
+
+_MANIFEST_REPUBLISH_INTERVAL_SECONDS = 300  # 5 minutes default
+
+
+class ManifestPublisher:
+    """Periodically re-publishes the node manifest to mDNS + UDP announcer.
+
+    Also triggered when the bus registry changes (capability added/removed).
+    """
+
+    def __init__(
+        self,
+        bus,
+        peer_registry,
+        mdns_announcer=None,
+        udp_announcer=None,
+        node_manifest_fn=None,
+        interval_seconds: int = _MANIFEST_REPUBLISH_INTERVAL_SECONDS,
+    ) -> None:
+        self._bus = bus
+        self._peer_registry = peer_registry
+        self._mdns_announcer = mdns_announcer
+        self._udp_announcer = udp_announcer
+        self._node_manifest_fn = node_manifest_fn
+        self._interval = interval_seconds
+        self._task: asyncio.Task | None = None
+
+    async def run(self) -> None:
+        """Publish immediately then republish every *interval_seconds*."""
+        while True:
+            await self._publish()
+            await asyncio.sleep(self._interval)
+
+    async def _publish(self) -> None:
+        try:
+            manifest = self._node_manifest_fn() if self._node_manifest_fn else {}
+            caps = [c.get("name") for c in manifest.get("capabilities", [])]
+            if self._mdns_announcer and hasattr(self._mdns_announcer, "republish"):
+                await self._mdns_announcer.republish(caps)
+            if self._udp_announcer and hasattr(self._udp_announcer, "republish"):
+                await self._udp_announcer.republish()
+        except Exception as exc:
+            _log.debug("ManifestPublisher._publish error: %s", exc)
+
 
-    def mesh_discover(self) -> None:
-        for node in self.nodes:
-            for other in self.nodes:
-                if node is not other:
-                    node.discover(other)

hearthnet/observability/doctor.py

@@ -297,3 +297,11 @@ def run_one(name: str) -> DoctorResult:
             message=f"Check raised an unexpected error: {exc}",
             extra={"error": str(exc)},
         )
+
+
+# ---------------------------------------------------------------------------
+# Spec-mandated alias: CheckResult == DoctorResult (X03 §3.5)
+# ---------------------------------------------------------------------------
+
+CheckResult = DoctorResult
+

hearthnet/observability/metrics.py

@@ -235,3 +235,92 @@ def signature_failures_total() -> Any:
         "Signature verification failures",
         ["reason"],
     )
+
+
+# ---------------------------------------------------------------------------
+# TrackioExporter — optional HuggingFace Trackio integration (X03 §23)
+# ---------------------------------------------------------------------------
+
+
+class TrackioExporter:
+    """Optional bridge to HuggingFace Trackio experiment tracker.
+
+    Activated only when ``config.observability.trackio_project`` is set.
+    Falls back to no-op if ``trackio`` is not installed.
+
+    Usage (from node.py or CLI)::
+
+        exporter = TrackioExporter(project="hearthnet-demo")
+        exporter.log_llm_call(latency_ms=120, tokens_in=50, tokens_out=80, model="llama3", backend="ollama", result="ok")
+    """
+
+    def __init__(
+        self,
+        project: str,
+        space: str | None = None,
+        run_name: str | None = None,
+    ) -> None:
+        self._project = project
+        self._space = space
+        self._run_name = run_name or "hearthnet"
+        self._run = None
+        self._enabled = False
+        self._try_init()
+
+    def _try_init(self) -> None:
+        try:
+            import trackio  # type: ignore[import]
+            self._run = trackio.init(project=self._project, name=self._run_name)
+            self._enabled = True
+        except ImportError:
+            pass  # trackio not installed — silently no-op
+        except Exception:
+            pass
+
+    @property
+    def enabled(self) -> bool:
+        return self._enabled
+
+    def log_llm_call(
+        self,
+        *,
+        latency_ms: float,
+        tokens_in: int,
+        tokens_out: int,
+        model: str,
+        backend: str,
+        result: str,
+    ) -> None:
+        if not self._enabled or self._run is None:
+            return
+        try:
+            self._run.log({
+                "latency_ms": latency_ms,
+                "tokens_in": tokens_in,
+                "tokens_out": tokens_out,
+                "model": model,
+                "backend": backend,
+                "result": result,
+            })
+        except Exception:
+            pass
+
+    def log_topology(self, mesh_size: int, online: bool, cap_count: int) -> None:
+        if not self._enabled or self._run is None:
+            return
+        try:
+            self._run.log({
+                "mesh_size": mesh_size,
+                "online": int(online),
+                "capability_count": cap_count,
+            })
+        except Exception:
+            pass
+
+    def close(self) -> None:
+        if self._run is not None:
+            try:
+                self._run.finish()
+            except Exception:
+                pass
+

hearthnet/observability/tracing.py

@@ -95,6 +95,11 @@ def attach(trace: Trace) -> None:
     _current_trace.set(trace)
 
 
+def detach() -> None:
+    """Clear the active trace from this context."""
+    _current_trace.set(None)  # type: ignore[arg-type]
+
+
 @contextmanager
 def span(name: str, **extras: object) -> Iterator[Span]:
     """Context-manager that records a Span on the current Trace (if any).

hearthnet/services/llm/backends/anthropic_api.py

@@ -0,0 +1,119 @@
+"""M04 — Anthropic API backend (cloud, opt-in).
+
+Uses Anthropic's Messages API: https://api.anthropic.com/v1/messages
+Requires ANTHROPIC_API_KEY env var. Online-only; M09 deregisters when offline.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from hearthnet.services.llm.backends.base import BackendModel, Token
+
+
+class AnthropicApiBackend:
+    """Anthropic Claude API — cloud LLM endpoint.
+
+    Online-only opt-in fallback. Set ANTHROPIC_API_KEY to enable.
+    Supports: claude-3-haiku-20240307 (fast/cheap), claude-3-sonnet-20240229.
+    """
+
+    name = "anthropic_api"
+    _ANTHROPIC_VERSION = "2023-06-01"
+
+    def __init__(
+        self,
+        model: str = "claude-3-haiku-20240307",
+        api_key_env: str = "ANTHROPIC_API_KEY",
+        base_url: str = "https://api.anthropic.com",
+    ) -> None:
+        self._model = model
+        self._api_key_env = api_key_env
+        self._base_url = base_url.rstrip("/")
+        self.models = [
+            BackendModel(
+                name="claude-3-haiku-20240307",
+                family="claude",
+                context_length=200_000,
+                requires_internet=True,
+            ),
+            BackendModel(
+                name="claude-3-sonnet-20240229",
+                family="claude",
+                context_length=200_000,
+                requires_internet=True,
+            ),
+        ]
+
+    def _get_key(self) -> str:
+        return os.environ.get(self._api_key_env, "")
+
+    def is_available(self) -> bool:
+        return bool(self._get_key())
+
+    async def warm(self) -> None:
+        pass
+
+    async def health(self) -> dict:
+        return {"ok": self.is_available(), "backend": self.name, "model": self._model}
+
+    async def chat(self, messages: list[dict], *, max_tokens: int = 1024, **kwargs):
+        """Async generator yielding Token objects."""
+        import urllib.request
+        import urllib.error
+
+        key = self._get_key()
+        if not key:
+            raise RuntimeError(f"{self._api_key_env} not set; Anthropic API unavailable")
+
+        # Separate system message
+        system = ""
+        user_messages = []
+        for m in messages:
+            if m.get("role") == "system":
+                system = m.get("content", "")
+            else:
+                user_messages.append({"role": m["role"], "content": m.get("content", "")})
+
+        payload: dict = {
+            "model": self._model,
+            "max_tokens": max_tokens,
+            "messages": user_messages or [{"role": "user", "content": "Hi"}],
+        }
+        if system:
+            payload["system"] = system
+
+        url = f"{self._base_url}/v1/messages"
+        req = urllib.request.Request(  # nosec B310
+            url,
+            data=json.dumps(payload).encode(),
+            headers={
+                "x-api-key": key,
+                "anthropic-version": self._ANTHROPIC_VERSION,
+                "content-type": "application/json",
+            },
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=60) as resp:  # nosec B310
+                data = json.loads(resp.read())
+        except urllib.error.HTTPError as exc:
+            body = exc.read().decode(errors="replace")
+            raise RuntimeError(f"Anthropic API {exc.code}: {body}") from exc
+        except OSError as exc:
+            raise RuntimeError(f"Anthropic API connection error: {exc}") from exc
+
+        text = ""
+        for block in data.get("content", []):
+            if block.get("type") == "text":
+                text += block.get("text", "")
+
+        yield Token(text=text, logprob=None, finish_reason=data.get("stop_reason", "stop"))
+
+    async def complete(self, prompt: str, *, max_tokens: int = 512, **kwargs):
+        """Async generator yielding Token objects."""
+        async for tok in self.chat([{"role": "user", "content": prompt}], max_tokens=max_tokens):
+            yield tok
+
+    async def close(self) -> None:
+        pass

hearthnet/services/llm/backends/hf_api.py

@@ -0,0 +1,102 @@
+"""M04 — Hugging Face Inference API backend (cloud, opt-in).
+
+Uses the HF Inference API: https://api-inference.huggingface.co/
+Requires HEARTHNET_HF_TOKEN env var. Online-only; M09 deregisters when offline.
+"""
+
+from __future__ import annotations
+
+import os
+from hearthnet.services.llm.backends.base import BackendModel, ChatResult, Token
+
+
+class HfApiBackend:
+    """Hugging Face Inference API — cloud LLM endpoint.
+
+    Online-only fallback. Set HEARTHNET_HF_TOKEN to enable.
+    Default model: HuggingFaceH4/zephyr-7b-beta (public, instruction-tuned).
+    """
+
+    name = "hf_api"
+
+    def __init__(
+        self,
+        model: str = "HuggingFaceH4/zephyr-7b-beta",
+        api_key_env: str = "HEARTHNET_HF_TOKEN",
+        base_url: str = "https://api-inference.huggingface.co",
+    ) -> None:
+        self._model = model
+        self._api_key_env = api_key_env
+        self._base_url = base_url.rstrip("/")
+        self.models = [
+            BackendModel(
+                name=model,
+                family="hf_api",
+                context_length=4096,
+                requires_internet=True,
+            )
+        ]
+
+    def _get_key(self) -> str:
+        return os.environ.get(self._api_key_env, "")
+
+    def is_available(self) -> bool:
+        return bool(self._get_key())
+
+    async def warm(self) -> None:
+        pass
+
+    async def health(self) -> dict:
+        return {"ok": self.is_available(), "backend": self.name, "model": self._model}
+
+    async def chat(self, messages: list[dict], *, max_tokens: int = 512, **kwargs):
+        """Async generator yielding Token objects (streaming)."""
+        import json
+        import urllib.request
+
+        key = self._get_key()
+        if not key:
+            raise RuntimeError(f"{self._api_key_env} not set; HF Inference API unavailable")
+
+        # Convert chat messages to a single prompt for text-generation endpoint
+        prompt = "\n".join(
+            f"{'User' if m.get('role') == 'user' else 'Assistant'}: {m.get('content', '')}"
+            for m in messages
+        )
+        prompt += "\nAssistant:"
+
+        url = f"{self._base_url}/models/{self._model}"
+        payload = json.dumps({
+            "inputs": prompt,
+            "parameters": {"max_new_tokens": max_tokens, "return_full_text": False},
+        }).encode()
+        req = urllib.request.Request(  # nosec B310
+            url,
+            data=payload,
+            headers={
+                "Authorization": f"Bearer {key}",
+                "Content-Type": "application/json",
+            },
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(req, timeout=30) as resp:  # nosec B310
+                data = json.loads(resp.read())
+        except Exception as exc:
+            raise RuntimeError(f"HF Inference API error: {exc}") from exc
+
+        text = ""
+        if isinstance(data, list) and data:
+            text = data[0].get("generated_text", "")
+        elif isinstance(data, dict):
+            text = data.get("generated_text", "")
+
+        yield Token(text=text, logprob=None, finish_reason="stop")
+
+    async def complete(self, prompt: str, *, max_tokens: int = 256, **kwargs):
+        """Async generator yielding Token objects."""
+        async for tok in self.chat([{"role": "user", "content": prompt}], max_tokens=max_tokens):
+            yield tok
+
+    async def close(self) -> None:
+        pass

hearthnet/services/llm/backends/lmstudio.py

@@ -0,0 +1,34 @@
+"""M04 — LmStudio OpenAI-compatible backend.
+
+LM Studio serves a local OpenAI-compatible API on http://localhost:1234/v1.
+Wraps OpenAICompatBackend with LM Studio defaults.
+"""
+
+from __future__ import annotations
+
+from hearthnet.services.llm.backends.openai_compat import OpenAICompatBackend
+
+
+class LmStudioBackend(OpenAICompatBackend):
+    """LM Studio local inference server.
+
+    Default endpoint: http://localhost:1234/v1
+    LM Studio exposes whichever model is currently loaded; it is discovered
+    dynamically via GET /v1/models on first availability check.
+    """
+
+    def __init__(
+        self,
+        base_url: str = "http://localhost:1234/v1",
+        model: str = "local-model",
+        api_key_env: str = "",
+    ) -> None:
+        super().__init__(
+            base_url=base_url,
+            api_key_env=api_key_env,
+            model=model,
+        )
+
+    @property
+    def name(self) -> str:  # type: ignore[override]
+        return "lmstudio"

hearthnet/transport/__init__.py

@@ -1,13 +1,19 @@
+from hearthnet.transport.backpressure import FlowControl, RateCheck, RateLimiter
 from hearthnet.transport.client import CallError, HttpClient
 from hearthnet.transport.server import HttpServer
-from hearthnet.transport.streams import SseWriter, encode_sse_frame
+from hearthnet.transport.streams import Frame, SseReader, SseWriter, encode_sse_frame
 from hearthnet.transport.tls import PinnedCerts, generate_self_signed_cert, load_or_generate_cert
 
 __all__ = [
     "CallError",
+    "FlowControl",
+    "Frame",
     "HttpClient",
     "HttpServer",
     "PinnedCerts",
+    "RateCheck",
+    "RateLimiter",
+    "SseReader",
     "SseWriter",
     "encode_sse_frame",
     "generate_self_signed_cert",

hearthnet/transport/backpressure.py

@@ -0,0 +1,133 @@
+"""X01 — Backpressure / flow control.
+
+Spec: docs/X01-transport.md §3.4
+
+FlowControl gates outbound work when a downstream consumer is slow.
+Used by HttpServer SSE streams and WebSocket pub-sub to avoid unbounded queues.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+
+class FlowControl:
+    """Leaky-bucket / semaphore flow control for streaming responses.
+
+    Usage::
+
+        fc = FlowControl(capacity=32)
+        async with fc.acquire():
+            await stream_chunk(data)
+
+    If the number of in-flight chunks reaches *capacity*, ``acquire()``
+    blocks until a slot is freed.  This creates natural back-pressure so
+    a slow HTTP client cannot cause the server to buffer unbounded data.
+    """
+
+    def __init__(self, capacity: int = 64) -> None:
+        if capacity < 1:
+            raise ValueError("capacity must be >= 1")
+        self._sem = asyncio.Semaphore(capacity)
+        self._capacity = capacity
+        self._total_acquired: int = 0
+        self._total_released: int = 0
+
+    @property
+    def capacity(self) -> int:
+        return self._capacity
+
+    @property
+    def in_flight(self) -> int:
+        return self._total_acquired - self._total_released
+
+    def acquire(self) -> "_AcquireContext":
+        return _AcquireContext(self)
+
+    async def _acquire(self) -> None:
+        await self._sem.acquire()
+        self._total_acquired += 1
+
+    def _release(self) -> None:
+        self._sem.release()
+        self._total_released += 1
+
+    def stats(self) -> dict:
+        return {
+            "capacity": self._capacity,
+            "in_flight": self.in_flight,
+            "total_acquired": self._total_acquired,
+            "total_released": self._total_released,
+        }
+
+
+class _AcquireContext:
+    def __init__(self, fc: FlowControl) -> None:
+        self._fc = fc
+
+    async def __aenter__(self) -> "_AcquireContext":
+        await self._fc._acquire()
+        return self
+
+    async def __aexit__(self, *_) -> None:
+        self._fc._release()
+
+
+# ---------------------------------------------------------------------------
+# RateCheck / RateLimiter (X01 §3.5)
+# ---------------------------------------------------------------------------
+
+
+class RateCheck:
+    """Simple sliding-window rate check (read-only, no blocking).
+
+    Use to check whether a call is within limits before proceeding.
+    Returns True if allowed, False if over limit.
+    """
+
+    def __init__(self, max_calls: int, window_seconds: float = 1.0) -> None:
+        self._max = max_calls
+        self._window = window_seconds
+        self._calls: list[float] = []
+
+    def check(self, now: float | None = None) -> bool:
+        import time
+        t = now if now is not None else time.monotonic()
+        cutoff = t - self._window
+        self._calls = [c for c in self._calls if c > cutoff]
+        if len(self._calls) < self._max:
+            self._calls.append(t)
+            return True
+        return False
+
+    def reset(self) -> None:
+        self._calls.clear()
+
+
+class RateLimiter:
+    """Async rate limiter — blocks until a slot is available.
+
+    Usage::
+
+        rl = RateLimiter(max_calls=10, window_seconds=1.0)
+        await rl.acquire()
+        # ... do work ...
+    """
+
+    def __init__(self, max_calls: int, window_seconds: float = 1.0) -> None:
+        self._max = max_calls
+        self._window = window_seconds
+        self._calls: list[float] = []
+        self._lock = asyncio.Lock()
+
+    async def acquire(self) -> None:
+        import time
+        while True:
+            async with self._lock:
+                t = time.monotonic()
+                cutoff = t - self._window
+                self._calls = [c for c in self._calls if c > cutoff]
+                if len(self._calls) < self._max:
+                    self._calls.append(t)
+                    return
+            await asyncio.sleep(self._window / self._max)

hearthnet/transport/streams.py

@@ -28,6 +28,66 @@ async def parse_sse_stream(lines: AsyncIterator[str]) -> AsyncIterator[dict]:
                 pass
 
 
+# ---------------------------------------------------------------------------
+# Frame — typed SSE frame (X01 §3.2)
+# ---------------------------------------------------------------------------
+
+
+class Frame:
+    """A single SSE frame with optional event tag and raw data.
+
+    Spec: X01-transport §3.2 — wire format is ``data: <json>\\n\\n``
+    with optional ``event: <tag>\\n`` prefix.
+    """
+
+    __slots__ = ("event", "data", "raw")
+
+    def __init__(self, data: dict, event: str | None = None) -> None:
+        self.data = data
+        self.event = event
+        self.raw = encode_sse_frame(data, event)
+
+    def __repr__(self) -> str:
+        return f"Frame(event={self.event!r}, data={self.data!r})"
+
+
+# ---------------------------------------------------------------------------
+# SseReader — parse an HTTP SSE response stream (X01 §3.2)
+# ---------------------------------------------------------------------------
+
+
+class SseReader:
+    """Parse a streaming HTTP response into Frame objects.
+
+    Typical usage with httpx::
+
+        async with httpx.AsyncClient() as client:
+            async with client.stream("POST", url, ...) as resp:
+                reader = SseReader(resp.aiter_lines())
+                async for frame in reader:
+                    handle(frame)
+    """
+
+    def __init__(self, lines: AsyncIterator[str]) -> None:
+        self._lines = lines
+
+    async def __aiter__(self) -> AsyncIterator[Frame]:
+        event_tag: str | None = None
+        async for line in self._lines:
+            if line.startswith("event:"):
+                event_tag = line[6:].strip()
+            elif line.startswith("data:"):
+                raw = line[5:].strip()
+                try:
+                    data = json.loads(raw)
+                except json.JSONDecodeError:
+                    data = {"raw": raw}
+                yield Frame(data, event_tag)
+                event_tag = None
+            elif not line.strip():
+                event_tag = None  # blank separator
+
+
 class SseWriter:
     """Async generator that yields SSE-formatted strings."""
 

hearthnet/ui/onboarding.py

@@ -250,3 +250,8 @@ def _iso_after(seconds: int) -> str:
     from datetime import datetime, timedelta
 
     return (datetime.now(UTC) + timedelta(seconds=seconds)).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+# Spec-mandated name (M13 §3.1)
+build_onboarding = build_onboarding_ui
+

hearthnet/ui/theme.py

@@ -0,0 +1,53 @@
+"""M08 — Gradio theme definitions.
+
+Spec: docs/M08-ui.md §7
+
+Two themes:
+  hearthnet_theme — default purple/dark theme used at all times
+  emergency_theme — red-accent override shown in emergency mode
+"""
+
+from __future__ import annotations
+
+try:
+    import gradio as gr
+
+    hearthnet_theme = gr.themes.Soft(
+        primary_hue="purple",
+        secondary_hue="violet",
+        neutral_hue="slate",
+        font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"],
+    ).set(
+        # CSS variable overrides (spec §7)
+        body_background_fill="#1a1a2e",
+        body_background_fill_dark="#0f0f1a",
+        block_background_fill="#16213e",
+        block_border_color="#7c3aed",
+        button_primary_background_fill="#7c3aed",
+        button_primary_background_fill_hover="#6d28d9",
+        button_primary_text_color="#ffffff",
+        input_background_fill="#0f3460",
+    )
+
+    emergency_theme = gr.themes.Soft(
+        primary_hue="red",
+        secondary_hue="orange",
+        neutral_hue="zinc",
+        font=[gr.themes.GoogleFont("Inter"), "ui-sans-serif", "system-ui", "sans-serif"],
+    ).set(
+        body_background_fill="#1a0a0a",
+        body_background_fill_dark="#0f0505",
+        block_background_fill="#2d0000",
+        block_border_color="#dc2626",
+        button_primary_background_fill="#dc2626",
+        button_primary_background_fill_hover="#b91c1c",
+        button_primary_text_color="#ffffff",
+        input_background_fill="#1f0000",
+    )
+
+except ImportError:
+    # Gradio not installed — provide None sentinels so imports don't fail
+    hearthnet_theme = None  # type: ignore[assignment]
+    emergency_theme = None  # type: ignore[assignment]
+
+__all__ = ["hearthnet_theme", "emergency_theme"]

hearthnet/ui/topology.py

@@ -0,0 +1,138 @@
+"""M08 — Topology visualisation component.
+
+Spec: docs/M08-ui.md §3.2
+
+Renders the live mesh topology and recent call traces as an HTML widget.
+Updates are pushed via TopologyComponent.push_trace() and push_topology().
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from collections import deque
+from typing import Any
+
+try:
+    import gradio as gr
+    _HAS_GRADIO = True
+except ImportError:
+    _HAS_GRADIO = False
+
+
+# Max recent call traces to keep in memory
+_MAX_TRACES = 200
+_MAX_TOPOLOGY_HISTORY = 10
+
+
+class TopologyComponent:
+    """Live mesh topology and call-trace viewer.
+
+    Renders an HTML card showing:
+    - Connected peers (node_id, capabilities count, latency)
+    - Recent bus call traces (capability, duration_ms, success/error)
+    - Local capability count
+
+    Call push_trace() / push_topology() from bus hooks to keep it live.
+    Integrate into Gradio UI via render().
+    """
+
+    def __init__(self, bus: Any = None) -> None:
+        self._bus = bus
+        self._traces: deque[dict] = deque(maxlen=_MAX_TRACES)
+        self._topology: dict = {}
+        self._last_updated: float = 0.0
+
+    def push_trace(self, event: Any) -> None:
+        """Accept a CallTraceEvent (or dict) and store it."""
+        if hasattr(event, "__dict__"):
+            rec = {
+                "ts": getattr(event, "ts", time.strftime("%H:%M:%S")),
+                "capability": getattr(event, "capability", "?"),
+                "duration_ms": getattr(event, "duration_ms", 0),
+                "success": getattr(event, "success", True),
+                "error": getattr(event, "error", None),
+                "peer_node_id": getattr(event, "peer_node_id", "local"),
+            }
+        elif isinstance(event, dict):
+            rec = event
+        else:
+            return
+        self._traces.appendleft(rec)
+        self._last_updated = time.monotonic()
+
+    def push_topology(self, snapshot: Any) -> None:
+        """Accept a TopologySnapshot (or dict) and store it."""
+        if isinstance(snapshot, dict):
+            self._topology = snapshot
+        elif hasattr(snapshot, "as_dict"):
+            self._topology = snapshot.as_dict()
+        elif hasattr(snapshot, "__dict__"):
+            self._topology = vars(snapshot)
+        self._last_updated = time.monotonic()
+
+    def render(self) -> Any:
+        """Return a Gradio HTML component showing current topology."""
+        if not _HAS_GRADIO:
+            raise ImportError("gradio is required for TopologyComponent.render()")
+
+        html = self._build_html()
+        return gr.HTML(value=html, label="Mesh Topology")
+
+    def _build_html(self) -> str:
+        peers = self._topology.get("peers", [])
+        local_caps = self._topology.get("local_capabilities", 0)
+        community = self._topology.get("community_id", "—")
+
+        # Build peer rows
+        peer_rows = ""
+        for p in peers[:20]:
+            nid = str(p.get("node_id", "?"))[:12]
+            caps = p.get("capabilities_count", "?")
+            lat = p.get("latency_ms", "?")
+            peer_rows += f"<tr><td>{nid}…</td><td>{caps}</td><td>{lat}ms</td></tr>"
+        if not peers:
+            peer_rows = "<tr><td colspan='3' style='color:#888'>No peers discovered yet</td></tr>"
+
+        # Build trace rows
+        trace_rows = ""
+        for t in list(self._traces)[:15]:
+            cap = str(t.get("capability", "?"))[:35]
+            dur = t.get("duration_ms", "?")
+            ok = "✓" if t.get("success", True) else "✗"
+            color = "#4ade80" if t.get("success", True) else "#f87171"
+            trace_rows += f"<tr><td style='color:{color}'>{ok}</td><td>{cap}</td><td>{dur}ms</td></tr>"
+        if not trace_rows:
+            trace_rows = "<tr><td colspan='3' style='color:#888'>No calls yet</td></tr>"
+
+        ts = time.strftime("%H:%M:%S") if self._last_updated else "never"
+
+        return f"""
+<div style="font-family:monospace;color:#e2e8f0;background:#16213e;padding:12px;border-radius:8px;border:1px solid #7c3aed">
+  <div style="display:flex;justify-content:space-between;margin-bottom:8px">
+    <span style="font-size:14px;font-weight:600;color:#a78bfa">Mesh Topology</span>
+    <span style="font-size:11px;color:#64748b">updated {ts}</span>
+  </div>
+  <div style="margin-bottom:6px;font-size:12px;color:#94a3b8">
+    Community: <b style="color:#c4b5fd">{community}</b> ·
+    Local caps: <b style="color:#c4b5fd">{local_caps}</b> ·
+    Peers: <b style="color:#c4b5fd">{len(peers)}</b>
+  </div>
+  <table style="width:100%;font-size:11px;border-collapse:collapse;margin-bottom:10px">
+    <thead><tr style="color:#7c3aed"><th>Node</th><th>Caps</th><th>Latency</th></tr></thead>
+    <tbody>{peer_rows}</tbody>
+  </table>
+  <div style="font-size:12px;color:#94a3b8;margin-bottom:4px">Recent calls</div>
+  <table style="width:100%;font-size:11px;border-collapse:collapse">
+    <thead><tr style="color:#7c3aed"><th></th><th>Capability</th><th>Duration</th></tr></thead>
+    <tbody>{trace_rows}</tbody>
+  </table>
+</div>
+"""
+
+    def as_dict(self) -> dict:
+        return {
+            "topology": self._topology,
+            "recent_traces": list(self._traces)[:20],
+            "last_updated": self._last_updated,
+        }