feat: activate real services, fix gossip, wire sponsor backends + EventLog

P1 gossip: _HttpxSyncClient adapter fixes _gossip_loop (wrong HttpClient args + missing httpx .get/.post). P2/P3: install_extended_services registers Embedding(bge-small)/Rerank/Ocr/Translation/Stt/Tts/Image* with graceful unavailable. P4: EvidenceService + CivilDefenseService bus adapters (research=True). P6: app.py uses real RagService+FederatedRagService with seeded corpus, multi-backend LlmService (HF + opt-in Nemotron/Modal/MiniCPM), opens EventLog and injects into Marketplace/Chat. Adds sentence-transformers, httpx, pytest-asyncio. No mocks; optional deps degrade honestly.

app.py

@@ -29,6 +29,7 @@ See docs/HOWTO.md for Raspberry Pi, Docker, and multi-node mesh setup.
 
 from __future__ import annotations
 
+import contextlib
 import os
 
 # ─────────────────────────────────────────────────────────────────────────────
@@ -169,7 +170,6 @@ def _build_node():
 
     from hearthnet.node import HearthNode
     from hearthnet.services.chat.service import ChatService
-    from hearthnet.services.demo import RagService as DemoRagService
     from hearthnet.services.files.service import FileService
     from hearthnet.services.llm.backends.hf_local import HfLocalBackend
     from hearthnet.services.llm.service import LlmService
@@ -252,36 +252,134 @@ def _build_node():
 
             HfLocalBackend.chat = _patched_chat  # type: ignore[method-assign]
 
-        llm = LlmService(backends=[backend])
+        backends: list = [backend]
+        # ── Sponsor cloud backends (opt-in via env) ───────────────────────
+        # NVIDIA Nemotron (prize track) — cloud NIM, no local availability check.
+        if os.getenv("NVIDIA_API_KEY"):
+            try:
+                from hearthnet.services.llm.backends.nemotron import NemotronBackend
+
+                backends.append(NemotronBackend(api_key_env="NVIDIA_API_KEY"))
+            except Exception:
+                pass
+        # Modal serverless GPU (prize track).
+        if os.getenv("MODAL_ENDPOINT"):
+            try:
+                from hearthnet.services.llm.backends.modal_backend import ModalBackend
+
+                modal_b = ModalBackend()
+                if modal_b.is_available():
+                    backends.append(modal_b)
+            except Exception:
+                pass
+        # MiniCPM local server (OpenBMB prize track).
+        _minicpm_url = os.getenv("MINICPM_URL")
+        if _minicpm_url:
+            try:
+                from hearthnet.services.llm.backends.openbmb import OpenBmbBackend
+
+                minicpm = OpenBmbBackend(base_url=_minicpm_url)
+                if minicpm.is_available():
+                    backends.append(minicpm)
+            except Exception:
+                pass
+
+        llm = LlmService(backends=backends)
     except Exception:
         llm = LlmService()  # _UnavailableBackend — shows clear error
 
     node.bus.register_service(llm)
 
-    # RAG — pre-seeded community corpus using demo RagService (in-memory)
-    rag = DemoRagService(corpus="community")
-    rag.documents = list(SEED_CORPUS)
-    node.bus.register_service(rag)
+    # ── Durable event log (ZeroGPU-safe; no mDNS/transport on a single Space) ──
+    event_log = None
+    try:
+        import tempfile
+        from pathlib import Path
+
+        from hearthnet.events import EventLog
+
+        _data_dir = Path(os.getenv("HEARTHNET_DATA_DIR", tempfile.gettempdir())) / "hearthnet-space"
+        _data_dir.mkdir(parents=True, exist_ok=True)
+        event_log = EventLog(_data_dir / "events.db", node.community_id, node.node_id)
+        node._event_log = event_log
+    except Exception:
+        event_log = None
+
+    # ── Blob store for content-addressed RAG documents ────────────────────
+    blob_store = None
+    try:
+        import tempfile
+        from pathlib import Path
+
+        from hearthnet.blobs.store import BlobStore
 
-    # Register a synthetic rag.list_corpora so the Ask tab can discover corpora
-    from hearthnet.bus.capability import CapabilityDescriptor, RouteRequest
+        blob_store = BlobStore(
+            Path(os.getenv("HEARTHNET_DATA_DIR", tempfile.gettempdir()))
+            / "hearthnet-space"
+            / "blobs"
+        )
+    except Exception:
+        blob_store = None
+
+    # ── Real semantic RAG (replaces the in-memory demo corpus) ────────────
+    from hearthnet.bus.capability import RouteRequest
+    from hearthnet.services.rag.federated import FederatedRagService
+    from hearthnet.services.rag.service import RagService
 
-    async def _list_corpora(req: RouteRequest) -> dict:
-        return {"output": {"corpora": ["community"]}, "meta": {}}
+    # Register the embedding backend first so rag.query routes through embed.text.
+    node.install_extended_services(research=True)
 
-    node.bus.register_capability(
-        CapabilityDescriptor(name="rag.list_corpora", version=(1, 0)),
-        _list_corpora,
+    rag = RagService(
+        corpus="community",
+        bus=node.bus,
+        event_log=event_log,
+        blob_store=blob_store,
     )
+    node.bus.register_service(rag)
+    node.bus.register_service(FederatedRagService(node.bus, corpus="community"))
+
+    # Seed the corpus through the real ingest path (content-addressed + logged).
+    async def _seed_corpus() -> None:
+        for doc in SEED_CORPUS:
+            with contextlib.suppress(Exception):
+                await rag.handle_ingest(
+                    RouteRequest(
+                        capability="rag.ingest",
+                        version_req=(1, 0),
+                        body={
+                            "input": {
+                                "corpus": "community",
+                                "documents": [
+                                    {
+                                        "id": doc["id"],
+                                        "title": doc["title"],
+                                        "text": doc["text"],
+                                    }
+                                ],
+                            }
+                        },
+                        caller=node.node_id,
+                        trace_id="seed",
+                        deadline_ms=0,
+                    )
+                )
 
-    # Marketplace, Chat, Files
-    node.bus.register_service(MarketplaceService())
-    node.bus.register_service(ChatService(node.node_id))
+    with contextlib.suppress(Exception):
+        import asyncio
+
+        asyncio.run(_seed_corpus())
+
+    # Marketplace, Chat, Files — now durably event-sourced where supported.
+    node.bus.register_service(
+        MarketplaceService(event_log=event_log, node_id=node.node_id)
+    )
+    node.bus.register_service(ChatService(node.node_id, event_log=event_log))
     node.bus.register_service(FileService())
 
     return node
 
 
+
 # Build node and Gradio app at import time (HF Spaces requires module-level `demo`)
 _node = _build_node()
 

hearthnet/civdef/service.py

@@ -213,3 +213,103 @@ class CivilDefenseService:
             "chain_valid": self._audit.verify_integrity(),
             "length": self._audit.length(),
         }
+
+    # ── Capability-bus adapter (registered only under research=True) ────────
+
+    name = "civdef"
+    version = "1.0"
+
+    def capabilities(self) -> list[tuple]:
+        from hearthnet.bus.capability import CapabilityDescriptor
+
+        return [
+            (
+                CapabilityDescriptor(
+                    name="civdef.alert.issue",
+                    version=(1, 0),
+                    stability="experimental",
+                    trust_required="trusted",
+                ),
+                self.handle_issue,
+                None,
+            ),
+            (
+                CapabilityDescriptor(
+                    name="civdef.alert.list",
+                    version=(1, 0),
+                    stability="experimental",
+                    idempotent=True,
+                ),
+                self.handle_list,
+                None,
+            ),
+            (
+                CapabilityDescriptor(
+                    name="civdef.cert.verify",
+                    version=(1, 0),
+                    stability="experimental",
+                    idempotent=True,
+                ),
+                self.handle_verify,
+                None,
+            ),
+            (
+                CapabilityDescriptor(
+                    name="civdef.audit.export",
+                    version=(1, 0),
+                    stability="experimental",
+                    idempotent=True,
+                ),
+                self.handle_audit,
+                None,
+            ),
+        ]
+
+    def register(self, bus: Any) -> None:
+        self._bus = bus
+        for cap, handler, predicate in self.capabilities():
+            bus.register_capability(cap, handler, predicate)
+
+    @staticmethod
+    def _alert_to_dict(alert: Alert) -> dict[str, Any]:
+        return {
+            "alert_id": alert.alert_id,
+            "severity": alert.severity,
+            "title": alert.title,
+            "body": alert.body,
+            "area": alert.area_description,
+            "issuer_node_id": alert.issuer_node_id,
+            "community_id": alert.community_id,
+            "issued_at": alert.issued_at,
+            "expires_at": alert.expires_at,
+        }
+
+    async def handle_issue(self, req: Any) -> dict:
+        inp = req.body.get("input", {})
+        title = str(inp.get("title", ""))
+        body = str(inp.get("body", ""))
+        area = str(inp.get("area", ""))
+        if not title or not area:
+            return {"error": "bad_request", "message": "title and area are required"}
+        alert = self.issue_alert(
+            severity=str(inp.get("severity", AlertSeverity.WARNING)),
+            title=title,
+            body=body,
+            area=area,
+            role_cert_id=inp.get("role_cert_id"),
+            community_id=str(inp.get("community_id", "")),
+            expires_in_hours=inp.get("expires_in_hours", 24.0),
+        )
+        return {"output": {"alert": self._alert_to_dict(alert)}, "meta": {}}
+
+    async def handle_list(self, req: Any) -> dict:
+        alerts = [self._alert_to_dict(a) for a in self.list_active_alerts()]
+        return {"output": {"alerts": alerts}, "meta": {"count": len(alerts)}}
+
+    async def handle_verify(self, req: Any) -> dict:
+        cert_id = str(req.body.get("input", {}).get("cert_id", ""))
+        return {"output": self.verify_cert(cert_id), "meta": {}}
+
+    async def handle_audit(self, req: Any) -> dict:
+        return {"output": self.export_audit(), "meta": {}}
+

hearthnet/evidence/service.py

@@ -0,0 +1,185 @@
+"""M30 — Evidence Graph bus service (experimental, Phase 3).
+
+Wraps the real in-memory :class:`ClaimStore` as capability-bus handlers so the
+content-addressed claim graph is reachable over the mesh. Registered only when a
+node opts into research features (``install_extended_services(research=True)``).
+
+Capabilities:
+  evidence.claim.add@1.0      — assert a claim, returns its content-addressed id
+  evidence.claim.attest@1.0   — vouch for an existing claim
+  evidence.claim.dispute@1.0  — dispute an existing claim
+  evidence.claim.find@1.0     — list claims about a subject
+  evidence.summary@1.0        — store statistics
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from hearthnet.bus.capability import CapabilityDescriptor, RouteRequest
+from hearthnet.evidence.store import (
+    Attestation,
+    Claim,
+    ClaimID,
+    ClaimSource,
+    ClaimStore,
+    Dispute,
+    SourceID,
+)
+
+
+class EvidenceService:
+    name = "evidence"
+    version = "1.0"
+
+    def __init__(self, community_id: str = "", store: ClaimStore | None = None) -> None:
+        self._community_id = community_id
+        self._store = store or ClaimStore()
+
+    def capabilities(self) -> list[tuple]:
+        return [
+            (
+                CapabilityDescriptor(
+                    name="evidence.claim.add",
+                    version=(1, 0),
+                    stability="experimental",
+                    trust_required="trusted",
+                    idempotent=True,
+                ),
+                self.handle_add,
+                None,
+            ),
+            (
+                CapabilityDescriptor(
+                    name="evidence.claim.attest",
+                    version=(1, 0),
+                    stability="experimental",
+                    trust_required="trusted",
+                ),
+                self.handle_attest,
+                None,
+            ),
+            (
+                CapabilityDescriptor(
+                    name="evidence.claim.dispute",
+                    version=(1, 0),
+                    stability="experimental",
+                    trust_required="trusted",
+                ),
+                self.handle_dispute,
+                None,
+            ),
+            (
+                CapabilityDescriptor(
+                    name="evidence.claim.find",
+                    version=(1, 0),
+                    stability="experimental",
+                    idempotent=True,
+                ),
+                self.handle_find,
+                None,
+            ),
+            (
+                CapabilityDescriptor(
+                    name="evidence.summary",
+                    version=(1, 0),
+                    stability="experimental",
+                    idempotent=True,
+                ),
+                self.handle_summary,
+                None,
+            ),
+        ]
+
+    def register(self, bus: Any) -> None:
+        for cap, handler, predicate in self.capabilities():
+            bus.register_capability(cap, handler, predicate)
+
+    # ── Handlers ───────────────────────────────────────────────────────────
+
+    async def handle_add(self, req: RouteRequest) -> dict:
+        inp = req.body.get("input", {})
+        subject = str(inp.get("subject", ""))
+        predicate = str(inp.get("predicate", "asserts"))
+        object_ = str(inp.get("object", ""))
+        if not subject or not object_:
+            return {"error": "bad_request", "message": "subject and object are required"}
+        sources = tuple(
+            ClaimSource(
+                source_id=SourceID(str(s.get("source_id", ""))),
+                source_type=str(s.get("source_type", "manual")),
+                url=s.get("url"),
+                reliability_score=float(s.get("reliability_score", 1.0)),
+            )
+            for s in inp.get("sources", [])
+        )
+        claim = Claim(
+            claim_id=ClaimID(""),  # replaced by content_id() inside add_claim
+            subject=subject,
+            predicate=predicate,
+            object_=object_,
+            asserted_by=str(req.caller or "unknown"),
+            sources=sources,
+            community_id=self._community_id,
+            confidence=float(inp.get("confidence", 1.0)),
+        )
+        cid = self._store.add_claim(claim)
+        return {"output": {"claim_id": cid}, "meta": {}}
+
+    async def handle_attest(self, req: RouteRequest) -> dict:
+        inp = req.body.get("input", {})
+        claim_id = ClaimID(str(inp.get("claim_id", "")))
+        if self._store.get_claim(claim_id) is None:
+            return {"error": "not_found", "message": "unknown claim_id"}
+        self._store.attest(
+            Attestation(claim_id=claim_id, attested_by=str(req.caller or "unknown"))
+        )
+        return {
+            "output": {
+                "claim_id": claim_id,
+                "attestations": self._store.attestation_count(claim_id),
+            },
+            "meta": {},
+        }
+
+    async def handle_dispute(self, req: RouteRequest) -> dict:
+        inp = req.body.get("input", {})
+        claim_id = ClaimID(str(inp.get("claim_id", "")))
+        if self._store.get_claim(claim_id) is None:
+            return {"error": "not_found", "message": "unknown claim_id"}
+        counter = inp.get("counter_claim_id")
+        self._store.dispute(
+            Dispute(
+                claim_id=claim_id,
+                disputed_by=str(req.caller or "unknown"),
+                reason=str(inp.get("reason", "")),
+                counter_claim_id=ClaimID(str(counter)) if counter else None,
+            )
+        )
+        return {"output": {"claim_id": claim_id, "disputed": True}, "meta": {}}
+
+    async def handle_find(self, req: RouteRequest) -> dict:
+        inp = req.body.get("input", {})
+        subject = str(inp.get("subject", ""))
+        claims = self._store.find_by_subject(subject)
+        return {
+            "output": {
+                "claims": [
+                    {
+                        "claim_id": c.content_id(),
+                        "subject": c.subject,
+                        "predicate": c.predicate,
+                        "object": c.object_,
+                        "asserted_by": c.asserted_by,
+                        "confidence": c.confidence,
+                        "attestations": self._store.attestation_count(c.content_id()),
+                        "disputed": self._store.is_disputed(c.content_id()),
+                    }
+                    for c in claims
+                ]
+            },
+            "meta": {"count": len(claims)},
+        }
+
+    async def handle_summary(self, req: RouteRequest) -> dict:
+        return {"output": self._store.summary(), "meta": {}}

hearthnet/node.py

@@ -32,6 +32,45 @@ _log = logging.getLogger(__name__)
 _GOSSIP_INTERVAL_SECONDS = 30
 
 
+class _HttpxSyncClient:
+    """Minimal httpx adapter for :class:`SyncClient`.
+
+    SyncClient treats a dict response as already-parsed JSON, so we return the
+    decoded body directly from ``get``/``post`` (avoiding SyncClient's
+    aiohttp-style ``await resp.json()`` path). Degrades to a no-op when httpx is
+    not installed.
+    """
+
+    def __init__(self) -> None:
+        self._client: Any = None
+        self.unavailable = False
+        try:
+            import httpx
+
+            self._client = httpx.AsyncClient(timeout=30.0)
+        except ImportError:
+            self.unavailable = True
+
+    async def get(self, url: str) -> dict[str, Any]:
+        resp = await self._client.get(url)
+        resp.raise_for_status()
+        return resp.json()
+
+    async def post(
+        self, url: str, *, data: Any = None, headers: dict[str, str] | None = None
+    ) -> dict[str, Any]:
+        resp = await self._client.post(url, content=data, headers=headers)
+        resp.raise_for_status()
+        return resp.json()
+
+    async def aclose(self) -> None:
+        if self._client is not None:
+            try:
+                await self._client.aclose()
+            except Exception:
+                pass
+
+
 @dataclass
 class NodeManifest:
     node_id: NodeID
@@ -258,9 +297,124 @@ class HearthNode:
         for service in services:
             self.bus.register_service(service)
 
-    # ------------------------------------------------------------------
-    # 15-step startup / shutdown
-    # ------------------------------------------------------------------
+        # Register the real auxiliary services (embed/rerank/ocr/translation/
+        # speech/image). Phase-3 research services stay off unless opted in.
+        self.install_extended_services(research=False)
+
+    def install_extended_services(
+        self,
+        *,
+        research: bool = False,
+        embed_model: str = "BAAI/bge-small-en-v1.5",
+    ) -> None:
+        """Register the real auxiliary services beyond the core set.
+
+        Always (each degrades gracefully to an "unavailable" response when its
+        optional backend/model is missing — never a mock):
+          M11 EmbeddingService   embed.text      (real semantic vectors when
+                                                  sentence-transformers present)
+          M24 RerankService      rerank.text
+          M17 OcrService         ocr.image / ocr.pdf
+          M18 TranslationService trans.text
+          M19 Stt/TtsService     stt.transcribe / tts.speak
+          M20 Image services     image.describe / image.generate
+
+        When ``research=True`` (opt-in; the demo Space enables it), also registers
+        the real Phase-3 services:
+          M30 EvidenceService      evidence.claim.*
+          M31 CivilDefenseService  civdef.*
+
+        Every registration is wrapped so a missing optional dependency can never
+        break node startup.
+        """
+
+        def _register(svc: Any) -> None:
+            if hasattr(svc, "capabilities"):
+                self.bus.register_service(svc)
+            elif hasattr(svc, "register"):
+                svc.register(self.bus)
+
+        # ── M11 Embedding (core for real RAG) ──────────────────────────────
+        try:
+            import importlib.util
+
+            from hearthnet.services.embedding.service import EmbeddingService
+
+            backend = None
+            if importlib.util.find_spec("sentence_transformers") is not None:
+                from hearthnet.services.embedding.backends import (
+                    SentenceTransformerBackend,
+                )
+
+                backend = SentenceTransformerBackend(model=embed_model)
+            _register(EmbeddingService(backend=backend))
+        except Exception as exc:
+            _log.warning("EmbeddingService registration skipped: %s", exc)
+
+        # ── Remaining always-on auxiliary services ─────────────────────────
+        _aux: list[tuple[str, Any]] = []
+        try:
+            from hearthnet.services.rerank.service import RerankService
+
+            _aux.append(("rerank", RerankService()))
+        except Exception as exc:
+            _log.debug("RerankService unavailable: %s", exc)
+        try:
+            from hearthnet.services.ocr.service import OcrService
+
+            _aux.append(("ocr", OcrService()))
+        except Exception as exc:
+            _log.debug("OcrService unavailable: %s", exc)
+        try:
+            from hearthnet.services.translation.service import TranslationService
+
+            _aux.append(("translation", TranslationService()))
+        except Exception as exc:
+            _log.debug("TranslationService unavailable: %s", exc)
+        try:
+            from hearthnet.services.speech.stt_service import SttService
+            from hearthnet.services.speech.tts_service import TtsService
+
+            _aux.append(("stt", SttService()))
+            _aux.append(("tts", TtsService()))
+        except Exception as exc:
+            _log.debug("Speech services unavailable: %s", exc)
+        try:
+            from hearthnet.services.image.describe_service import ImageDescribeService
+
+            _aux.append(("image.describe", ImageDescribeService()))
+        except Exception as exc:
+            _log.debug("ImageDescribeService unavailable: %s", exc)
+        try:
+            from hearthnet.services.image.generate_service import ImageGenerateService
+
+            _aux.append(("image.generate", ImageGenerateService()))
+        except Exception as exc:
+            _log.debug("ImageGenerateService unavailable: %s", exc)
+
+        for label, svc in _aux:
+            try:
+                _register(svc)
+            except Exception as exc:
+                _log.warning("%s registration skipped: %s", label, exc)
+
+        if not research:
+            return
+
+        # ── Phase-3 research services (opt-in only) ────────────────────────
+        try:
+            from hearthnet.evidence.service import EvidenceService
+
+            _register(EvidenceService(community_id=self.community_id))
+        except Exception as exc:
+            _log.warning("EvidenceService registration skipped: %s", exc)
+        try:
+            from hearthnet.civdef.service import CivilDefenseService
+
+            _register(CivilDefenseService())
+        except Exception as exc:
+            _log.warning("CivilDefenseService registration skipped: %s", exc)
+
 
     async def start(
         self,
@@ -340,9 +494,7 @@ class HearthNode:
 
         # ── Step 8: Emergency detector ────────────────────────────────
         try:
-            self._emergency_task = asyncio.create_task(
-                self.detector.run(), name="emergency-detector"
-            )
+            await self.detector.start()
         except Exception as exc:
             _log.warning("Emergency detector start failed (non-fatal): %s", exc)
 
@@ -474,32 +626,39 @@ class HearthNode:
     async def _gossip_loop(self, interval: int) -> None:
         """Periodically sync event log with all known peers (X02 gossip)."""
         from hearthnet.events.sync import SyncClient
-        from hearthnet.transport.client import HttpClient
 
-        http_client = HttpClient(self.node_id, self.community_id)
+        http_client = _HttpxSyncClient()
+        if http_client.unavailable:
+            _log.info("Gossip sync disabled: httpx not installed")
+            return
         sync_client = SyncClient(self._event_log, http_client)
 
-        while True:
-            await asyncio.sleep(interval)
-            for peer in self.peers.all():
-                if not peer.endpoints:
-                    continue
-                ep = peer.endpoints[0]
-                if ep.transport == "memory":
-                    continue  # in-process; no HTTP needed
-                peer_url = f"http://{ep.host}:{ep.port}"
-                try:
-                    result = await sync_client.sync_with(peer_url, self.community_id)
-                    if result.received_count or result.sent_count:
+        try:
+            while True:
+                await asyncio.sleep(interval)
+                for peer in self.peers.all():
+                    if not peer.endpoints:
+                        continue
+                    ep = peer.endpoints[0]
+                    if ep.transport == "memory":
+                        continue  # in-process; no HTTP needed
+                    peer_url = f"http://{ep.host}:{ep.port}"
+                    try:
+                        result = await sync_client.sync_with(peer_url, self.community_id)
+                        if result.received_count or result.sent_count:
+                            _log.debug(
+                                "Gossip with %s: sent=%d recv=%d ms=%d",
+                                peer.display_name,
+                                result.sent_count,
+                                result.received_count,
+                                result.duration_ms,
+                            )
+                    except Exception as exc:
                         _log.debug(
-                            "Gossip with %s: sent=%d recv=%d ms=%d",
-                            peer.display_name,
-                            result.sent_count,
-                            result.received_count,
-                            result.duration_ms,
+                            "Gossip sync with %s failed: %s", peer.display_name, exc
                         )
-                except Exception as exc:
-                    _log.debug("Gossip sync with %s failed: %s", peer.display_name, exc)
+        finally:
+            await http_client.aclose()
 
     async def _state_bus_to_pubsub(self) -> None:
         """Forward StateBus state changes to the WebSocket pubsub (X06)."""

requirements-dev.txt

@@ -4,4 +4,5 @@ bandit[toml]>=1.7.10,<2.0
 mypy>=1.10,<2.0
 pylint>=3.2,<4.0
 pytest>=8.5.0,<9.0
+pytest-asyncio>=0.24,<1.0
 ruff>=0.5.0,<1.0

requirements.txt

@@ -8,3 +8,5 @@ sentencepiece>=0.2.0
 torch==2.11.0
 transformers>=4.45.0
 qrcode[svg]>=7.4
+sentence-transformers>=3.0.0
+httpx>=0.27.0

tests/test_extended_services.py

@@ -0,0 +1,66 @@
+"""install_extended_services registers the real auxiliary services."""
+
+from __future__ import annotations
+
+import asyncio
+
+from hearthnet.node import HearthNode
+
+
+def _node() -> HearthNode:
+    node = HearthNode("ed25519:ext", "Ext", "ed25519:test-community")
+    node.install_extended_services(research=False)
+    return node
+
+
+def test_extended_services_register() -> None:
+    caps = {e.descriptor.name for e in _node().bus.registry.all_local()}
+    for cap in ("embed.text", "rerank.text", "ocr.image", "trans.text"):
+        assert cap in caps, f"missing {cap}"
+
+
+def test_embed_text_returns_vectors() -> None:
+    """embed.text must return one vector per input, even without ML deps
+    (SimpleHashBackend fallback) — no mock, real deterministic vectors."""
+    node = _node()
+
+    async def _run() -> dict:
+        return await node.bus.call(
+            "embed.text", (1, 0), {"input": {"texts": ["hello", "world"]}}
+        )
+
+    out = asyncio.run(_run())
+    embeddings = out["output"]["embeddings"]
+    assert len(embeddings) == 2
+    assert all(isinstance(v, (int, float)) for v in embeddings[0])
+
+
+def test_rag_query_uses_registered_embedder() -> None:
+    """With embed.text registered, rag.query routes embeddings through the bus."""
+    from hearthnet.services.rag.service import RagService
+
+    node = _node()
+    rag = RagService(corpus="community", bus=node.bus)
+    node.bus.register_service(rag)
+
+    async def _run() -> dict:
+        await node.bus.call(
+            "rag.ingest",
+            (1, 0),
+            {
+                "input": {
+                    "corpus": "community",
+                    "documents": [
+                        {"id": "d1", "title": "Water", "text": "Boil rainwater before drinking."}
+                    ],
+                }
+            },
+        )
+        return await node.bus.call(
+            "rag.query",
+            (1, 0),
+            {"params": {"corpus": "community"}, "input": {"query": "water", "k": 1}},
+        )
+
+    out = asyncio.run(_run())
+    assert out["output"]["chunks"]

tests/test_gossip_sync.py

@@ -0,0 +1,57 @@
+"""Regression: _gossip_loop builds a working httpx adapter (not a broken HttpClient)."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+
+from hearthnet.node import _HttpxSyncClient
+
+
+def test_httpx_sync_client_constructs() -> None:
+    client = _HttpxSyncClient()
+    # httpx is a declared dependency, so the adapter must be available.
+    assert client.unavailable is False
+    asyncio.run(client.aclose())
+
+
+def test_sync_client_accepts_adapter() -> None:
+    """SyncClient must accept the adapter without raising at construction."""
+    from hearthnet.events.sync import SyncClient
+
+    class _FakeLog:
+        def head(self) -> int:
+            return 0
+
+        def since(self, _n: int):
+            return []
+
+        def append_received(self, _e) -> bool:
+            return False
+
+    adapter = _HttpxSyncClient()
+    sync = SyncClient(_FakeLog(), adapter)
+    assert sync is not None
+    asyncio.run(adapter.aclose())
+
+
+def test_gossip_loop_no_peers_is_safe() -> None:
+    """A node with no peers must run the gossip loop body without raising."""
+    from hearthnet.node import HearthNode
+
+    node = HearthNode("ed25519:g", "G", "ed25519:test-community")
+
+    class _Log:
+        def head(self) -> int:
+            return 0
+
+    node._event_log = _Log()
+
+    async def _run() -> None:
+        task = asyncio.create_task(node._gossip_loop(interval=0))
+        await asyncio.sleep(0.05)
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+
+    asyncio.run(_run())

tests/test_phase3_services.py

@@ -0,0 +1,91 @@
+"""Phase 3 research services (M30 Evidence, M31 Civil Defense) — real impls."""
+
+from __future__ import annotations
+
+import asyncio
+
+from hearthnet.node import HearthNode
+
+
+def _research_node() -> HearthNode:
+    node = HearthNode("ed25519:research", "Research", "ed25519:test-community")
+    node.install_extended_services(research=True)
+    return node
+
+
+def test_research_services_register_only_when_opted_in() -> None:
+    plain = HearthNode("ed25519:plain", "Plain", "ed25519:test-community")
+    plain.install_extended_services(research=False)
+    plain_caps = {e.descriptor.name for e in plain.bus.registry.all_local()}
+    assert "evidence.claim.add" not in plain_caps
+    assert "civdef.alert.issue" not in plain_caps
+
+    node = _research_node()
+    caps = {e.descriptor.name for e in node.bus.registry.all_local()}
+    for cap in (
+        "evidence.claim.add",
+        "evidence.claim.attest",
+        "evidence.claim.find",
+        "civdef.alert.issue",
+        "civdef.alert.list",
+        "civdef.audit.export",
+    ):
+        assert cap in caps
+
+
+def test_evidence_claim_roundtrip() -> None:
+    node = _research_node()
+
+    async def _run() -> dict:
+        add = await node.bus.call(
+            "evidence.claim.add",
+            (1, 0),
+            {
+                "input": {
+                    "subject": "well:village-1",
+                    "predicate": "water_status",
+                    "object": "potable",
+                }
+            },
+        )
+        claim_id = add["output"]["claim_id"]
+        await node.bus.call(
+            "evidence.claim.attest", (1, 0), {"input": {"claim_id": claim_id}}
+        )
+        return await node.bus.call(
+            "evidence.claim.find", (1, 0), {"input": {"subject": "well:village-1"}}
+        )
+
+    found = asyncio.run(_run())
+    claims = found["output"]["claims"]
+    assert len(claims) == 1
+    assert claims[0]["object"] == "potable"
+    assert claims[0]["attestations"] == 1
+
+
+def test_civdef_alert_and_audit_chain() -> None:
+    node = _research_node()
+
+    async def _run() -> tuple[dict, dict]:
+        await node.bus.call(
+            "civdef.alert.issue",
+            (1, 0),
+            {
+                "input": {
+                    "severity": "warning",
+                    "title": "Boil water notice",
+                    "body": "Boil tap water before drinking.",
+                    "area": "Issum, Kreis Kleve, NRW",
+                }
+            },
+        )
+        listed = await node.bus.call("civdef.alert.list", (1, 0), {"input": {}})
+        audit = await node.bus.call("civdef.audit.export", (1, 0), {"input": {}})
+        return listed, audit
+
+    listed, audit = asyncio.run(_run())
+    assert len(listed["output"]["alerts"]) == 1
+    assert listed["output"]["alerts"][0]["title"] == "Boil water notice"
+    # Tamper-evident audit chain must verify.
+    assert audit["output"]["chain_valid"] is True
+    assert audit["output"]["length"] >= 1