feat: 15 targeted improvements — RAG persistence, bus failover, agent hardening, deps sync

RAG / storage
- CorpusStore: chroma → SQLite → in-memory fallback; mkdir unconditional;
WARNING log at startup shows active backend
- RagService: _log.warning on blob_store.put / event_log.append_local failures
instead of silent pass
- Expose corpus stats (backend, persistent, chunks) in Settings tab
- search_corpus tool bound to rag.federated_query (scatter-gather path)
- corpus param plumbed into body["params"] so router _corpus_matches fires

Agent / LLM tools
- Brace-matching JSON parser (_extract_json_object) replaces fragile {.*?} regex
- max_iterations default lowered 6→4; one-shot worked example in system_prompt

Bus
- Failover to _best_alternative when sole provider is quarantined (route→None)
- CapabilityDescriptor.schema_hash prefix corrected: "blake3:" → "sha256:"

Boot / deps
- MoE expert self-registers after seed corpus thread completes
- requirements.txt: add blake3>=0.4.0 and click>=8.1 (were in pyproject.toml only)
- scripts/check_deps_sync.py: CI script to keep the two files in sync

Docs / cleanup
- README: architecture diagram SQLite, M05 description updated; token
signature gap documented in security section
- hearthnet/services/file/ deleted (dead code; only plural files/ is imported)
- transport/server.py: remove duplicate UTC=UTC assignment

Tests
- 13 new passing tests in tests/test_improvements_batch.py and
tests/test_federated_rag.py (SQLite temp-dir isolation + Windows lock fix)

README.md

@@ -47,7 +47,7 @@ license: apache-2.0
   <img src="https://img.shields.io/badge/OpenBMB-MiniCPM%20multi--model-1f6feb" alt="OpenBMB">
 </p>
 
-> **Build Small Hackathon entry** — Backyard AI track · 🐜 Tiny Titan · 🤖 Best Agent
+> **Build Small Hackathon entry** — Backyard AI track · 🐜 Tiny Titan · 🤖 Best Agent 🫥 press e or a to see the easter egg.
 >
 > 📺 **Demo video:** <a href="https://huggingface.co/spaces/build-small-hackathon/HearthNet/resolve/main/hf_hackathon_screenrecording_v1.webm">HF Space Recording</a> · <a href="https://videos.simpleshow.com/8vSfxilim8">Simple Show Demo</a>
 
@@ -57,7 +57,8 @@ license: apache-2.0
   Your browser does not support the video tag.
 </video>
 
-> 📣 **Social post:** *(many)*
+> 📣 **Social post:** [tweet on x](https://twitter.com/zX14_7/status/2064853015622775047) [tweet on x](https://twitter.com/zX14_7/status/2064853015622775047) 
+
 >
 > **June 14 bug-fix release:** 8 critical bugs fixed — seed corpus now actually ingested,
 > node lifecycle corrected (`stop()` previously silently no-oped), sticky session memory
@@ -86,7 +87,7 @@ intelligent routing bus, and work **completely offline**. When the internet is a
 
 ## Features
 
-### � Agent Mode (ReAct tool calling)
+###  Agent Mode (ReAct tool calling)
 Flip the **Agent mode** toggle in the Ask tab and the model stops being a chatbot and starts being an **agent**: it plans, calls real mesh tools over several steps, reads the results, and only then answers. Every step is shown live — **Thought → Tool → Observation → Answer**.
 
 The agent's tools are bound to **real capabilities already on the bus** (no mock handlers):
@@ -94,7 +95,7 @@ The agent's tools are bound to **real capabilities already on the bus** (no mock
 
 > 💡 **Try the browser agent:** press **`a`** (or just type **`hearthnet`**) anywhere on the dashboard to open the in-browser WebLLM agent showcase. Press **`e`** for the live mesh/news ticker, **`Esc`** to close.
 
-### �🧠 Intelligent Routing (NEW)
+### 🧠 Intelligent Routing (NEW)
 When you ask a question, the bus scores available LLM nodes by latency, load, and reliability. Your request goes to the **best node right now** — whether it's local, your neighbour's device, or a peer across the internet. Failover is automatic: if the preferred node can't help, the next-best provider takes over **invisibly**.
 
 **Routing Trace** shows you exactly where your request was served:
@@ -381,6 +382,8 @@ If no suitable backend is available: clear error message returned. Never silent,
 - **X3DH + Double Ratchet** — end-to-end encrypted chat (M23)
 - **BLAKE3** — content-addressed file blobs (tamper-evident)
 - **localhost-only CLI** — all admin HTTP restricted to 127.0.0.1
+- **Capability token `exp` claim** — checked in `bus.handle_call()` before routing; expired tokens receive `{"error": "token_expired"}` without hitting any handler
+- **Token signature verification** — Ed25519 signature checking is implemented in `AuthService` (`auth.token.verify`) and is available on the bus. The HTTP transport (`/bus/v1/call`) currently passes tokens to `handle_call()` where expiry is enforced; full per-request signature verification on inbound HTTP calls is a planned hardening step.
 - **Bandit HIGH findings: 0** (verified in CI)
 
 ---
@@ -402,7 +405,7 @@ If no suitable backend is available: clear error message returned. Never silent,
        ┌──────────▼┐  ┌──▼───┐ ┌▼──────────┐  ┌────────────┐
        │ LLM (M04) │  │ RAG  │ │ MoE (M27) │  │ Chat (M10) │
        │llama.cpp  │  │(M05) │ │ Expert    │  │ Marketplace│
-       │ Ollama    │  │Chroma│ │ Registry  │  │ (M06) Files│
+       │ Ollama    │  │SQLite│ │ Registry  │  │ (M06) Files│
        │HF Transfm │  │Embed │ └───────────┘  └────────────┘
        └─────┬─────┘  └──┬───┘
              └─────┬──────┘
@@ -426,7 +429,7 @@ If no suitable backend is available: clear error message returned. Never silent,
 | M02 | Peer discovery (mDNS, UDP broadcast, PeerRegistry) | ✅ |
 | M03 | Capability bus (schema validation, routing, tracing) | ✅ |
 | M04 | LLM service (llama.cpp, Ollama, HF Transformers, cloud fallback) | ✅ |
-| M05 | RAG (chunker, ChromaDB, IngestPipeline, semantic search) | ✅ |
+| M05 | RAG (chunker, SQLite/ChromaDB vector store, IngestPipeline, federated scatter-gather) | ✅ |
 | M06 | Marketplace (event-sourced, Lamport-clocked posts) | ✅ |
 | M07 | File blobs (BLAKE3 hash, content-addressed, chunked transfer) | ✅ |
 | M08 | Gradio UI (8 tabs: Ask, Chat, Mesh, Marketplace, Files, Emergency, Settings, Getting Started) | ✅ |

app.py

@@ -450,6 +450,36 @@ def _build_node():
     _seed_thread.start()
     _seed_thread.join(timeout=60)  # wait up to 60 s; don't block Space startup indefinitely
 
+    # Register this node's LLM model as an expert in the MoE registry so
+    # route_expert tool calls return meaningful results instead of an empty list.
+    try:
+        _moe_tags = list({
+            doc.get("id", "").split(".")[0]
+            for doc in SEED_CORPUS
+            if doc.get("id")
+        } | {"emergency", "mesh", "community"})
+        loop_moe = asyncio.new_event_loop()
+        loop_moe.run_until_complete(
+            node.bus.call(
+                "moe.register",
+                (1, 0),
+                {
+                    "input": {
+                        "expert_id": f"model:{MODEL_ID}",
+                        "expert_type": "model",
+                        "topic_tags": _moe_tags,
+                        "confidence_score": 0.6,
+                        "community_id": node.community_id,
+                        "name": MODEL_ID.split("/")[-1],
+                        "ttl_seconds": 0,
+                    }
+                },
+            )
+        )
+        loop_moe.close()
+    except Exception:
+        pass
+
     # 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, bus=node.bus))

hearthnet/bus/__init__.py

@@ -143,6 +143,13 @@ class CapabilityBus:
 
         entry = self.router.route_sticky(req) if req.session_id else self.router.route(req)
         if entry is None:
+            # No direct route — try any alternative before giving up.
+            # Covers the quarantined-sole-provider case: route() skips quarantined
+            # entries, but _best_alternative can still find an unquarantined remote.
+            alternative = self._best_alternative(req, exclude=set())
+            if alternative is not None:
+                result = await self._execute_entry(alternative, req, local_only)
+                return self._stamp_route(result, alternative, local_only)
             raise BusError("not_found", f"no provider for {req.capability}@{req.version_req}")
         result = await self._execute_entry(entry, req, local_only)
 

hearthnet/bus/capability.py

@@ -45,7 +45,7 @@ class CapabilityDescriptor:
             "response_schema": self.response_schema,
             "stream_schema": self.stream_schema,
         }
-        return "blake3:" + hashlib.sha256(_canonical_json(payload)).hexdigest()
+        return "sha256:" + hashlib.sha256(_canonical_json(payload)).hexdigest()
 
 
 @dataclass

hearthnet/services/file/__init__.py

@@ -1,5 +0,0 @@
-from __future__ import annotations
-
-from hearthnet.services.file.service import FileService
-
-__all__ = ["FileService"]

hearthnet/services/file/service.py

@@ -1,108 +0,0 @@
-from __future__ import annotations
-
-import base64
-from typing import Any
-
-from hearthnet.blobs.store import BlobStore
-from hearthnet.bus.capability import CapabilityDescriptor, RouteRequest
-
-
-class FileService:
-    name = "file"
-    version = "1.0"
-
-    def __init__(self, store: BlobStore) -> None:
-        self.store = store
-
-    def capabilities(self) -> list[tuple[Any, ...]]:
-        return [
-            (
-                CapabilityDescriptor(
-                    name="file.read",
-                    params={},
-                    trust_required="member",
-                    max_concurrent=8,
-                ),
-                self.handle_read,
-                None,
-            ),
-            (
-                CapabilityDescriptor(
-                    name="file.list",
-                    params={},
-                    trust_required="member",
-                    max_concurrent=8,
-                ),
-                self.handle_list,
-                None,
-            ),
-            (
-                CapabilityDescriptor(
-                    name="file.advertise",
-                    params={},
-                    trust_required="member",
-                    max_concurrent=4,
-                ),
-                self.handle_advertise,
-                None,
-            ),
-            (
-                CapabilityDescriptor(
-                    name="file.put",
-                    params={},
-                    trust_required="trusted",
-                    max_concurrent=2,
-                    timeout_seconds=600,
-                ),
-                self.handle_put,
-                None,
-            ),
-        ]
-
-    async def handle_read(self, req: RouteRequest) -> dict[str, Any]:
-        """input: {cid: str} → output: {cid, size_bytes, filename, chunks: [...]}"""
-        cid = req.body.get("input", {}).get("cid", "")
-        if not self.store.has(cid):
-            return {"error": "not_found", "message": f"Blob {cid} not found"}
-        manifest = self.store.get_manifest(cid)
-        return {
-            "output": {
-                "cid": manifest.cid,
-                "size_bytes": manifest.size_bytes,
-                "filename": manifest.filename,
-                "chunks": [
-                    {"index": c.index, "cid": c.cid, "size_bytes": c.size_bytes}
-                    for c in manifest.chunks
-                ],
-            },
-            "meta": {},
-        }
-
-    async def handle_list(self, req: RouteRequest) -> dict[str, Any]:
-        blobs = self.store.list_blobs()
-        return {
-            "output": {
-                "blobs": [
-                    {"cid": b.cid, "size_bytes": b.size_bytes, "filename": b.filename}
-                    for b in blobs
-                ]
-            },
-            "meta": {},
-        }
-
-    async def handle_advertise(self, req: RouteRequest) -> dict[str, Any]:
-        """input: {cid, filename, size_bytes} → acknowledge, actual transfer is separate"""
-        inp = req.body.get("input", {})
-        return {"output": {"acknowledged": True, "cid": inp.get("cid")}, "meta": {}}
-
-    async def handle_put(self, req: RouteRequest) -> dict[str, Any]:
-        """input: {data_b64: str, filename: str} → store blob → output: {cid, size_bytes}"""
-        inp = req.body.get("input", {})
-        data_b64 = inp.get("data_b64", "")
-        filename = inp.get("filename")
-        try:
-            data = base64.b64decode(data_b64)
-        except Exception:
-            return {"error": "bad_request", "message": "Invalid base64 data"}
-        manifest = self.store.put(data, filename=filename)
-        return {"output": {"cid": manifest.cid, "size_bytes": manifest.size_bytes}, "meta": {}}

hearthnet/services/llm/tools.py

@@ -193,17 +193,24 @@ class ToolExecutor:
                 )
 
         # 2. Bus-dispatched capability.
-        # NOTE: the HearthNet CapabilityBus API is positional:
-        #   bus.call(capability_name, version_tuple, body_dict)
-        # (see hearthnet/bus and ui/tabs/ask.py). An earlier draft constructed a
-        # RouteRequest and called bus.call(req) — that never matched the real bus
-        # and is why the tool path was never exercised. Use the real API here.
         if definition and definition.bound_capability and self._bus is not None:
             try:
+                args = dict(call.arguments)
+                # Plumb corpus and top_k into body["params"] so the router's
+                # _corpus_matches predicate sees them; leave them in input too
+                # so the handler can read them if it wants.
+                bus_params: dict = {}
+                if "corpus" in args:
+                    bus_params["corpus"] = args["corpus"]
+                if "top_k" in args:
+                    bus_params["top_k"] = args["top_k"]
+                call_body: dict = {"input": args}
+                if bus_params:
+                    call_body["params"] = bus_params
                 resp = await self._bus.call(
                     definition.bound_capability,
                     definition.bound_version or (1, 0),
-                    {"input": call.arguments},
+                    call_body,
                 )
                 if isinstance(resp, dict) and "error" in resp:
                     return ToolResult(
@@ -251,22 +258,26 @@ class ToolExecutor:
     def system_prompt(self) -> str:
         """Build the ReAct system prompt that teaches the model to call tools.
 
-        Mirrors the proven browser-agent format (webagent/src/agent/runtime.js):
-        the model emits a line ``action: {json}`` to call a tool, and receives an
-        ``Observation:`` back. When it has the answer it replies normally with no
-        ``action:`` line. This JSON-in-text protocol works on tiny models that
-        have no native function-calling API (Tiny Titan friendly).
+        One concrete worked example is included because tiny models (SmolLM2,
+        Phi-3-mini) follow few-shot examples far more reliably than abstract
+        format rules. Keep the example short so it fits in context on 135M models.
         """
         return (
             "You are a HearthNet agent. You can use tools to answer questions about "
             "the local mesh, documents, neighbours, and the world.\n\n"
             "Available tools:\n"
             f"{self.tool_help()}\n\n"
-            "To use a tool, output EXACTLY one line:\n"
+            "To use a tool, output EXACTLY one line starting with 'action:':\n"
             'action: {"tool": "<tool_name>", "<arg>": "<value>"}\n'
             "Then stop and wait. You will receive a line starting with 'Observation:'.\n"
             "You may use tools several times. When you have enough information, "
-            "reply to the user directly in plain text with NO 'action:' line.\n"
+            "reply to the user directly in plain text with NO 'action:' line.\n\n"
+            "Example:\n"
+            "User: What do I do if water is cut off?\n"
+            'action: {"tool": "search_corpus", "query": "water supply cut off emergency"}\n'
+            "Observation: Store at least 3 litres per person per day. Boil before drinking.\n"
+            "You should store at least 3 litres of water per person per day and boil it "
+            "before drinking during an outage.\n\n"
             "Keep tool arguments minimal and valid JSON."
         )
 
@@ -276,7 +287,7 @@ class ToolExecutor:
         call_llm: Callable[[list[dict]], Any],
         *,
         history: list[dict] | None = None,
-        max_iterations: int = 6,
+        max_iterations: int = 4,
         on_step: Callable[[dict], Any] | None = None,
     ) -> dict:
         """Run a ReAct tool-use loop and return ``{"final", "steps"}``.
@@ -309,7 +320,10 @@ class ToolExecutor:
                 if asyncio.iscoroutine(res):
                     await res
 
-        action_re = re.compile(r"action\s*:\s*(\{.*?\})", re.IGNORECASE | re.DOTALL)
+        # action_re finds the start of the action: prefix; we then use
+        # _extract_json_object to find the true closing brace so nested objects
+        # and arrays inside tool arguments are captured correctly.
+        action_re = re.compile(r"action\s*:\s*(\{)", re.IGNORECASE)
         final_text = ""
 
         for _ in range(max(1, max_iterations)):
@@ -324,8 +338,15 @@ class ToolExecutor:
 
             await _emit({"type": "thought", "text": text[: match.start()].strip()})
 
+            # Use brace-matching parser instead of non-greedy regex so nested
+            # objects/arrays inside tool arguments are captured in full.
+            brace_start = match.start(1)
+            raw_json = _extract_json_object(text, brace_start)
+            if raw_json is None:
+                raw_json = match.group(1)  # fallback to regex capture
+
             try:
-                action = json.loads(match.group(1))
+                action = json.loads(raw_json)
             except json.JSONDecodeError:
                 chat.append({"role": "assistant", "content": text})
                 chat.append(
@@ -369,7 +390,7 @@ class ToolExecutor:
         raw = await call_llm(chat)
         final_text = (raw if isinstance(raw, str) else str(raw)).strip()
         # Strip any trailing action line the model may still emit.
-        final_text = action_re.sub("", final_text).strip()
+        final_text = re.sub(r"action\s*:\s*\{[^}]*\}", "", final_text).strip()
         await _emit({"type": "final", "text": final_text})
         return {"final": final_text, "steps": steps}
 
@@ -401,6 +422,45 @@ class ToolExecutor:
         return messages
 
 
+# ---------------------------------------------------------------------------
+# JSON brace-matching helper
+# ---------------------------------------------------------------------------
+
+
+def _extract_json_object(text: str, start: int) -> str | None:
+    """Return the JSON object starting at text[start] (must be '{').
+
+    Walks forward counting '{'/'}' while respecting string literals (so braces
+    inside quoted strings don't throw off the count). Returns the full object
+    string including the outer braces, or None if no matching close-brace is
+    found. This replaces the non-greedy ``{.*?}`` regex which truncates at the
+    first '}' and breaks on nested objects or multi-element arrays.
+    """
+    if start >= len(text) or text[start] != "{":
+        return None
+    depth = 0
+    in_string = False
+    escape_next = False
+    i = start
+    while i < len(text):
+        ch = text[i]
+        if escape_next:
+            escape_next = False
+        elif ch == "\\" and in_string:
+            escape_next = True
+        elif ch == '"':
+            in_string = not in_string
+        elif not in_string:
+            if ch == "{":
+                depth += 1
+            elif ch == "}":
+                depth -= 1
+                if depth == 0:
+                    return text[start : i + 1]
+        i += 1
+    return None
+
+
 # ---------------------------------------------------------------------------
 # Default tool set
 # ---------------------------------------------------------------------------
@@ -428,7 +488,7 @@ def default_tool_set(bus: Any) -> ToolExecutor:
                 },
                 "required": ["query"],
             },
-            bound_capability="rag.query",
+            bound_capability="rag.federated_query",
             bound_version=(1, 0),
         ),
         ToolDefinition(

hearthnet/services/rag/service.py

@@ -1,11 +1,14 @@
 from __future__ import annotations
 
+import logging
 from pathlib import Path
 from typing import Any
 
 from hearthnet.bus.capability import CapabilityDescriptor, RouteRequest
 from hearthnet.services.rag.store import CorpusStore, list_corpora
 
+_log = logging.getLogger(__name__)
+
 
 class RagService:
     name = "rag"
@@ -146,8 +149,8 @@ class RagService:
             try:
                 manifest = self._blob_store.put(text.encode("utf-8"), filename=title)
                 blob_cid = manifest.cid
-            except Exception:
-                pass
+            except Exception as exc:
+                _log.warning("RAG blob_store.put failed for '%s': %s", title, exc)
 
         # Emit rag.document.ingested event so peers learn a new doc exists (X02).
         if not result.was_duplicate and self._event_log is not None:
@@ -166,8 +169,8 @@ class RagService:
                     author,
                     payload,
                 )
-            except Exception:
-                pass
+            except Exception as exc:
+                _log.warning("RAG event_log.append_local failed for '%s': %s", title, exc)
 
         return {
             "output": {

hearthnet/services/rag/store.py

@@ -1,11 +1,16 @@
 from __future__ import annotations
 
+import json
+import logging
+import sqlite3
 import uuid
 from dataclasses import dataclass
 from pathlib import Path
 
 from hearthnet.services.rag.chunker import Chunk
 
+_log = logging.getLogger(__name__)
+
 
 @dataclass(frozen=True)
 class ScoredChunk:
@@ -14,9 +19,16 @@ class ScoredChunk:
 
 
 class CorpusStore:
-    """In-memory vector store with cosine similarity.
+    """Persistent vector store — chromadb if available, SQLite otherwise.
+
+    Backend selection (in order of preference):
+      1. chromadb PersistentClient  — if chromadb is installed
+      2. SQLite (one .db file per corpus) — always available, survives restart
+      3. in-memory list              — last resort if SQLite also fails
 
-    Uses chromadb if available, else falls back to in-memory list.
+    The active backend is logged at WARNING level so it is visible in Space logs.
+    ``self._dir.mkdir()`` runs unconditionally so the corpora directory always
+    exists regardless of which backend wins.
     """
 
     def __init__(self, corpora_dir: Path, corpus_name: str) -> None:
@@ -25,21 +37,62 @@ class CorpusStore:
         self._use_chroma = False
         self._chroma_client = None
         self._collection = None
-        # Fallback: in-memory list of (chunk, embedding)
+        self._db: sqlite3.Connection | None = None
+        # Pure in-memory fallback (only used when SQLite init also fails)
         self._items: list[tuple[Chunk, list[float]]] = []
+
+        # Always create the directory — independent of which backend is chosen.
+        self._dir.mkdir(parents=True, exist_ok=True)
+
         self._try_init_chroma()
+        if not self._use_chroma:
+            self._init_sqlite()
+
+        if self._use_chroma:
+            backend = "chroma"
+        elif self._db is not None:
+            backend = "sqlite"
+        else:
+            backend = "in-memory/ephemeral"
+        _log.warning("RAG vector store: using %s backend for corpus '%s'", backend, corpus_name)
+
+    # ------------------------------------------------------------------
+    # Backend initialisation
+    # ------------------------------------------------------------------
 
     def _try_init_chroma(self) -> None:
         try:
             import chromadb  # type: ignore[import-untyped]
 
-            self._dir.mkdir(parents=True, exist_ok=True)
             self._chroma_client = chromadb.PersistentClient(path=str(self._dir / self._corpus))
             self._collection = self._chroma_client.get_or_create_collection(self._corpus)
             self._use_chroma = True
         except ImportError:
             pass
 
+    def _init_sqlite(self) -> None:
+        db_path = self._dir / f"{self._corpus}.db"
+        try:
+            self._db = sqlite3.connect(str(db_path), check_same_thread=False)
+            self._db.execute("""
+                CREATE TABLE IF NOT EXISTS chunks (
+                    id TEXT PRIMARY KEY,
+                    doc_cid TEXT,
+                    chunk_text TEXT NOT NULL,
+                    metadata_json TEXT NOT NULL DEFAULT '{}',
+                    embedding_json TEXT NOT NULL
+                )
+            """)
+            self._db.execute("CREATE INDEX IF NOT EXISTS idx_doc_cid ON chunks(doc_cid)")
+            self._db.commit()
+        except Exception as exc:
+            _log.warning("RAG SQLite init failed, using in-memory fallback: %s", exc)
+            self._db = None
+
+    # ------------------------------------------------------------------
+    # Write path
+    # ------------------------------------------------------------------
+
     def add(self, chunks: list[Chunk], embeddings: list[list[float]]) -> None:
         """Add chunks with their embeddings."""
         if self._use_chroma and self._collection is not None:
@@ -52,10 +105,31 @@ class CorpusStore:
                 documents=documents,
                 metadatas=metadatas,
             )
+        elif self._db is not None:
+            rows = [
+                (
+                    str(uuid.uuid4()),
+                    chunk.metadata.get("doc_cid"),
+                    chunk.text,
+                    json.dumps(dict(chunk.metadata)),
+                    json.dumps(emb),
+                )
+                for chunk, emb in zip(chunks, embeddings, strict=False)
+            ]
+            self._db.executemany(
+                "INSERT INTO chunks(id, doc_cid, chunk_text, metadata_json, embedding_json)"
+                " VALUES (?,?,?,?,?)",
+                rows,
+            )
+            self._db.commit()
         else:
             for chunk, emb in zip(chunks, embeddings, strict=False):
                 self._items.append((chunk, emb))
 
+    # ------------------------------------------------------------------
+    # Read path
+    # ------------------------------------------------------------------
+
     def query(self, embedding: list[float], k: int = 5) -> list[ScoredChunk]:
         """Return top-k chunks by cosine similarity."""
         if self._use_chroma and self._collection is not None:
@@ -70,39 +144,89 @@ class CorpusStore:
             scored: list[ScoredChunk] = []
             docs = results.get("documents", [[]])[0]
             metas = results.get("metadatas", [[]])[0]
-            # chromadb distances are L2 by default; convert to similarity
+            # chromadb distances are L2; convert to similarity score
             distances = results.get("distances", [[]])[0]
             for doc, meta, dist in zip(docs, metas, distances, strict=False):
                 score = 1.0 / (1.0 + dist)
                 scored.append(ScoredChunk(chunk=Chunk(text=doc, metadata=meta), score=score))
             return scored
+
+        # SQLite: load all rows, cosine-rank in Python
+        if self._db is not None:
+            rows = self._db.execute(
+                "SELECT chunk_text, metadata_json, embedding_json FROM chunks"
+            ).fetchall()
+            if not rows:
+                return []
+            scored_items: list[tuple[Chunk, float]] = []
+            for chunk_text, meta_json, emb_json in rows:
+                try:
+                    meta = json.loads(meta_json)
+                    emb = json.loads(emb_json)
+                    score = self._cosine_similarity(embedding, emb)
+                    scored_items.append((Chunk(text=chunk_text, metadata=meta), score))
+                except Exception:
+                    continue
+            scored_items.sort(key=lambda x: x[1], reverse=True)
+            return [ScoredChunk(chunk=c, score=s) for c, s in scored_items[:k]]
+
+        # Pure in-memory fallback
         if not self._items:
             return []
-        scored_items = [
+        mem_scored = [
             (chunk, self._cosine_similarity(embedding, emb)) for chunk, emb in self._items
         ]
-        scored_items.sort(key=lambda x: x[1], reverse=True)
-        return [ScoredChunk(chunk=chunk, score=score) for chunk, score in scored_items[:k]]
+        mem_scored.sort(key=lambda x: x[1], reverse=True)
+        return [ScoredChunk(chunk=chunk, score=score) for chunk, score in mem_scored[:k]]
 
     def has_doc(self, doc_cid: str) -> bool:
         """True if any chunk with this doc_cid exists."""
         if self._use_chroma and self._collection is not None:
             results = self._collection.get(where={"doc_cid": doc_cid}, limit=1, include=[])
             return len(results.get("ids", [])) > 0
+        if self._db is not None:
+            row = self._db.execute(
+                "SELECT 1 FROM chunks WHERE doc_cid = ? LIMIT 1", (doc_cid,)
+            ).fetchone()
+            return row is not None
         return any(c.metadata.get("doc_cid") == doc_cid for c, _ in self._items)
 
     def count(self) -> int:
         if self._use_chroma and self._collection is not None:
             return self._collection.count()
+        if self._db is not None:
+            row = self._db.execute("SELECT COUNT(*) FROM chunks").fetchone()
+            return row[0] if row else 0
         return len(self._items)
 
     def clear(self) -> None:
         if self._use_chroma and self._collection is not None and self._chroma_client is not None:
             self._chroma_client.delete_collection(self._corpus)
             self._collection = self._chroma_client.get_or_create_collection(self._corpus)
+        elif self._db is not None:
+            self._db.execute("DELETE FROM chunks")
+            self._db.commit()
         else:
             self._items.clear()
 
+    def corpus_info(self) -> dict:
+        """Return backend metadata — exposed via Settings tab and node manifest."""
+        if self._use_chroma:
+            backend = "chroma"
+            persistent = True
+        elif self._db is not None:
+            backend = "sqlite"
+            persistent = True
+        else:
+            backend = "in-memory"
+            persistent = False
+        return {
+            "backend": backend,
+            "persistent": persistent,
+            "chunks": self.count(),
+            "corpus": self._corpus,
+        }
+
     def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
         dot = sum(x * y for x, y in zip(a, b, strict=False))
         na = sum(x**2 for x in a) ** 0.5
@@ -114,15 +238,15 @@ def list_corpora(corpora_dir: Path) -> list[str]:
     """List corpus names found under corpora_dir."""
     if not corpora_dir.exists():
         return []
-    return sorted(p.name for p in corpora_dir.iterdir() if p.is_dir())
+    return sorted(p.name for p in corpora_dir.iterdir() if p.is_dir() or p.suffix == ".db")
 
 
 def corpus_info(corpora_dir: Path, corpus: str) -> dict:
-    """Return {corpus, exists, count_chunks}."""
-    corpus_path = corpora_dir / corpus
-    exists = corpus_path.exists()
-    count = 0
+    """Return {corpus, exists, count_chunks, backend, persistent}."""
+    corpus_dir = corpora_dir / corpus
+    db_path = corpora_dir / f"{corpus}.db"
+    exists = corpus_dir.exists() or db_path.exists()
     if exists:
         store = CorpusStore(corpora_dir, corpus)
-        count = store.count()
-    return {"corpus": corpus, "exists": exists, "count_chunks": count}
+        return store.corpus_info()
+    return {"corpus": corpus, "exists": False, "count_chunks": 0, "backend": "none", "persistent": False}

hearthnet/transport/server.py

@@ -22,10 +22,9 @@ from __future__ import annotations
 import asyncio
 from collections.abc import Callable
 from datetime import datetime, timezone as _tz
-UTC = _tz.utc
 from typing import Any
 
-UTC = UTC
+UTC = _tz.utc
 
 try:
     import uvicorn

hearthnet/ui/app.py

@@ -235,7 +235,8 @@ class UiApp:
                 with gr.Tab("Emergency"):
                     build_emergency_tab(self._bus, self._state_bus)
                 with gr.Tab("Settings"):
-                    build_settings_tab(self._config, self._meta, bus=self._bus)
+                    _rag_svc = getattr(self._node, "_rag_service", None)
+                    build_settings_tab(self._config, self._meta, bus=self._bus, rag_service=_rag_svc)
                 with gr.Tab("Getting Started"):
                     build_getting_started_tab()
 

hearthnet/ui/tabs/settings.py

@@ -41,7 +41,7 @@ def _qr_svg(data: str) -> str:
         )
 
 
-def build_settings_tab(config=None, meta: dict | None = None, bus=None):
+def build_settings_tab(config=None, meta: dict | None = None, bus=None, rag_service=None):
     import gradio as gr
 
     meta = meta or {}
@@ -118,6 +118,26 @@ See the **Mesh** tab for a visual graph.
 
             refresh_peers_btn.click(get_peers, outputs=peers_out)
 
+        # --- RAG corpus status -------------------------------------------
+        with gr.Accordion("📚 RAG Knowledge Base", open=True):
+            gr.Markdown("""
+Shows the active vector store backend and how many document chunks are indexed.
+**sqlite** = persists across restarts. **chroma** = best quality. **in-memory** = wiped on restart.
+""")
+            rag_status_out = gr.JSON(label="Corpus status", value={})
+            refresh_rag_btn = gr.Button("🔄 Refresh Corpus Stats", size="sm")
+
+            def get_rag_status():
+                if rag_service is None:
+                    return {"status": "no rag_service wired"}
+                try:
+                    store = rag_service._store
+                    return store.corpus_info()
+                except Exception as exc:
+                    return {"error": str(exc)}
+
+            refresh_rag_btn.click(get_rag_status, outputs=rag_status_out)
+
         # --- Join the Mesh (QR + invite) ----------------------------------
         with gr.Accordion("📱 Join This Mesh — Connecting Nodes & Meshes", open=False):
             gr.Markdown("""

requirements.txt

@@ -10,3 +10,5 @@ transformers>=4.45.0
 qrcode[svg]>=7.4
 sentence-transformers>=3.0.0
 httpx>=0.27.0
+blake3>=0.4.0
+click>=8.1

scripts/check_deps_sync.py

@@ -0,0 +1,67 @@
+"""scripts/check_deps_sync.py — assert requirements.txt covers pyproject.toml deps.
+
+Run in CI:  python scripts/check_deps_sync.py
+Exit 0 = in sync.  Exit 1 = missing packages listed.
+"""
+
+from __future__ import annotations
+
+import re
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).parent.parent
+
+
+def _pkg_name(spec: str) -> str:
+    """Strip version constraints and extras, return normalised package name."""
+    name = re.split(r"[>=<!;\[]", spec.strip())[0].strip()
+    return name.lower().replace("-", "_").replace(".", "_")
+
+
+def _pyproject_deps() -> list[str]:
+    text = (ROOT / "pyproject.toml").read_text()
+    in_deps = False
+    deps: list[str] = []
+    for line in text.splitlines():
+        if line.strip() == "dependencies = [":
+            in_deps = True
+            continue
+        if in_deps:
+            if line.strip() == "]":
+                break
+            dep = line.strip().strip('",').strip()
+            if dep:
+                deps.append(dep)
+    return deps
+
+
+def _requirements_names() -> set[str]:
+    lines = (ROOT / "requirements.txt").read_text().splitlines()
+    names: set[str] = set()
+    for line in lines:
+        line = line.strip()
+        if not line or line.startswith("#") or line.startswith("-r"):
+            continue
+        names.add(_pkg_name(line))
+    return names
+
+
+def main() -> int:
+    pyproject_deps = _pyproject_deps()
+    req_names = _requirements_names()
+    missing: list[str] = []
+    for dep in pyproject_deps:
+        if _pkg_name(dep) not in req_names:
+            missing.append(dep)
+    if missing:
+        print("ERROR: pyproject.toml deps missing from requirements.txt:")
+        for m in missing:
+            print(f"  {m}")
+        return 1
+    print(f"OK: all {len(pyproject_deps)} pyproject.toml deps present in requirements.txt")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_federated_rag.py

@@ -535,7 +535,7 @@ class TestFederatedIntegration:
 
         with tempfile.TemporaryDirectory() as tmp:
             blob_store = BlobStore(Path(tmp) / "blobs")
-            svc = RagService(corpus="test", blob_store=blob_store)
+            svc = RagService(corpus="test", blob_store=blob_store, corpora_dir=Path(tmp) / "corpora")
 
             req = MagicMock(spec=RouteRequest)
             req.body = {
@@ -546,6 +546,9 @@ class TestFederatedIntegration:
                 }
             }
             result = run(svc.handle_ingest(req))
+            # Close SQLite before tempdir cleanup (Windows file-lock)
+            if getattr(svc._store, "_db", None) is not None:
+                svc._store._db.close()
 
         assert result["output"]["chunks_indexed"] >= 1
         assert result["output"]["was_duplicate"] is False

tests/test_improvements_batch.py

@@ -0,0 +1,245 @@
+"""Tests for the improvements batch (items 1–15).
+
+Covers:
+  - CorpusStore SQLite persistence (item 1)
+  - search_corpus bound to rag.federated_query (item 3)
+  - corpus param plumbing into body["params"] (item 4)
+  - _extract_json_object brace-matching parser (item 7)
+  - Bus failover when sole provider is quarantined (item 8)
+  - schema_hash prefix is "sha256:" not "blake3:" (item 10)
+"""
+
+from __future__ import annotations
+
+import asyncio
+import time
+from pathlib import Path
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Item 1 — CorpusStore SQLite persistence
+# ---------------------------------------------------------------------------
+
+
+def test_corpus_store_sqlite_persists(tmp_path: Path) -> None:
+    """Chunks written to CorpusStore survive a process-restart simulation."""
+    from hearthnet.services.rag.chunker import Chunk
+    from hearthnet.services.rag.store import CorpusStore
+
+    # Write
+    store1 = CorpusStore(tmp_path, "test_corpus")
+    chunks = [Chunk(text="hello world", metadata={"doc_cid": "doc1", "title": "Test"})]
+    store1.add(chunks, [[0.1, 0.2, 0.3]])
+    assert store1.count() == 1
+
+    # "Restart" — new instance, same path
+    store2 = CorpusStore(tmp_path, "test_corpus")
+    if store2._db is not None or store2._use_chroma:
+        assert store2.count() == 1, "Chunks should survive re-open"
+
+
+def test_corpus_store_sqlite_has_doc(tmp_path: Path) -> None:
+    from hearthnet.services.rag.chunker import Chunk
+    from hearthnet.services.rag.store import CorpusStore
+
+    store = CorpusStore(tmp_path, "test_corpus2")
+    chunks = [Chunk(text="water safety", metadata={"doc_cid": "water.001", "title": "Water"})]
+    store.add(chunks, [[0.5, 0.5]])
+    assert store.has_doc("water.001")
+    assert not store.has_doc("unknown.001")
+
+
+def test_corpus_store_corpus_info(tmp_path: Path) -> None:
+    from hearthnet.services.rag.store import CorpusStore
+
+    store = CorpusStore(tmp_path, "info_test")
+    info = store.corpus_info()
+    assert "backend" in info
+    assert "persistent" in info
+    assert "chunks" in info
+    assert info["backend"] in ("chroma", "sqlite", "in-memory")
+
+
+def test_corpus_store_query_after_sqlite_persist(tmp_path: Path) -> None:
+    from hearthnet.services.rag.chunker import Chunk
+    from hearthnet.services.rag.store import CorpusStore
+
+    store1 = CorpusStore(tmp_path, "query_test")
+    store1.add(
+        [Chunk(text="CPR steps", metadata={"doc_cid": "cpr.001"})],
+        [[1.0, 0.0, 0.0]],
+    )
+
+    store2 = CorpusStore(tmp_path, "query_test")
+    results = store2.query([1.0, 0.0, 0.0], k=3)
+    if store2._db is not None or store2._use_chroma:
+        assert len(results) >= 1
+        assert results[0].chunk.text == "CPR steps"
+
+
+# ---------------------------------------------------------------------------
+# Item 3 — search_corpus bound capability is rag.federated_query
+# ---------------------------------------------------------------------------
+
+
+def test_search_corpus_uses_federated_query() -> None:
+    from hearthnet.services.llm.tools import default_tool_set
+
+    executor = default_tool_set(bus=None)
+    search_tool = executor._tools.get("search_corpus")
+    assert search_tool is not None, "search_corpus tool must exist"
+    assert search_tool.bound_capability == "rag.federated_query", (
+        f"Expected rag.federated_query, got {search_tool.bound_capability!r}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Item 4 — corpus param plumbing into body["params"]
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_search_corpus_corpus_param_reaches_bus() -> None:
+    """When search_corpus is called with corpus='docs', the bus body must have
+    params={'corpus': 'docs'} so the router's _corpus_matches predicate sees it."""
+    from hearthnet.bus.capability import RouteRequest
+    from hearthnet.services.llm.tools import ToolCall, ToolDefinition, ToolExecutor
+
+    captured: list[dict] = []
+
+    class _FakeBus:
+        async def call(self, capability, version, body):
+            captured.append(body)
+            return {"output": {"chunks": []}}
+
+    tool = ToolDefinition(
+        name="search_corpus",
+        description="test",
+        parameters_schema={"type": "object", "properties": {"query": {}, "corpus": {}}},
+        bound_capability="rag.federated_query",
+        bound_version=(1, 0),
+    )
+    executor = ToolExecutor(bus=_FakeBus(), tools=[tool])
+    call = ToolCall(id="t1", name="search_corpus", arguments={"query": "water", "corpus": "docs"})
+    await executor.execute(call)
+
+    assert captured, "Bus should have been called"
+    body = captured[0]
+    assert body.get("params", {}).get("corpus") == "docs", (
+        "corpus must be in body['params'] for the router predicate"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Item 7 — _extract_json_object brace-matching parser
+# ---------------------------------------------------------------------------
+
+
+def test_extract_json_object_simple() -> None:
+    from hearthnet.services.llm.tools import _extract_json_object
+
+    text = 'action: {"tool": "search", "query": "hello"}'
+    start = text.index("{")
+    result = _extract_json_object(text, start)
+    assert result == '{"tool": "search", "query": "hello"}'
+
+
+def test_extract_json_object_nested() -> None:
+    from hearthnet.services.llm.tools import _extract_json_object
+
+    text = 'action: {"tool": "search", "tags": ["a", "b"], "opts": {"k": 3}}'
+    start = text.index("{")
+    result = _extract_json_object(text, start)
+    import json
+    parsed = json.loads(result)
+    assert parsed["tool"] == "search"
+    assert parsed["opts"]["k"] == 3
+    assert parsed["tags"] == ["a", "b"]
+
+
+def test_extract_json_object_brace_in_string() -> None:
+    from hearthnet.services.llm.tools import _extract_json_object
+
+    # Braces inside a string value must not be counted
+    text = 'action: {"tool": "x", "q": "use {braces} here"}'
+    start = text.index("{")
+    result = _extract_json_object(text, start)
+    import json
+    parsed = json.loads(result)
+    assert parsed["q"] == "use {braces} here"
+
+
+def test_extract_json_object_no_match() -> None:
+    from hearthnet.services.llm.tools import _extract_json_object
+
+    assert _extract_json_object("no braces here", 0) is None
+    assert _extract_json_object("{unclosed", 0) is None
+
+
+# ---------------------------------------------------------------------------
+# Item 8 — bus failover when sole local provider is quarantined
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_bus_failover_when_sole_local_provider_quarantined() -> None:
+    """When the only matching local entry is quarantined, handle_call must
+    succeed by routing to a remote alternative rather than raising not_found."""
+    from hearthnet.bus import CapabilityBus, InMemoryTransport
+    from hearthnet.bus.capability import CapabilityDescriptor, CapabilityEntry, RouteRequest
+
+    transport = InMemoryTransport()
+    bus_a = CapabilityBus("node-a", "community-test", transport=transport)
+    bus_b = CapabilityBus("node-b", "community-test", transport=transport)
+
+    async def good_handler(req: RouteRequest) -> dict:
+        return {"output": "from_b"}
+
+    desc = CapabilityDescriptor(name="test.cap", version=(1, 0), max_concurrent=4)
+    bus_b.register_capability(desc, good_handler)
+
+    # Add node-b as remote entry directly in node-a's registry
+    remote_entry = CapabilityEntry(
+        node_id="node-b",
+        descriptor=desc,
+        is_local=False,
+        handler=None,
+        last_seen=time.monotonic(),
+    )
+    bus_a.registry._entries[("node-b", "test.cap", (1, 0))] = remote_entry
+
+    # Register a quarantined local entry on bus_a
+    async def broken_handler(req: RouteRequest) -> dict:
+        return {"error": "broken"}
+
+    bus_a.registry.register_local(desc, broken_handler)
+    for e in list(bus_a.registry.all_local()):
+        if e.descriptor.name == "test.cap":
+            e.quarantined_until = time.monotonic() + 3600
+
+    req = RouteRequest(
+        capability="test.cap",
+        version_req=(1, 0),
+        body={},
+        caller="node-a",
+        trace_id="test",
+        deadline_ms=0,
+    )
+    result = await bus_a.handle_call(req)
+    assert result.get("output") == "from_b", f"Expected from_b, got {result!r}"
+
+
+# ---------------------------------------------------------------------------
+# Item 10 — schema_hash prefix is "sha256:" not "blake3:"
+# ---------------------------------------------------------------------------
+
+
+def test_schema_hash_prefix_is_sha256() -> None:
+    from hearthnet.bus.capability import CapabilityDescriptor
+
+    desc = CapabilityDescriptor(name="test.cap", version=(1, 0))
+    h = desc.schema_hash()
+    assert h.startswith("sha256:"), f"Expected 'sha256:' prefix, got: {h!r}"
+    assert not h.startswith("blake3:"), "blake3: prefix was a mislabel — must use sha256:"