feat(shard_manager): port to FULL Rust — dichotomy wave

src/rune/shard_manager.py

@@ -23,14 +23,41 @@ SHARD_INDEX = os.path.expanduser("~/.config/opencode/memory/shard_index.json")
 
 
 class ShardManager:
-    """Manages topic-based SQLite shards for unbounded memory growth."""
+    """Manages topic-based SQLite shards for unbounded memory growth.
+
+    FULL Rust core (src/rust/shard_manager) when built — strangler law:
+    RUNE_SM_CORE=python pins the Python fallback, which remains the parity
+    oracle. The sqlite/sqlite-vec resource boundary is served as ops by
+    sm_loader; connection construction stays Python-side."""
 
     def __init__(self):
+        core = self._core()
+        if core is not None:
+            self.index = core.init(SHARD_DIR, SHARD_INDEX)
+            return
         os.makedirs(SHARD_DIR, exist_ok=True)
         self.index = self._load_index()
 
+    @staticmethod
+    def _core():
+        """The Rust shard_manager core when available (strangler law).
+        RUNE_SM_CORE=python pins the fallback."""
+        if os.environ.get("RUNE_SM_CORE", "auto") == "python":
+            return None
+        try:
+            from rune import sm_loader
+        except Exception:  # noqa: BLE001 — any loader failure means fallback, never a broken call
+            return None
+        try:
+            return sm_loader if sm_loader.available() else None
+        except Exception:  # noqa: BLE001 — availability probe must never crash the fallback
+            return None
+
     def _load_index(self) -> Dict[str, str]:
         """Load shard index from disk."""
+        core = self._core()
+        if core is not None:
+            return core.load_index(SHARD_INDEX)
         if os.path.exists(SHARD_INDEX):
             try:
                 with open(SHARD_INDEX) as f:
@@ -41,11 +68,20 @@ class ShardManager:
 
     def _save_index(self):
         """Save shard index to disk."""
+        core = self._core()
+        if core is not None:
+            core.save_index(SHARD_INDEX, self.index)
+            return
         with open(SHARD_INDEX, "w") as f:
             json.dump(self.index, f, indent=2)
 
     def create_shard(self, topic: str) -> str:
         """Create a new shard for a topic. Returns shard path."""
+        core = self._core()
+        if core is not None:
+            res = core.create_shard(SHARD_DIR, SHARD_INDEX, topic, self.index)
+            self.index = res["index"]
+            return res["shard_path"]
         shard_name = f"shard_{topic.replace(' ', '_').replace('/', '_')}.db"
         shard_path = os.path.join(SHARD_DIR, shard_name)
 
@@ -94,6 +130,9 @@ class ShardManager:
 
         Returns the topic name. Falls back to 'general' if no match.
         """
+        core = self._core()
+        if core is not None:
+            return core.route_query(query, self.index)
         query_lower = query.lower()
 
         # Check existing shards for keyword matches
@@ -125,10 +164,15 @@ class ShardManager:
         extension loaded PER CONNECTION — a fresh connection without it cannot
         see or query entries_vec, even though the schema created it ("no such
         table" in clean environments). Mirror memory_store.connect()."""
-        if topic not in self.index:
-            self.create_shard(topic)
-
-        shard_path = self.index[topic]
+        core = self._core()
+        if core is not None:
+            res = core.get_shard(SHARD_DIR, SHARD_INDEX, topic, self.index)
+            self.index = res["index"]
+            shard_path = res["shard_path"]
+        else:
+            if topic not in self.index:
+                self.create_shard(topic)
+            shard_path = self.index[topic]
         db = sqlite3.connect(shard_path)
         db.row_factory = sqlite3.Row
         db.enable_load_extension(True)
@@ -142,6 +186,9 @@ class ShardManager:
 
     def merge_results(self, results_list: List[List[dict]]) -> List[dict]:
         """Merge results from multiple shards, deduplicate, sort by relevance."""
+        core = self._core()
+        if core is not None:
+            return core.merge_results(results_list)
         seen = set()
         merged = []
 
@@ -158,6 +205,9 @@ class ShardManager:
 
     def get_shard_stats(self) -> Dict[str, int]:
         """Get entry counts per shard."""
+        core = self._core()
+        if core is not None:
+            return core.get_shard_stats(self.index)
         stats = {}
         for topic, path in self.index.items():
             try:
@@ -171,6 +221,9 @@ class ShardManager:
 
     def total_entries(self) -> int:
         """Get total entries across all shards."""
+        core = self._core()
+        if core is not None:
+            return core.total_entries(self.index)
         stats = self.get_shard_stats()
         return sum(stats.values())
 

src/rune/sm_loader.py

@@ -1,9 +1,27 @@
 #!/usr/bin/env python3
-"""ctypes loader for the Rust structural_model core (facade burn-down).
+"""ctypes loader for the FULL Rust shard_manager core (dichotomy wave,
+issue #2479 tranche). Mirrors broker_loader.py's contract:
 
-FFI contract (doc-envelope law):
-  * sm_query({relationships, eid, op}) -> list/str/None
-  * sm_free_string(ptr)
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * sm_init({shard_dir, index_path}) -> index dict
+  * sm_load_index({index_path}) -> index dict
+  * sm_save_index({index_path, index}) -> null
+  * sm_create_shard({shard_dir, index_path, topic, index}, ops_cb)
+        -> {shard_path, index}
+  * sm_route_query({query, index}) -> topic string
+  * sm_merge_results({results_list}) -> merged list
+  * sm_get_shard({shard_dir, index_path, topic, index}, ops_cb)
+        -> {shard_path, index}
+  * sm_get_shard_stats({index}, ops_cb) -> stats dict
+  * sm_total_entries({index}, ops_cb) -> int
+
+Ops seam (resource boundary the kernel cannot own): the Rust kernel issues
+`ensure_schema` / `count_entries` ops through a C callback; THIS loader
+serves them against real sqlite3 + sqlite_vec. Connection construction
+stays Python-side (see shard_manager.ShardManager.get_shard).
+
+Free: capability_scoring's cs_free_string (shared process). Override the
+.so with SM_RUST_LIB (a FILE path).
 """
 
 from __future__ import annotations
@@ -11,19 +29,20 @@ from __future__ import annotations
 import ctypes
 import json
 import os
+import sqlite3
 from pathlib import Path
 
 _REPO = Path(__file__).resolve().parents[2]
-_CRATE_DIR = _REPO / "src" / "rust" / "structural_model"
-_LIB_PATH = Path(
-    os.environ.get(
-        "SM_RUST_LIB", _CRATE_DIR / "target" / "release" / "libstructural_model.so"
-    )
-)
+_CRATE_DIR = _REPO / "src" / "rust" / "shard_manager"
+_LIB_PATH = Path(os.environ.get("SM_RUST_LIB", _CRATE_DIR / "target" / "release" / "libshard_manager.so"))
 
 _lib = None
 
 
+class ShardCoreUnavailable(FileNotFoundError):
+    """The Rust shard_manager core is not built. Callers fall back to Python."""
+
+
 def available() -> bool:
     return _LIB_PATH.is_file()
 
@@ -32,32 +51,189 @@ def _load() -> ctypes.CDLL:
     global _lib
     if _lib is not None:
         return _lib
-    if not _LIB_PATH.exists():
-        raise FileNotFoundError(f"structural_model not built: {_LIB_PATH}")
+    if not _LIB_PATH.is_file():
+        raise ShardCoreUnavailable(str(_LIB_PATH))
     lib = ctypes.CDLL(str(_LIB_PATH))
-    lib.sm_free_string.argtypes = [ctypes.c_void_p]
-    lib.sm_free_string.restype = None
-    fn = lib.sm_query
-    fn.argtypes = [ctypes.c_char_p]
-    fn.restype = ctypes.c_void_p
+    for sym in (
+        "sm_init",
+        "sm_load_index",
+        "sm_save_index",
+        "sm_route_query",
+        "sm_merge_results",
+    ):
+        fn = getattr(lib, sym)
+        fn.argtypes = [ctypes.c_void_p]
+        fn.restype = ctypes.c_void_p
+    for sym in (
+        "sm_create_shard",
+        "sm_get_shard",
+        "sm_get_shard_stats",
+        "sm_total_entries",
+    ):
+        fn = getattr(lib, sym)
+        fn.argtypes = [ctypes.c_void_p, _OPS_CB]
+        fn.restype = ctypes.c_void_p
+    # free comes from capability_scoring (same process) if loaded; else our own malloc path
+    try:
+        lib.cs_free_string
+    except AttributeError:
+        scoring = ctypes.CDLL(
+            str(_REPO / "src" / "rust" / "capability_scoring" / "target" / "release" / "libcapability_scoring.so")
+        )
+        scoring.cs_free_string.argtypes = [ctypes.c_void_p]
+        lib._free = scoring.cs_free_string
     _lib = lib
     return lib
 
 
-def _call(sym: str, payload: dict):
+def _call(sym: str, payload: dict, with_ops: bool = False):
     lib = _load()
-    ptr = getattr(lib, sym)(json.dumps(payload).encode("utf-8"))
+    data = json.dumps(payload).encode()
+    buf = ctypes.create_string_buffer(data)
+    args = [ctypes.cast(buf, ctypes.c_void_p)]
+    if with_ops:
+        args.append(_ops_cb)
+    ptr = getattr(lib, sym)(*args)
     if not ptr:
         raise RuntimeError(f"{sym} returned null")
-    try:
-        text = ctypes.cast(ptr, ctypes.c_char_p).value.decode("utf-8")
-    finally:
-        lib.sm_free_string(ptr)
-    env = json.loads(text)
-    if not env.get("ok"):
-        raise RuntimeError(f"{sym} core error: {env.get('err')}")
-    return env["value"]
+    out = ctypes.string_at(ptr).decode()
+    free = getattr(lib, "_free", None)
+    if free is not None:
+        free(ptr)
+    doc = json.loads(out)
+    if not doc.get("ok"):
+        raise RuntimeError(doc.get("err", f"{sym} failed"))
+    return doc["value"]
 
 
-def query(relationships: list, eid: str, op: str):
-    return _call("sm_query", {"relationships": relationships, "eid": eid, "op": op})
+# ── ops seam: the loader serves the kernel's sqlite ops ──────────────────────
+
+_OPS_CB = ctypes.CFUNCTYPE(ctypes.c_void_p, ctypes.c_void_p)
+_last_ops_buf = None
+
+SCHEMA = """
+            CREATE TABLE IF NOT EXISTS entries (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                text TEXT NOT NULL,
+                importance REAL DEFAULT 0.5,
+                created_at TEXT NOT NULL,
+                last_accessed TEXT,
+                topic TEXT DEFAULT '',
+                source TEXT DEFAULT '',
+                status TEXT DEFAULT 'active',
+                wash_count INTEGER DEFAULT 0,
+                last_washed TEXT,
+                rock_status TEXT DEFAULT 'washable',
+                tier TEXT DEFAULT 'warm'
+            );
+            CREATE VIRTUAL TABLE IF NOT EXISTS entries_fts USING fts5(
+                text, content='', tokenize='porter unicode61'
+            );
+            CREATE VIRTUAL TABLE IF NOT EXISTS entries_vec USING vec0(
+                embedding float[256]
+            );
+            CREATE INDEX IF NOT EXISTS idx_entries_status ON entries(status);
+            CREATE INDEX IF NOT EXISTS idx_entries_topic ON entries(topic);
+            CREATE INDEX IF NOT EXISTS idx_entries_wash_count ON entries(wash_count);
+            CREATE INDEX IF NOT EXISTS idx_entries_tier ON entries(tier);
+        """
+
+
+def _execute_op(op: dict) -> dict:
+    kind = op.get("op")
+    args = op.get("args") or {}
+    if kind == "ensure_schema":
+        # mirror oracle create_shard lines 55-86 (sqlite + sqlite_vec + schema)
+        path = args["path"]
+        db = sqlite3.connect(path)
+        try:
+            db.enable_load_extension(True)
+            import sqlite_vec
+
+            sqlite_vec.load(db)  # #26: entries_vec virtual table needs the extension per connection
+            db.executescript(SCHEMA)
+        finally:
+            db.close()
+        return {"ok": True}
+    if kind == "count_entries":
+        # mirror oracle get_shard_stats lines 163-169 (plain connect; any
+        # failure degrades to the except branch → kernel writes 0)
+        try:
+            path = args["path"]
+            db = sqlite3.connect(path)
+            try:
+                count = db.execute("SELECT COUNT(*) FROM entries WHERE status='active'").fetchone()[0]
+            finally:
+                db.close()
+            return {"ok": True, "value": count}
+        except Exception:  # noqa: BLE001 — deliberate: best-effort degrade, crash would be worse
+            return {"ok": False}
+    return {"ok": False, "err": f"unknown op {kind!r}"}
+
+
+def _ops_handler(op_ptr) -> int:
+    global _last_ops_buf
+    op = json.loads(ctypes.string_at(op_ptr).decode())
+    result = _execute_op(op)
+    _last_ops_buf = ctypes.create_string_buffer(json.dumps(result).encode())
+    return ctypes.cast(_last_ops_buf, ctypes.c_void_p).value
+
+
+_ops_cb = _OPS_CB(_ops_handler)
+
+
+# ── typed public API ─────────────────────────────────────────────────────────
+
+
+def init(shard_dir: str, index_path: str) -> dict:
+    return _call("sm_init", {"shard_dir": shard_dir, "index_path": index_path})
+
+
+def load_index(index_path: str) -> dict:
+    return _call("sm_load_index", {"index_path": index_path})
+
+
+def save_index(index_path: str, index: dict) -> None:
+    _call("sm_save_index", {"index_path": index_path, "index": index})
+
+
+def create_shard(shard_dir: str, index_path: str, topic: str, index: dict) -> dict:
+    return _call(
+        "sm_create_shard",
+        {
+            "shard_dir": shard_dir,
+            "index_path": index_path,
+            "topic": topic,
+            "index": index,
+        },
+        with_ops=True,
+    )
+
+
+def route_query(query: str, index: dict) -> str:
+    return _call("sm_route_query", {"query": query, "index": index})
+
+
+def merge_results(results_list: list) -> list:
+    return _call("sm_merge_results", {"results_list": results_list})
+
+
+def get_shard(shard_dir: str, index_path: str, topic: str, index: dict) -> dict:
+    return _call(
+        "sm_get_shard",
+        {
+            "shard_dir": shard_dir,
+            "index_path": index_path,
+            "topic": topic,
+            "index": index,
+        },
+        with_ops=True,
+    )
+
+
+def get_shard_stats(index: dict) -> dict:
+    return _call("sm_get_shard_stats", {"index": index}, with_ops=True)
+
+
+def total_entries(index: dict) -> int:
+    return _call("sm_total_entries", {"index": index}, with_ops=True)

src/rust/shard_manager/Cargo.lock

@@ -3,103 +3,12 @@
 version = 4
 
 [[package]]
-name = "itoa"
-version = "1.0.18"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
-
-[[package]]
-name = "memchr"
-version = "2.8.3"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98"
-
-[[package]]
-name = "proc-macro2"
-version = "1.0.107"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9"
-dependencies = [
- "unicode-ident",
-]
-
-[[package]]
-name = "quote"
-version = "1.0.47"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001"
-dependencies = [
- "proc-macro2",
-]
-
-[[package]]
-name = "serde"
-version = "1.0.229"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
-dependencies = [
- "serde_core",
-]
-
-[[package]]
-name = "serde_core"
-version = "1.0.229"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48"
-dependencies = [
- "serde_derive",
-]
-
-[[package]]
-name = "serde_derive"
-version = "1.0.229"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn",
-]
-
-[[package]]
-name = "serde_json"
-version = "1.0.151"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
-dependencies = [
- "itoa",
- "memchr",
- "serde",
- "serde_core",
- "zmij",
-]
+name = "capability_scoring"
+version = "0.1.0"
 
 [[package]]
 name = "shard_manager"
 version = "0.1.0"
 dependencies = [
- "serde_json",
+ "capability_scoring",
 ]
-
-[[package]]
-name = "syn"
-version = "3.0.6"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee"
-dependencies = [
- "proc-macro2",
- "quote",
- "unicode-ident",
-]
-
-[[package]]
-name = "unicode-ident"
-version = "1.0.26"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954"
-
-[[package]]
-name = "zmij"
-version = "1.0.23"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b"

src/rust/shard_manager/Cargo.toml

@@ -7,4 +7,4 @@ edition = "2021"
 crate-type = ["rlib", "cdylib", "staticlib"]
 
 [dependencies]
-serde_json = "1"
+capability_scoring = { path = "../capability_scoring" }

src/rust/shard_manager/frozen_sm_truth.json

@@ -0,0 +1,752 @@
+{
+ "module": "shard_manager",
+ "pinned": "RUNE_SM_CORE=python",
+ "oracle": "src/rune/shard_manager.py",
+ "cases": [
+  {
+   "id": "load_missing",
+   "fn": "_load_index",
+   "setup": {
+    "dirs": [
+     "shards"
+    ]
+   },
+   "input": {
+    "index_path": "shard_index.json"
+   },
+   "ops": [],
+   "expected": {},
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "load_corrupt",
+   "fn": "_load_index",
+   "setup": {
+    "dirs": [
+     "shards"
+    ],
+    "index_file": "CORRUPT"
+   },
+   "input": {
+    "index_path": "shard_index.json"
+   },
+   "ops": [],
+   "expected": {},
+   "effects": {
+    "index_file": "CORRUPT"
+   },
+   "created_files": []
+  },
+  {
+   "id": "load_valid",
+   "fn": "_load_index",
+   "setup": {
+    "dirs": [
+     "shards"
+    ],
+    "index_file": {
+     "alpha": "shards/shard_alpha.db",
+     "beta": "shards/shard_beta.db"
+    }
+   },
+   "input": {
+    "index_path": "shard_index.json"
+   },
+   "ops": [],
+   "expected": {
+    "alpha": "shards/shard_alpha.db",
+    "beta": "shards/shard_beta.db"
+   },
+   "effects": {
+    "index_file": {
+     "alpha": "shards/shard_alpha.db",
+     "beta": "shards/shard_beta.db"
+    }
+   },
+   "created_files": []
+  },
+  {
+   "id": "init_creates_dir",
+   "fn": "__init__",
+   "setup": {},
+   "input": {
+    "shard_dir": "shards",
+    "index_path": "shard_index.json"
+   },
+   "ops": [],
+   "expected": {
+    "index": {},
+    "dir_exists": true
+   },
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "create_fresh",
+   "fn": "create_shard",
+   "setup": {
+    "dirs": [
+     "shards"
+    ]
+   },
+   "input": {
+    "shard_dir": "shards",
+    "index_path": "shard_index.json",
+    "topic": "deployment",
+    "index": {}
+   },
+   "ops": [
+    {
+     "op": "ensure_schema",
+     "args": {
+      "path": "shards/shard_deployment.db"
+     },
+     "result": null,
+     "ok": true
+    }
+   ],
+   "expected": "shards/shard_deployment.db",
+   "effects": {
+    "index_file": {
+     "deployment": "shards/shard_deployment.db"
+    }
+   },
+   "created_files": [
+    "shards/shard_deployment.db"
+   ]
+  },
+  {
+   "id": "create_file_exists",
+   "fn": "create_shard",
+   "setup": {
+    "dirs": [
+     "shards"
+    ],
+    "files": [
+     "shards/shard_deployment.db"
+    ]
+   },
+   "input": {
+    "shard_dir": "shards",
+    "index_path": "shard_index.json",
+    "topic": "deployment",
+    "index": {
+     "other": "shards/shard_other.db"
+    }
+   },
+   "ops": [],
+   "expected": "shards/shard_deployment.db",
+   "effects": {},
+   "created_files": [
+    "shards/shard_deployment.db"
+   ]
+  },
+  {
+   "id": "create_name_normalize",
+   "fn": "create_shard",
+   "setup": {
+    "dirs": [
+     "shards"
+    ]
+   },
+   "input": {
+    "shard_dir": "shards",
+    "index_path": "shard_index.json",
+    "topic": "guitar academy/basics",
+    "index": {}
+   },
+   "ops": [
+    {
+     "op": "ensure_schema",
+     "args": {
+      "path": "shards/shard_guitar_academy_basics.db"
+     },
+     "result": null,
+     "ok": true
+    }
+   ],
+   "expected": "shards/shard_guitar_academy_basics.db",
+   "effects": {
+    "index_file": {
+     "guitar academy/basics": "shards/shard_guitar_academy_basics.db"
+    }
+   },
+   "created_files": [
+    "shards/shard_guitar_academy_basics.db"
+   ]
+  },
+  {
+   "id": "create_topic_in_index",
+   "fn": "create_shard",
+   "setup": {
+    "dirs": [
+     "shards"
+    ],
+    "files": [
+     "shards/shard_reuse.db"
+    ]
+   },
+   "input": {
+    "shard_dir": "shards",
+    "index_path": "shard_index.json",
+    "topic": "reuse",
+    "index": {
+     "reuse": "shards/shard_reuse.db"
+    }
+   },
+   "ops": [],
+   "expected": "shards/shard_reuse.db",
+   "effects": {},
+   "created_files": [
+    "shards/shard_reuse.db"
+   ]
+  },
+  {
+   "id": "route_overlap",
+   "fn": "route_query",
+   "setup": {},
+   "input": {
+    "query": "how is deployment going",
+    "index": {
+     "deployment": "p1",
+     "general": "p2"
+    }
+   },
+   "ops": [],
+   "expected": "deployment",
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "route_tie_first_wins",
+   "fn": "route_query",
+   "setup": {},
+   "input": {
+    "query": "b d",
+    "index": {
+     "a_b": "p1",
+     "c_d": "p2"
+    }
+   },
+   "ops": [],
+   "expected": "a_b",
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "route_multiword_beats",
+   "fn": "route_query",
+   "setup": {},
+   "input": {
+    "query": "machine learning basics",
+    "index": {
+     "general": "p1",
+     "machine_learning": "p2"
+    }
+   },
+   "ops": [],
+   "expected": "machine_learning",
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "route_no_match_long_words",
+   "fn": "route_query",
+   "setup": {},
+   "input": {
+    "query": "what is my progress today",
+    "index": {}
+   },
+   "ops": [],
+   "expected": "what_progress_today",
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "route_no_match_short",
+   "fn": "route_query",
+   "setup": {},
+   "input": {
+    "query": "hi",
+    "index": {}
+   },
+   "ops": [],
+   "expected": "general",
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "route_empty",
+   "fn": "route_query",
+   "setup": {},
+   "input": {
+    "query": "",
+    "index": {}
+   },
+   "ops": [],
+   "expected": "general",
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "route_case_insensitive",
+   "fn": "route_query",
+   "setup": {},
+   "input": {
+    "query": "DEPLOYMENT status",
+    "index": {
+     "deployment": "p1"
+    }
+   },
+   "ops": [],
+   "expected": "deployment",
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "merge_dedupe_prefix",
+   "fn": "merge_results",
+   "setup": {},
+   "input": {
+    "results_list": [
+     [
+      {
+       "text": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxaaa",
+       "relevance": 1
+      },
+      {
+       "text": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxbbb",
+       "relevance": 2
+      }
+     ],
+     [
+      {
+       "text": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxccc",
+       "relevance": 3
+      }
+     ]
+    ]
+   },
+   "ops": [],
+   "expected": [
+    {
+     "text": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxaaa",
+     "relevance": 1
+    }
+   ],
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "merge_missing_text",
+   "fn": "merge_results",
+   "setup": {},
+   "input": {
+    "results_list": [
+     [
+      {
+       "wash_count": 2
+      },
+      {
+       "text": "",
+       "x": 1
+      }
+     ]
+    ]
+   },
+   "ops": [],
+   "expected": [
+    {
+     "wash_count": 2
+    }
+   ],
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "merge_sort_relevance",
+   "fn": "merge_results",
+   "setup": {},
+   "input": {
+    "results_list": [
+     [
+      {
+       "text": "a",
+       "relevance": 1
+      }
+     ],
+     [
+      {
+       "text": "b",
+       "relevance": 5
+      }
+     ],
+     [
+      {
+       "text": "c",
+       "relevance": 3
+      }
+     ]
+    ]
+   },
+   "ops": [],
+   "expected": [
+    {
+     "text": "b",
+     "relevance": 5
+    },
+    {
+     "text": "c",
+     "relevance": 3
+    },
+    {
+     "text": "a",
+     "relevance": 1
+    }
+   ],
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "merge_sort_wash_count",
+   "fn": "merge_results",
+   "setup": {},
+   "input": {
+    "results_list": [
+     [
+      {
+       "text": "a",
+       "wash_count": 2
+      }
+     ],
+     [
+      {
+       "text": "b",
+       "wash_count": 7
+      }
+     ]
+    ]
+   },
+   "ops": [],
+   "expected": [
+    {
+     "text": "b",
+     "wash_count": 7
+    },
+    {
+     "text": "a",
+     "wash_count": 2
+    }
+   ],
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "merge_ties_stable",
+   "fn": "merge_results",
+   "setup": {},
+   "input": {
+    "results_list": [
+     [
+      {
+       "text": "a"
+      }
+     ],
+     [
+      {
+       "text": "b"
+      }
+     ],
+     [
+      {
+       "text": "c"
+      }
+     ]
+    ]
+   },
+   "ops": [],
+   "expected": [
+    {
+     "text": "a"
+    },
+    {
+     "text": "b"
+    },
+    {
+     "text": "c"
+    }
+   ],
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "merge_empty",
+   "fn": "merge_results",
+   "setup": {},
+   "input": {
+    "results_list": [
+     [],
+     []
+    ]
+   },
+   "ops": [],
+   "expected": [],
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "merge_mixed_keys",
+   "fn": "merge_results",
+   "setup": {},
+   "input": {
+    "results_list": [
+     [
+      {
+       "text": "a",
+       "wash_count": 1
+      },
+      {
+       "text": "b",
+       "relevance": 0.5
+      }
+     ],
+     [
+      {
+       "text": "c"
+      }
+     ]
+    ]
+   },
+   "ops": [],
+   "expected": [
+    {
+     "text": "a",
+     "wash_count": 1
+    },
+    {
+     "text": "b",
+     "relevance": 0.5
+    },
+    {
+     "text": "c"
+    }
+   ],
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "getshard_existing",
+   "fn": "get_shard",
+   "setup": {
+    "dirs": [
+     "shards"
+    ],
+    "files": [
+     "shards/shard_reuse.db"
+    ]
+   },
+   "input": {
+    "shard_dir": "shards",
+    "index_path": "shard_index.json",
+    "topic": "reuse",
+    "index": {
+     "reuse": "shards/shard_reuse.db"
+    }
+   },
+   "ops": [],
+   "expected": "shards/shard_reuse.db",
+   "effects": {},
+   "created_files": [
+    "shards/shard_reuse.db"
+   ]
+  },
+  {
+   "id": "getshard_creates",
+   "fn": "get_shard",
+   "setup": {
+    "dirs": [
+     "shards"
+    ]
+   },
+   "input": {
+    "shard_dir": "shards",
+    "index_path": "shard_index.json",
+    "topic": "brand new",
+    "index": {}
+   },
+   "ops": [
+    {
+     "op": "ensure_schema",
+     "args": {
+      "path": "shards/shard_brand_new.db"
+     },
+     "result": null,
+     "ok": true
+    }
+   ],
+   "expected": "shards/shard_brand_new.db",
+   "effects": {
+    "index_file": {
+     "brand new": "shards/shard_brand_new.db"
+    }
+   },
+   "created_files": [
+    "shards/shard_brand_new.db"
+   ]
+  },
+  {
+   "id": "stats_counts",
+   "fn": "get_shard_stats",
+   "setup": {
+    "dirs": [
+     "shards"
+    ],
+    "dbs": {
+     "shards/shard_t1.db": {
+      "entries": [
+       [
+        "a",
+        "active"
+       ],
+       [
+        "b",
+        "active"
+       ],
+       [
+        "c",
+        "washed"
+       ],
+       [
+        "d",
+        "active"
+       ]
+      ]
+     },
+     "shards/shard_t2.db": {
+      "entries": []
+     }
+    }
+   },
+   "input": {
+    "index": {
+     "t1": "shards/shard_t1.db",
+     "t2": "shards/shard_t2.db"
+    }
+   },
+   "ops": [
+    {
+     "op": "count_entries",
+     "args": {
+      "path": "shards/shard_t1.db"
+     },
+     "result": 3,
+     "ok": true
+    },
+    {
+     "op": "count_entries",
+     "args": {
+      "path": "shards/shard_t2.db"
+     },
+     "result": 0,
+     "ok": true
+    }
+   ],
+   "expected": {
+    "t1": 3,
+    "t2": 0
+   },
+   "effects": {},
+   "created_files": [
+    "shards/shard_t1.db",
+    "shards/shard_t2.db"
+   ]
+  },
+  {
+   "id": "stats_broken_path",
+   "fn": "get_shard_stats",
+   "setup": {},
+   "input": {
+    "index": {
+     "gone": "no_such_dir/shard_gone.db"
+    }
+   },
+   "ops": [
+    {
+     "op": "count_entries",
+     "args": {
+      "path": "no_such_dir/shard_gone.db"
+     },
+     "result": null,
+     "ok": false
+    }
+   ],
+   "expected": {
+    "gone": 0
+   },
+   "effects": {},
+   "created_files": []
+  },
+  {
+   "id": "total_sums",
+   "fn": "total_entries",
+   "setup": {
+    "dirs": [
+     "shards"
+    ],
+    "dbs": {
+     "shards/shard_t1.db": {
+      "entries": [
+       [
+        "a",
+        "active"
+       ],
+       [
+        "b",
+        "active"
+       ]
+      ]
+     },
+     "shards/shard_t2.db": {
+      "entries": [
+       [
+        "c",
+        "active"
+       ]
+      ]
+     }
+    }
+   },
+   "input": {
+    "index": {
+     "t1": "shards/shard_t1.db",
+     "t2": "shards/shard_t2.db"
+    }
+   },
+   "ops": [
+    {
+     "op": "count_entries",
+     "args": {
+      "path": "shards/shard_t1.db"
+     },
+     "result": 2,
+     "ok": true
+    },
+    {
+     "op": "count_entries",
+     "args": {
+      "path": "shards/shard_t2.db"
+     },
+     "result": 1,
+     "ok": true
+    }
+   ],
+   "expected": 3,
+   "effects": {},
+   "created_files": [
+    "shards/shard_t1.db",
+    "shards/shard_t2.db"
+   ]
+  }
+ ]
+}

src/rust/shard_manager/src/branch_battery.rs

@@ -0,0 +1,219 @@
+//! branch_battery — verbatim replay of frozen_sm_truth.json (27 cases frozen
+//! from the LIVE Python oracle, RUNE_SM_CORE=python pinned). Every case:
+//! materialize the setup in a scratch cwd, call the kernel, compare the
+//! result AND the issued ops (exact order, exact args, recorded results
+//! replayed through a mock ops callback), then compare on-disk index effects.
+
+use super::*;
+use std::cell::RefCell;
+use std::ffi::CString;
+use std::os::raw::c_char;
+
+const TRUTH: &str = include_str!("../frozen_sm_truth.json");
+
+thread_local! {
+    static OPS: RefCell<Vec<Json>> = const { RefCell::new(Vec::new()) };
+    static LAST_BUF: RefCell<Option<CString>> = const { RefCell::new(None) };
+}
+
+extern "C" fn mock_ops_cb(op_ptr: *const c_char) -> *const c_char {
+    let op_json = unsafe { CStr::from_ptr(op_ptr) }.to_str().unwrap();
+    let op = pyjson::parse(op_json).unwrap();
+    let next = OPS.with(|q| {
+        let mut q = q.borrow_mut();
+        assert!(!q.is_empty(), "kernel issued an op the oracle never did: {}", op_json);
+        q.remove(0)
+    });
+    assert_eq!(
+        op.get("op"), next.get("op"),
+        "op order divergence: kernel issued {}, truth expects {:?}",
+        op_json,
+        next.get("op")
+    );
+    assert_eq!(
+        op.get("args"), next.get("args"),
+        "op args divergence on op {:?}: kernel {}, truth {:?}",
+        next.get("op"),
+        op_json,
+        next.get("args")
+    );
+    let ok = matches!(next.get("ok"), Some(Json::Bool(true)));
+    let value = next.get("result").cloned().unwrap_or(Json::Null);
+    let resp = pyjson::dumps(&Json::Obj(vec![
+        ("ok".into(), Json::Bool(ok)),
+        ("value".into(), value),
+    ]));
+    let buf = CString::new(resp).unwrap();
+    let ptr = buf.as_ptr();
+    LAST_BUF.with(|b| *b.borrow_mut() = Some(buf));
+    ptr
+}
+
+fn setup_case(case: &Json, root: &std::path::Path) {
+    std::fs::create_dir_all(root).unwrap();
+    std::env::set_current_dir(root).unwrap();
+    if let Some(Json::Arr(dirs)) = case.get("setup").and_then(|s| s.get("dirs")) {
+        for d in dirs {
+            if let Json::Str(d) = d {
+                std::fs::create_dir_all(d).unwrap();
+            }
+        }
+    }
+    if let Some(Json::Arr(files)) = case.get("setup").and_then(|s| s.get("files")) {
+        for f in files {
+            if let Json::Str(f) = f {
+                if let Some(parent) = std::path::Path::new(f).parent() {
+                    std::fs::create_dir_all(parent).unwrap();
+                }
+                std::fs::write(f, b"").unwrap();
+            }
+        }
+    }
+    if let Some(setup) = case.get("setup") {
+        match setup.get("index_file") {
+            Some(Json::Str(s)) if s == "CORRUPT" => {
+                std::fs::write("shard_index.json", b"not json {{{").unwrap();
+            }
+            Some(v) => {
+                std::fs::write("shard_index.json", pyjson::dumps(v)).unwrap();
+            }
+            None => {}
+        }
+    }
+}
+
+fn run_kernel(case: &Json) -> Json {
+    let inp = case.get("input").unwrap();
+    let index = inp.get("index").cloned().unwrap_or(Json::Obj(Vec::new()));
+    let fn_name = case.get("fn").and_then(|f| match f {
+        Json::Str(s) => Some(s.as_str()),
+        _ => None,
+    }).unwrap();
+    match fn_name {
+        "_load_index" => load_index(
+            inp.get("index_path").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+        ).unwrap(),
+        "__init__" => {
+            let idx = init(
+                inp.get("shard_dir").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                inp.get("index_path").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+            ).unwrap();
+            let dir_exists = std::path::Path::new(
+                inp.get("shard_dir").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+            ).is_dir();
+            Json::Obj(vec![
+                ("index".into(), idx),
+                ("dir_exists".into(), Json::Bool(dir_exists)),
+            ])
+        }
+        "create_shard" => {
+            let (path, idx) = create_shard(
+                inp.get("shard_dir").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                inp.get("index_path").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                inp.get("topic").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                &index,
+                mock_ops_cb,
+            ).unwrap();
+            Json::Obj(vec![
+                ("shard_path".into(), Json::Str(path)),
+                ("index".into(), idx),
+            ])
+        }
+        "route_query" => Json::Str(route_query(
+            inp.get("query").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+            &index,
+        )),
+        "merge_results" => merge_results(inp.get("results_list").unwrap()),
+        "get_shard" => {
+            let (path, idx) = get_shard(
+                inp.get("shard_dir").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                inp.get("index_path").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                inp.get("topic").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                &index,
+                mock_ops_cb,
+            ).unwrap();
+            Json::Obj(vec![
+                ("shard_path".into(), Json::Str(path)),
+                ("index".into(), idx),
+            ])
+        }
+        "get_shard_stats" => get_shard_stats(&index, mock_ops_cb).unwrap(),
+        "total_entries" => Json::Num(total_entries(&index, mock_ops_cb).unwrap() as f64),
+        other => panic!("unknown truth fn {}", other),
+    }
+}
+
+/// Project the kernel result into the truth's `expected` shape and compare.
+fn assert_case(case: &Json, result: &Json) {
+    let id = match case.get("id") {
+        Some(Json::Str(s)) => s.as_str(),
+        _ => "?",
+    };
+    let fn_name = case.get("fn").and_then(|f| match f {
+        Json::Str(s) => Some(s.as_str()),
+        _ => None,
+    }).unwrap();
+    let expected = case.get("expected").unwrap();
+    match fn_name {
+        "create_shard" | "get_shard" => {
+            assert_eq!(
+                result.get("shard_path"), Some(expected),
+                "case {}: shard_path mismatch",
+                id
+            );
+        }
+        _ => assert_eq!(result, expected, "case {}: result mismatch", id),
+    }
+    // on-disk index effect (oracle's _save_index footprint)
+    if let Some(Json::Obj(_)) = case.get("effects").and_then(|e| e.get("index_file")) {
+        let want = case.get("effects").unwrap().get("index_file").unwrap();
+        let got = pyjson::parse(&std::fs::read_to_string("shard_index.json").unwrap()).unwrap();
+        assert_eq!(&got, want, "case {}: index file effect mismatch", id);
+    }
+}
+
+#[test]
+fn battery_replays_frozen_truth_verbatim() {
+    let truth = pyjson::parse(TRUTH).unwrap();
+    let cases = match truth.get("cases") {
+        Some(Json::Arr(c)) => c.clone(),
+        _ => panic!("truth has no cases"),
+    };
+    assert_eq!(cases.len(), 27, "frozen truth case count");
+    let base = std::env::temp_dir().join(format!("sm_battery_{}", std::process::id()));
+    let _ = std::fs::remove_dir_all(&base);
+    std::fs::create_dir_all(&base).unwrap();
+    let mut replayed = 0;
+    for (i, case) in cases.iter().enumerate() {
+        let root = base.join(format!("case_{:02}", i));
+        let _ = std::fs::remove_dir_all(&root);
+        setup_case(case, &root);
+        OPS.with(|q| {
+            *q.borrow_mut() = case
+                .get("ops")
+                .and_then(|o| match o {
+                    Json::Arr(a) => Some(a.clone()),
+                    _ => None,
+                })
+                .unwrap_or_default();
+        });
+        let result = run_kernel(case);
+        OPS.with(|q| {
+            let remaining = q.borrow();
+            assert!(
+                remaining.is_empty(),
+                "case {}: kernel issued fewer ops than the oracle ({} pending: {:?})",
+                case.get("id").and_then(|v| match v { Json::Str(s) => Some(s.as_str()), _ => None }).unwrap(),
+                remaining.len(),
+                remaining
+            );
+        });
+        assert_case(case, &result);
+        replayed += 1;
+        let _ = std::env::set_current_dir(&base);
+        let _ = std::fs::remove_dir_all(&root);
+    }
+    std::env::set_current_dir(&base).ok();
+    let _ = std::fs::remove_dir_all(&base);
+    assert_eq!(replayed, 27);
+}

src/rust/shard_manager/src/lib.rs

@@ -1,71 +1,401 @@
-//! shard_manager — shard-based storage routing (W6).
-//! PARITY port of src/rune/shard_manager.py: topic → shard path mapping.
+//! shard_manager — FULL Rust port of src/rune/shard_manager.py (dichotomy
+//! wave #2479 tranche). Topic → shard routing, index persistence, result
+//! merging, per-shard entry counts.
+//!
+//! Parity oracle: frozen_sm_truth.json (27 cases frozen from the LIVE Python
+//! oracle 2026-10-11, RUNE_SM_CORE=python pinned). The sqlite/sqlite-vec
+//! resource boundary is the ops seam: the kernel issues `ensure_schema` /
+//! `count_entries` ops through a C callback; the Python loader serves them
+//! against real sqlite3 + sqlite_vec. Connection construction stays
+//! Python-side (resource handle cannot cross FFI) — mirror of oracle lines
+//! 132-141 lives in the facade, both core modes.
 
-/// Generate a shard file name from a topic (same normalization as Python).
+pub use capability_scoring::pyjson;
+use pyjson::Json;
+
+use std::collections::HashSet;
+use std::ffi::{CStr, CString};
+use std::os::raw::c_char;
+
+/// Ops callback: receives a NUL-terminated op JSON
+/// `{"op": ..., "args": {...}}`, returns a NUL-terminated result JSON
+/// `{"ok": bool, "value": ...}` owned by the callee, valid until the next
+/// invocation of the callback. A null return means op failure.
+pub type OpsCb = extern "C" fn(*const c_char) -> *const c_char;
+
+pub const OP_ENSURE_SCHEMA: &str = "ensure_schema";
+pub const OP_COUNT_ENTRIES: &str = "count_entries";
+
+// ── doc-envelope FFI plumbing ────────────────────────────────────────────────
+
+fn read_arg(ptr: *const c_char) -> Result<Json, String> {
+    if ptr.is_null() {
+        return Err("null argument pointer".into());
+    }
+    let x = unsafe { CStr::from_ptr(ptr) }.to_str().map_err(|e| e.to_string())?;
+    pyjson::parse(x)
+}
+
+fn envelope(r: Result<Json, String>) -> *mut c_char {
+    let v = match r {
+        Ok(x) => Json::Obj(vec![
+            ("ok".into(), Json::Bool(true)),
+            ("value".into(), x),
+        ]),
+        Err(e) => Json::Obj(vec![
+            ("ok".into(), Json::Bool(false)),
+            ("err".into(), Json::Str(e)),
+        ]),
+    };
+    CString::new(pyjson::dumps(&v))
+        .map(|c| c.into_raw())
+        .unwrap_or(std::ptr::null_mut())
+}
+
+/// Issue one op through the callback; `None` on any failure (kernel then
+/// degrades exactly like the oracle's best-effort branches).
+fn issue_op(cb: OpsCb, op: &str, args: Json) -> Option<Json> {
+    let payload = CString::new(pyjson::dumps(&Json::Obj(vec![
+        ("op".into(), Json::Str(op.into())),
+        ("args".into(), args),
+    ])))
+    .ok()?;
+    let ret = cb(payload.as_ptr());
+    if ret.is_null() {
+        return None;
+    }
+    let s = unsafe { CStr::from_ptr(ret) }.to_str().ok()?;
+    let v = pyjson::parse(s).ok()?;
+    match v.get("ok") {
+        Some(Json::Bool(true)) => Some(v.get("value").cloned().unwrap_or(Json::Null)),
+        _ => None,
+    }
+}
+
+// ── index helpers (Python dict semantics: insert-order, in-place update) ────
+
+fn index_set(index: &mut Json, topic: &str, path: &str) {
+    if let Json::Obj(pairs) = index {
+        if let Some(slot) = pairs.iter_mut().find(|(k, _)| k == topic) {
+            slot.1 = Json::Str(path.into());
+            return;
+        }
+        pairs.push((topic.into(), Json::Str(path.into())));
+    }
+}
+
+/// os.path.join(dir, name) for the posix shapes this module produces.
+fn join_path(dir: &str, name: &str) -> String {
+    if dir.is_empty() {
+        return name.into();
+    }
+    if dir.ends_with('/') {
+        format!("{}{}", dir, name)
+    } else {
+        format!("{}/{}", dir, name)
+    }
+}
+
+// ── kernel (name-preserving port of the 9 oracle fns) ───────────────────────
+
+/// __init__: makedirs(SHARD_DIR, exist_ok=True) + _load_index()
+pub fn init(shard_dir: &str, index_path: &str) -> Result<Json, String> {
+    std::fs::create_dir_all(shard_dir).map_err(|e| e.to_string())?;
+    load_index(index_path)
+}
+
+/// _load_index: missing file → {}; JSONDecodeError → {}; else the parsed dict.
+pub fn load_index(index_path: &str) -> Result<Json, String> {
+    let p = std::path::Path::new(index_path);
+    if !p.exists() {
+        return Ok(Json::Obj(Vec::new()));
+    }
+    let raw = std::fs::read_to_string(p).map_err(|e| e.to_string())?;
+    Ok(pyjson::parse(&raw).unwrap_or(Json::Obj(Vec::new())))
+}
+
+/// _save_index: json.dump(self.index, f, indent=2) — no trailing newline.
+pub fn save_index(index_path: &str, index: &Json) -> Result<(), String> {
+    std::fs::write(index_path, pyjson::dump_indent2(index)).map_err(|e| e.to_string())
+}
+
+/// Shard file name for a topic (oracle: replace(' ', '_').replace('/', '_')).
 pub fn shard_name(topic: &str) -> String {
     format!("shard_{}.db", topic.replace([' ', '/'], "_"))
 }
 
-/// Get the full shard path for a topic
-pub fn shard_path(shard_dir: &str, topic: &str) -> String {
-    let name = shard_name(topic);
-    format!("{}/{}", shard_dir.trim_end_matches('/'), name)
+/// create_shard: schema creation via the ensure_schema op; early-exists
+/// branch returns the path WITHOUT touching the index (oracle line 52-53).
+/// On creation the index is updated and persisted (oracle lines 88-89).
+pub fn create_shard(
+    shard_dir: &str,
+    index_path: &str,
+    topic: &str,
+    index: &Json,
+    cb: OpsCb,
+) -> Result<(String, Json), String> {
+    let shard_path = join_path(shard_dir, &shard_name(topic));
+    if std::path::Path::new(&shard_path).exists() {
+        return Ok((shard_path, index.clone()));
+    }
+    issue_op(cb, OP_ENSURE_SCHEMA, Json::Obj(vec![(
+        "path".into(),
+        Json::Str(shard_path.clone()),
+    )]))
+    .ok_or_else(|| "ensure_schema op failed".to_string())?;
+    let mut idx = index.clone();
+    index_set(&mut idx, topic, &shard_path);
+    save_index(index_path, &idx)?;
+    Ok((shard_path, idx))
 }
 
-/// Check if a shard exists
-pub fn shard_exists(shard_dir: &str, topic: &str) -> bool {
-    std::path::Path::new(&shard_path(shard_dir, topic)).exists()
+/// route_query: keyword overlap over the index (insertion order, strictly
+/// greater wins → first topic on ties), then the >3-char 3-word fallback.
+pub fn route_query(query: &str, index: &Json) -> String {
+    let query_lower = query.to_lowercase();
+    let query_words: HashSet<&str> = query_lower.split_whitespace().collect();
+
+    let mut best_topic: Option<&str> = None;
+    let mut best_score: i64 = 0;
+    if let Json::Obj(pairs) = index {
+        for (topic, _) in pairs {
+            let topic_lower = topic.to_lowercase();
+            let topic_words: HashSet<&str> = topic_lower.split('_').collect();
+            let overlap = topic_words.intersection(&query_words).count() as i64;
+            if overlap > best_score {
+                best_score = overlap;
+                best_topic = Some(topic.as_str());
+            }
+        }
+    }
+    if let (Some(topic), true) = (best_topic, best_score > 0) {
+        return topic.to_string();
+    }
+
+    let words: Vec<&str> = query
+        .split_whitespace()
+        .filter(|w| w.chars().count() > 3)
+        .take(3)
+        .collect();
+    if !words.is_empty() {
+        return words.join("_").to_lowercase();
+    }
+    "general".to_string()
 }
 
+fn merge_key(x: &Json) -> f64 {
+    for key in ["relevance", "wash_count"] {
+        if let Some(Json::Num(n)) = x.get(key) {
+            return *n;
+        }
+    }
+    0.0
+}
+
+/// merge_results: dedupe by text[:50] (chars), stable sort by
+/// relevance-else-wash_count-else-0, descending.
+pub fn merge_results(results_list: &Json) -> Json {
+    let mut seen: HashSet<String> = HashSet::new();
+    let mut merged: Vec<Json> = Vec::new();
+    if let Json::Arr(shards) = results_list {
+        for results in shards {
+            if let Json::Arr(rows) = results {
+                for r in rows {
+                    let key: String = match r.get("text") {
+                        Some(Json::Str(s)) => s.chars().take(50).collect(),
+                        _ => String::new(),
+                    };
+                    if seen.insert(key) {
+                        merged.push(r.clone());
+                    }
+                }
+            }
+        }
+    }
+    merged.sort_by(|a, b| {
+        merge_key(b)
+            .partial_cmp(&merge_key(a))
+            .unwrap_or(std::cmp::Ordering::Equal)
+    });
+    Json::Arr(merged)
+}
+
+/// get_shard: resolve-or-create the shard path (connection opening stays
+/// Python-side). Creation path mirrors create_shard incl. index persistence.
+pub fn get_shard(
+    shard_dir: &str,
+    index_path: &str,
+    topic: &str,
+    index: &Json,
+    cb: OpsCb,
+) -> Result<(String, Json), String> {
+    let existing = index.get(topic).and_then(|v| match v {
+        Json::Str(s) => Some(s.clone()),
+        _ => None,
+    });
+    if let Some(path) = existing {
+        return Ok((path, index.clone()));
+    }
+    create_shard(shard_dir, index_path, topic, index, cb)
+}
+
+/// get_shard_stats: one count_entries op per topic, index order; op failure
+/// degrades to 0 (oracle except branch).
+pub fn get_shard_stats(index: &Json, cb: OpsCb) -> Result<Json, String> {
+    let mut stats = Vec::new();
+    if let Json::Obj(pairs) = index {
+        for (topic, path) in pairs {
+            let path = match path {
+                Json::Str(s) => s.clone(),
+                _ => String::new(),
+            };
+            let count = issue_op(cb, OP_COUNT_ENTRIES, Json::Obj(vec![(
+                "path".into(),
+                Json::Str(path),
+            )]))
+            .and_then(|v| match v {
+                Json::Num(n) => Some(n as i64),
+                _ => None,
+            })
+            .unwrap_or(0);
+            stats.push((topic.clone(), Json::Num(count as f64)));
+        }
+    }
+    Ok(Json::Obj(stats))
+}
+
+/// total_entries: sum of get_shard_stats values.
+pub fn total_entries(index: &Json, cb: OpsCb) -> Result<i64, String> {
+    let stats = get_shard_stats(index, cb)?;
+    let mut total: i64 = 0;
+    if let Json::Obj(pairs) = &stats {
+        for (_, v) in pairs {
+            if let Json::Num(n) = v {
+                total += *n as i64;
+            }
+        }
+    }
+    Ok(total)
+}
+
+// ── FFI surface ──────────────────────────────────────────────────────────────
+
+#[no_mangle]
+pub extern "C" fn sm_init(doc_ptr: *const c_char) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let dir = str_field(&doc, "shard_dir")?;
+        let idx_path = str_field(&doc, "index_path")?;
+        init(dir, idx_path)
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_load_index(doc_ptr: *const c_char) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        load_index(str_field(&doc, "index_path")?)
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_save_index(doc_ptr: *const c_char) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let idx_path = str_field(&doc, "index_path")?;
+        let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new()));
+        save_index(idx_path, &index)?;
+        Ok(Json::Null)
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_create_shard(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let dir = str_field(&doc, "shard_dir")?;
+        let idx_path = str_field(&doc, "index_path")?;
+        let topic = str_field(&doc, "topic")?;
+        let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new()));
+        let (shard_path, idx) = create_shard(dir, idx_path, topic, &index, cb)?;
+        Ok(Json::Obj(vec![
+            ("shard_path".into(), Json::Str(shard_path)),
+            ("index".into(), idx),
+        ]))
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_route_query(doc_ptr: *const c_char) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let query = str_field(&doc, "query")?;
+        let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new()));
+        Ok(Json::Str(route_query(query, &index)))
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_merge_results(doc_ptr: *const c_char) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let results_list = doc.get("results_list").cloned().unwrap_or(Json::Arr(Vec::new()));
+        Ok(merge_results(&results_list))
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_get_shard(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let dir = str_field(&doc, "shard_dir")?;
+        let idx_path = str_field(&doc, "index_path")?;
+        let topic = str_field(&doc, "topic")?;
+        let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new()));
+        let (shard_path, idx) = get_shard(dir, idx_path, topic, &index, cb)?;
+        Ok(Json::Obj(vec![
+            ("shard_path".into(), Json::Str(shard_path)),
+            ("index".into(), idx),
+        ]))
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_get_shard_stats(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new()));
+        get_shard_stats(&index, cb)
+    };
+    envelope(run())
+}
+
+#[no_mangle]
+pub extern "C" fn sm_total_entries(doc_ptr: *const c_char, cb: OpsCb) -> *mut c_char {
+    let run = || -> Result<Json, String> {
+        let doc = read_arg(doc_ptr)?;
+        let index = doc.get("index").cloned().unwrap_or(Json::Obj(Vec::new()));
+        Ok(Json::Num(total_entries(&index, cb)? as f64))
+    };
+    envelope(run())
+}
+
+fn str_field<'a>(doc: &'a Json, key: &str) -> Result<&'a str, String> {
+    match doc.get(key) {
+        Some(Json::Str(s)) => Ok(s),
+        _ => Err(format!("missing string field '{}'", key)),
+    }
+}
+
+// ── frozen-truth battery ─────────────────────────────────────────────────────
+
+#[path = "branch_battery.rs"]
 #[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn test_shard_name() {
-        assert_eq!(shard_name("deployment"), "shard_deployment.db");
-        assert_eq!(shard_name("some topic"), "shard_some_topic.db");
-        assert_eq!(shard_name("a/b"), "shard_a_b.db");
-    }
-
-    #[test]
-    fn test_shard_path() {
-        let path = shard_path("/tmp/shards", "test");
-        assert!(path.contains("/tmp/shards/shard_test.db"));
-    }
-
-    #[test]
-    fn test_shard_exists() {
-        let dir = "/tmp/shard_test_dir";
-        std::fs::create_dir_all(dir).unwrap();
-        let path = shard_path(dir, "existing");
-        std::fs::write(&path, b"").unwrap();
-        assert!(shard_exists(dir, "existing"));
-        assert!(!shard_exists(dir, "nonexistent"));
-        std::fs::remove_dir_all(dir).ok();
-    }
-}
-
-#[cfg(test)]
-mod branch_battery {
-    use super::*;
-
-    #[test]
-    fn shard_name_format() {
-        // ORACLE: shard_name returns "shard_<Topic>.db" format
-        assert_eq!(shard_name("Memory"), "shard_Memory.db");
-    }
-
-    #[test]
-    fn shard_path_correct() {
-        assert_eq!(
-            shard_path("/tmp/shards", "test"),
-            "/tmp/shards/shard_test.db"
-        );
-    }
-
-    #[test]
-    fn shard_exists_false_for_missing() {
-        assert!(!shard_exists("/tmp/nonexistent-shards-xyz", "test"));
-    }
-}
+mod branch_battery;

tests/test_shard_manager.py

@@ -22,11 +22,15 @@ SPEC.loader.exec_module(sm_mod)
 
 @pytest.fixture
 def shard_env(tmp_path, monkeypatch):
-    """Point ShardManager's SHARD_DIR at a temp home — no importlib.reload
-    (module-constant patching is deterministic and side-effect free)."""
+    """Point ShardManager's SHARD_DIR/SHARD_INDEX at a temp home — no
+    importlib.reload (module-constant patching is deterministic and
+    side-effect free). SHARD_INDEX must be patched in the Rust-core mode too:
+    the kernel receives the index path explicitly (no _load_index indirection
+    to monkeypatch)."""
     shard_dir = tmp_path / "shards"
     shard_dir.mkdir()
     monkeypatch.setattr(sm_mod, "SHARD_DIR", str(shard_dir))
+    monkeypatch.setattr(sm_mod, "SHARD_INDEX", str(tmp_path / "shard_index.json"))
     # fresh index per test
     monkeypatch.setattr(sm_mod.ShardManager, "_load_index", lambda self: {})
     monkeypatch.setattr(sm_mod.ShardManager, "_save_index", lambda self: None)