feat(spine): close PARTIAL gaps — dichotomy wave

Port the 5 audit-missing fns name-preserving into src/rust/spine/src/gap.rs
(_module_imports, _gather_meta_inputs, _deployed_source_files,
_gather_drift_inputs, _fresh_doc; IO behind injectable seams) plus
machinery twins over the proven rune_spine kernels, with doc-envelope
sp_* FFI (freed via cs_free_string). Audit spine 20/25 -> 25/25
(fn-level 1335 -> 1340). Battery replays the 21-case harness truth
(byte-verified vs live oracle) + 35-case clocked battery VERBATIM;
live parity 74/74 through the real FFI surface; clippy -D warnings
clean; pytest no new failures either pin. Fixes pre-landing: clock-pin
args leak (structural_store.now passes gmtime() positionally),
malformed-import SyntaxError branch, test_ prefix in drift tested
stems, kernel_absent battery dispatch.

scripts/check_spine_gap_parity.py

@@ -0,0 +1,265 @@
+#!/usr/bin/env python3
+"""Live parity: the spine gap kernels over their REAL FFI surface (the
+src/rust/spine cdylib, sp_* doc-envelope symbols) vs the live Python
+oracle pinned RUNE_SPINE_CORE=python. Dichotomy wave #2479.
+
+Replays the SAME corpus the frozen battery replays (the harness cases +
+the clocked sandbox cases from gen_spine_gap_truth.py — regenerated LIVE
+here, not read from the frozen file: ladder layer 4 wants port vs oracle,
+frozen files prove layer 3). Returned pointers are freed through
+cs_free_string (capability_scoring owns the free symbol — loader ABI law;
+both cdylibs resolve the same process allocator).
+
+Comparison law (branch_battery_gap.rs): key-order-insensitive canon
+(spine._py_sorted on both sides) + byte-equal dumps; array order exact;
+error messages exact modulo the loader's "sp_* core error: " envelope
+prefix, which the oracle adds when the stub routes through spine_loader.
+"""
+from __future__ import annotations
+
+import ctypes
+import json
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "scripts"))
+sys.path.insert(0, str(REPO / "src"))
+
+# Importing the generator pins RUNE_SPINE_CORE=python (poisoned-truth law)
+# and freezes the process clock at FROZEN_NOW — the oracle side below runs
+# under exactly the pins that froze the truth.
+import gen_spine_gap_truth as gen  # noqa: E402
+
+from rune import spine  # noqa: E402
+
+SPINE_SO = REPO / "src" / "rust" / "spine" / "target" / "release" / "libspine.so"
+CS_SO = (
+    REPO
+    / "src"
+    / "rust"
+    / "capability_scoring"
+    / "target"
+    / "release"
+    / "libcapability_scoring.so"
+)
+
+SYMS = (
+    "sp_module_imports", "sp_gather_meta_inputs", "sp_deployed_source_files",
+    "sp_gather_drift_inputs", "sp_fresh_doc", "sp_rust_core",
+    "sp_py_sorted", "sp_py_declared_surfaces", "sp_py_project_meta_surfaces",
+    "sp_py_ingest_umbrellas_data", "sp_py_entity", "sp_py_build_from_inputs",
+    "sp_py_drift_from_inputs", "sp_py_export_umbrellas", "sp_py_query",
+)
+
+_spine = ctypes.CDLL(str(SPINE_SO))
+_cs = ctypes.CDLL(str(CS_SO))
+_cs.cs_free_string.argtypes = [ctypes.c_void_p]
+_cs.cs_free_string.restype = None
+for _s in SYMS:
+    _fn = getattr(_spine, _s)
+    _fn.argtypes = [ctypes.c_char_p]
+    _fn.restype = ctypes.c_void_p  # char* — never the default c_int
+
+
+def call(sym: str, payload: dict):
+    fn = getattr(_spine, sym)
+    ptr = fn(json.dumps(payload).encode("utf-8"))
+    if not ptr:
+        raise RuntimeError(f"{sym} returned a null pointer")
+    try:
+        text = ctypes.cast(ptr, ctypes.c_char_p).value.decode("utf-8")
+    finally:
+        _cs.cs_free_string(ptr)
+    env = json.loads(text)
+    if not env.get("ok"):
+        raise RuntimeError(env.get("err"))
+    return env["value"]
+
+
+def canon(v) -> str:
+    return json.dumps(spine._py_sorted(v), sort_keys=True)
+
+
+def strip_loader_prefix(msg: str) -> str:
+    for p in ("sp_project core error: ", "sp_drift core error: ",
+              "sp_export core error: ", "sp_query core error: "):
+        if msg.startswith(p):
+            return msg[len(p):]
+    return msg
+
+
+PASS = 0
+FAIL: list[str] = []
+
+
+def check(label: str, ok: bool, detail: str = ""):
+    global PASS
+    if ok:
+        PASS += 1
+    else:
+        FAIL.append(f"{label}: {detail}")
+
+
+def compare_case(label: str, rust_out, case: dict):
+    if "error" in case:
+        want = case["error"]["message"]
+        if isinstance(rust_out, RuntimeError):
+            check(label, strip_loader_prefix(want) == str(rust_out),
+                  f"oracle={want!r} rust={rust_out!r}")
+        else:
+            check(label, False, f"oracle raised {want!r}, rust returned {rust_out!r}")
+        return
+    want = case["expected"]
+    if "post_doc" in case:
+        # doc-mutating: oracle return is the count; the doc mutation is
+        # frozen separately — compare the kernel's {count, doc} wrapper.
+        check(label, isinstance(rust_out, dict) and "count" in rust_out and "doc" in rust_out,
+              f"rust wrapper shape: {rust_out!r}"[:200])
+        if isinstance(rust_out, dict) and "count" in rust_out:
+            check(label + "/count", rust_out["count"] == want,
+                  f"oracle={want!r} rust={rust_out['count']!r}")
+            check(label + "/doc", canon(rust_out["doc"]) == canon(case["post_doc"]),
+                  "doc divergence")
+        return
+    check(label, canon(rust_out) == canon(want),
+          f"oracle={canon(want)[:300]} rust={canon(rust_out)[:300]}")
+
+
+def sandbox(files: dict) -> Path:
+    root = Path(tempfile.mkdtemp(prefix="spine-gap-parity-"))
+    for rel, content in files.items():
+        p = root / rel
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(content)
+    return root
+
+
+def rust_replay(case: dict):
+    """Mirror of the battery dispatch — through the FFI symbols."""
+    fn, name, args = case["fn"], case["name"], case["args"]
+    root = sandbox(case.get("files") or {})
+    try:
+        if fn == "_module_imports":
+            module = args[0]
+            text = (root / "src" / "rune" / f"{module}.py").read_text() \
+                if (root / "src" / "rune" / f"{module}.py").is_file() else None
+            return call("sp_module_imports",
+                        {"module": module, "src_text": text})
+        if fn == "_gather_meta_inputs":
+            return call("sp_gather_meta_inputs", {"root": str(root)})
+        if fn == "_deployed_source_files":
+            return call("sp_deployed_source_files",
+                        {"root": str(root / "deployed")})
+        if fn == "_rust_core":
+            # both frozen branches: loader importable, .so absent
+            return call("sp_rust_core",
+                        {"pin": "auto", "loader_importable": True,
+                         "so_present": False})
+        if fn == "_py_project_meta_surfaces":
+            return call("sp_py_project_meta_surfaces",
+                        {"doc": args[0], "inputs": args[1],
+                         "now": gen.STAMP})
+        if fn == "_py_ingest_umbrellas_data":
+            return call("sp_py_ingest_umbrellas_data",
+                        {"doc": args[0], "data": args[1], "now": gen.STAMP})
+        if fn == "_py_build_from_inputs":
+            core_up = name != "kernel_absent_raises"
+            return call("sp_py_build_from_inputs",
+                        {"inputs": args[0], "core_available": core_up})
+        if fn == "_gather_drift_inputs":
+            tested = sorted(
+                p.stem[len("test_"):]
+                for p in (root / "tests").glob("test_*.py")
+            )
+            src_stems = [p.stem for p in (root / "src" / "rune").glob("*.py")]
+            return call("sp_gather_drift_inputs", {
+                "root": str(root),
+                "registry": ["canary_alpha", "canary_beta"],
+                "graph": {"alpha": ["json", "rune"], "beta": ["sys"]},
+                "tested": tested,
+                "src_stems": src_stems,
+                "deployed_obs": None,
+            })
+        if fn == "_py_drift_from_inputs":
+            return call("sp_py_drift_from_inputs",
+                        {"doc": args[0], "inputs": args[1],
+                         "core_available": False})
+        if fn == "_py_export_umbrellas":
+            return call("sp_py_export_umbrellas",
+                        {"doc": args[0], "now": gen.STAMP,
+                         "core_available": False})
+        if fn == "_py_query":
+            return call("sp_py_query",
+                        {"doc": args[0], "id": args[1],
+                         "core_available": False})
+        if fn == "_fresh_doc":
+            return call("sp_fresh_doc", {"root": str(root)})
+        raise AssertionError(f"unknown fn {fn}")
+    except RuntimeError as e:
+        return e
+    finally:
+        import shutil
+        shutil.rmtree(root, ignore_errors=True)
+
+
+def main() -> int:
+    # 1. harness corpus (arg-pure) — oracle re-run LIVE via run_oracle.
+    for fn_name, args, kwargs in gen.harness_cases():
+        case = gen.run_oracle(fn_name, list(args), dict(kwargs))
+        case = {"expected": case.get("expected"), **case}
+        label = f"harness/{fn_name}:{json.dumps(gen.snap(list(args)))[:60]}"
+        if fn_name == "_py_sorted":
+            rust = call("sp_py_sorted", {"v": args[0]})
+        elif fn_name == "_py_declared_surfaces":
+            rust = call("sp_py_declared_surfaces", {"manifest": args[0]})
+        elif fn_name == "_py_entity":
+            rust = call("sp_py_entity", {"doc": args[0], "id": args[1]})
+        elif fn_name == "_rust_core":
+            rust = call("sp_rust_core",
+                        {"pin": "python", "loader_importable": True,
+                         "so_present": True})
+        else:
+            raise AssertionError(fn_name)
+        compare_case(label, rust, case)
+
+    # 2. clocked corpus — clocked_cases() re-runs the live oracle per case.
+    for case in gen.clocked_cases():
+        label = f"clocked/{case['fn']}/{case['name']}"
+        rust = rust_replay(case)
+        compare_case(label, rust, case)
+
+    # 3. the branches the frozen corpus cannot carry.
+    r = call("sp_rust_core", {"pin": "auto", "loader_importable": True,
+                              "so_present": True})
+    check("extra/rust_core_auto_present", r == "spine_loader", repr(r))
+    r = call("sp_rust_core", {"pin": "auto", "loader_importable": False,
+                              "so_present": True})
+    check("extra/rust_core_no_loader", r is None, repr(r))
+
+    # 4. _fresh_doc build-and-save seam: injected built doc returned AND
+    #    persisted; the persisted bytes cross-checked against the PYTHON
+    #    oracle's canonical save text (spine.save under the pin).
+    root = sandbox({})
+    built = gen.seeded_doc()
+    got = call("sp_fresh_doc", {"root": str(root), "built": built})
+    check("extra/fresh_doc_builds", canon(got) == canon(built), "doc mismatch")
+    persisted = (root / "store" / "spine" / "current.json").read_text()
+    oracle_path = root / "oracle_save.json"
+    spine.save(gen.snap(built), oracle_path)
+    check("extra/fresh_doc_save_text", persisted == oracle_path.read_text(),
+          "persisted text != python oracle save text")
+    import shutil
+    shutil.rmtree(root, ignore_errors=True)
+
+    total = PASS + len(FAIL)
+    print(f"spine gap live parity: {PASS}/{total} pass"
+          f"{'' if not FAIL else ' — FAILURES:'}")
+    for f in FAIL:
+        print("  FAIL", f)
+    return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/gen_spine_gap_truth.py

@@ -0,0 +1,515 @@
+#!/usr/bin/env python3
+"""gen_spine_gap_truth.py — spine gap-closing truth battery (contract
+BUILDER-2026-10-13T08:11:47Z-9b1f32-factory-gap-sp).
+
+Closes the PARTIAL audit gaps: the 8 audit-named fns (_py_sorted,
+_module_imports, _py_declared_surfaces, _py_project_meta_surfaces,
+_gather_meta_inputs, _rust_core, _deployed_source_files,
+_py_build_from_inputs) plus their 7 remaining audit-missing siblings
+(_py_ingest_umbrellas_data, _gather_drift_inputs, _py_drift_from_inputs,
+_py_export_umbrellas, _py_query, _py_entity, _fresh_doc) — 35/35.
+
+TWO artifacts, two laws:
+
+1. frozen_spine_gap_truth.json — SHARED-HARNESS truth (scripts/
+   freeze_truth.py): arg-pure, stamp-free, JSON-safe cases only. The
+   harness re-runs the live oracle per case with NO monkeypatching, so
+   only functions of the recorded args qualify. Byte-verified:
+
+     PYTHONPATH=src python3 scripts/freeze_truth.py \
+         --verify src/rust/spine/frozen_spine_gap_truth.json \
+         --module rune.spine
+
+2. frozen_spine_gap_clocked.json — the sandbox/clock/error battery
+   (rune_spine facade-battery + spine_language law): cases whose oracle
+   needs module-global IO roots (SRC_DIR/ROOT/WORKFLOWS_DIR/DEPLOYED_LINK
+   → per-case temp sandboxes, fixture bytes recorded back as `files`),
+   the pinned process clock (structural_store.now → time.strftime
+   patched to FROZEN_NOW), set→sorted-list conversion (_module_imports),
+   and exception capture as {"error": {"type", "message"}}. Replayed
+   VERBATIM by src/rust/spine/src/branch_battery_gap.rs.
+
+RUNE_SPINE_CORE=python pins the pure-oracle path (poisoned-truth law);
+the _py_build_from_inputs group runs UNPINNED — that stub is
+delegation-only since 2026-10-10 and raises without the kernel.
+
+Usage: python3 scripts/gen_spine_gap_truth.py   (from repo root)
+"""
+
+import copy
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "scripts"))
+sys.path.insert(0, str(REPO / "src"))
+
+from freeze_truth import freeze, write_envelope  # noqa: E402
+
+FROZEN_NOW = "2026-10-13T08:30:00"
+STAMP = FROZEN_NOW + "Z"
+
+os.environ["RUNE_FROZEN_NOW"] = FROZEN_NOW
+os.environ["RUNE_SPINE_CORE"] = "python"  # poisoned-truth law (pure oracle)
+
+import time  # noqa: E402
+
+import rune.structural_store as ss  # noqa: E402
+from rune import spine  # noqa: E402
+
+# Pin the wall clock at the source: structural_store.now/add_entity stamp
+# observed_at via time.strftime — freeze it process-wide so clocked cases
+# are byte-stable (other formats pass through to the real clock).
+_real_strftime = time.strftime
+
+
+def _pinned_strftime(fmt="%a %b %d %H:%M:%S %Y", *args):
+    # structural_store.now/spine.now pass time.gmtime() positionally —
+    # the pin must hold for the stamp format REGARDLESS of args, or the
+    # frozen truth carries the real wall clock (byte-unstable).
+    if fmt == "%Y-%m-%dT%H:%M:%SZ":
+        return STAMP
+    if args:
+        return _real_strftime(fmt, *args)
+    return _real_strftime(fmt)
+
+
+time.strftime = _pinned_strftime
+
+GENERATED_VIEW_MARKER = spine.GENERATED_VIEW_MARKER
+
+# ── fixtures ────────────────────────────────────────────────────────────
+
+ALPHA_SRC = '''"""Alpha module."""
+import json
+import os.path
+from rune import spine
+from . import sibling
+from .helpers import thing
+'''
+BETA_SRC = '''import sys  # trailing comment
+from rune.structural_store import add_entity as ae, promote
+try:
+    import optional_dep
+except ImportError:
+    optional_dep = None
+'''
+GAMMA_SRC = "x = 1\nimport  [oops\n"
+EMPTY_SRC = ""
+PLAIN_SRC = "VALUE = 42\n"
+SEMI_SRC = "import json; import os.path\nfrom rune import spine, query\n"
+CONT_SRC = (
+    "from rune.structural_store import (\n"
+    "    add_entity,\n"
+    "    promote,\n"
+    ")\n"
+    "import a.b as c\n"
+)
+
+SRC_FILES = {
+    "src/rune/alpha.py": ALPHA_SRC,
+    "src/rune/beta.py": BETA_SRC,
+    "src/rune/gamma.py": GAMMA_SRC,
+    "src/rune/empty.py": EMPTY_SRC,
+    "src/rune/plain.py": PLAIN_SRC,
+    "src/rune/semi.py": SEMI_SRC,
+    "src/rune/cont.py": CONT_SRC,
+}
+
+
+def umbrella_manifest() -> dict:
+    return {"umbrellas": {
+        "cognition": {
+            "description": "thinking parts",
+            "modules": ["alpha"],
+            "rust_crates": ["kernel_a", "ghost_crate"],
+            "go_surfaces": ["rune-entry"],
+            "configs": ["go-version.txt", "phantom.toml"],
+        },
+        "governance": {"description": "meta"},
+    }}
+
+
+def meta_tree() -> dict:
+    return {
+        "config/umbrellas.json": json.dumps(umbrella_manifest()),
+        "src/rust/kernel_a/lib.rs": "// crate\n",
+        "src/rust/kernel_a/canary_kernel_a.py": "# canary\n",
+        "src/go/rune-entry/main.go": "package main\n",
+        "go-version.txt": "1.24\n",
+        ".gitea/workflows/ci.yml": "on: push\n",
+    }
+
+
+def deployed_tree() -> dict:
+    return {
+        "src/rune/alpha.py": ALPHA_SRC,
+        "src/rust/kernel_a/canary_kernel_a.py": "# canary\n",
+        "src/rust/kernel_a/lib.rs": "// not a canary\n",
+        "README.md": "# not a source\n",
+    }
+
+
+def drift_tree() -> dict:
+    return {
+        "pyproject.toml": '[project]\nname = "rune"\nversion = "9.9.9"\n',
+        "src/rune/alpha.py": ALPHA_SRC,
+        "src/rune/beta.py": BETA_SRC,
+        "tests/test_alpha.py": "# test\n",
+        "tests/test_beta.py": "# test\n",
+    }
+
+
+def fresh_doc() -> dict:
+    doc = ss.new_doc()
+    doc["generated_at"] = STAMP  # deterministic under the pinned clock
+    return doc
+
+
+def meta_inputs(over: dict | None = None) -> dict:
+    base = {
+        "config_manifest": umbrella_manifest(),
+        "crate_dirs": ["kernel_a"],
+        "crate_canaries": {"kernel_a": ["canary_kernel_a"]},
+        "go_dirs": ["rune-entry"],
+        "config_files": ["go-version.txt"],
+        "workflows": ["ci.yml"],
+    }
+    base.update(over or {})
+    return base
+
+
+def build_inputs(over: dict | None = None) -> dict:
+    base = {
+        "now": STAMP,
+        "existing": None,
+        "manifest": umbrella_manifest(),
+        "manifest_path": "config/umbrellas.json",
+        "graph": {"alpha": ["json", "os", "rune"], "beta": ["sys"]},
+        "registry": ["canary_alpha"],
+        "tested": ["alpha"],
+        "main_present": True,
+        "present_modules": ["alpha", "beta"],
+        "deployed_obs": None,
+        "deployed_sources": [],
+        "canary_files": ["canary_alpha"],
+    }
+    base.update(meta_inputs())
+    base.update(over or {})
+    return base
+
+
+def seeded_doc() -> dict:
+    doc = fresh_doc()
+    doc["relationships"].append({
+        "type": "belongs_to", "from": "module:alpha",
+        "to": "subsystem:cognition", "evidence_refs": ["seed"],
+        "observed_at": STAMP,
+    })
+    return doc
+
+
+# ── sandbox machinery ───────────────────────────────────────────────────
+
+_SAVED_GLOBALS = (spine.SRC_DIR, spine.ROOT, spine.WORKFLOWS_DIR,
+                  spine.DEPLOYED_LINK, spine.TESTS_DIR, spine.CURRENT,
+                  spine.STORE_DIR, spine.SNAPSHOTS)
+
+
+def _sandbox(files: dict) -> Path:
+    root = Path(tempfile.mkdtemp(prefix="spine-gap-truth-"))
+    for rel, content in files.items():
+        p = root / rel
+        p.parent.mkdir(parents=True, exist_ok=True)
+        p.write_text(content)
+    return root
+
+
+def _point_globals_at(root: Path) -> None:
+    spine.ROOT = root
+    spine.SRC_DIR = root / "src" / "rune"
+    spine.WORKFLOWS_DIR = root / ".gitea" / "workflows"
+    spine.TESTS_DIR = root / "tests"
+    spine.DEPLOYED_LINK = root / "deployed"
+    spine.STORE_DIR = root / "store" / "spine"
+    spine.CURRENT = spine.STORE_DIR / "current.json"
+    spine.SNAPSHOTS = spine.STORE_DIR / "snapshots"
+
+
+def _restore_globals() -> None:
+    (spine.SRC_DIR, spine.ROOT, spine.WORKFLOWS_DIR, spine.DEPLOYED_LINK,
+     spine.TESTS_DIR, spine.CURRENT, spine.STORE_DIR, spine.SNAPSHOTS) = _SAVED_GLOBALS
+
+
+def snap(value):
+    """JSON-safe snapshot: sets → sorted lists; everything else deep."""
+    if isinstance(value, (set, frozenset)):
+        return sorted(value)
+    if isinstance(value, dict):
+        return {k: snap(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [snap(v) for v in value]
+    return copy.deepcopy(value)
+
+
+def run_oracle(fn_name: str, args: list, kwargs: dict, live_doc: dict | None = None):
+    """Call the oracle fn, capturing results OR exceptions. live_doc (the
+    doc-mutating cases) is passed by reference — the caller snapshots it
+    AFTER the call to freeze the in-place mutation contract."""
+    fn = getattr(spine, fn_name)
+    try:
+        if live_doc is not None:
+            result = fn(live_doc, *copy.deepcopy(args), **copy.deepcopy(kwargs))
+        else:
+            result = fn(*copy.deepcopy(args), **copy.deepcopy(kwargs))
+        return {"expected": snap(result)}
+    except Exception as e:  # noqa: BLE001 — the error IS the frozen contract
+        return {"error": {"type": type(e).__name__, "message": str(e)}}
+
+
+# ── part 1: shared-harness truth (arg-pure cases) ───────────────────────
+
+def harness_cases() -> list[tuple]:
+    cases: list[tuple] = []
+
+    for v in [
+        {"b": 1, "a": {"z": 2, "c": 3}},
+        [{"a": 2, "b": 1}, {"a": 9}, {"a": 1, "b": 1}],
+        [3, "x", None, True, 1.5, "a"],
+        {"s": "héllo", "i": 7, "f": 2.5, "n": None, "b": False},
+        {"l": [], "d": {}},
+        {"k": [{"z": 1, "a": 2}, {"b": []}]},
+        42,
+        "zzz",
+        None,
+        [{"a": 1}, {"a": 1}, {"a": 2}],
+    ]:
+        cases.append(("_py_sorted", (v,), {}))
+
+    for manifest in [
+        umbrella_manifest(),
+        {"umbrellas": {}},
+        {"umbrellas": {"solo": {"description": "d"}}},
+        {},
+        {"umbrellas": {"u": {
+            "rust_crates": ["k"], "go_surfaces": ["g"], "configs": ["c"],
+            "modules": ["m"], "depends_on": ["d"], "future": 1}}},
+    ]:
+        cases.append(("_py_declared_surfaces", (manifest,), {}))
+
+    ents_doc = {
+        "entities": [
+            {"id": "module:m_a", "kind": "module", "maturity": "WIRED",
+             "aliases": ["module:m_old"]},
+            {"id": "capability:c1", "kind": "capability", "maturity": "DECLARED"},
+        ],
+        "relationships": [],
+    }
+    for eid in ("module:m_a", "module:m_old", "module:ghost", ""):
+        cases.append(("_py_entity", (copy.deepcopy(ents_doc), eid), {}))
+    cases.append(("_py_entity", ({"entities": []}, "x"), {}))
+
+    cases.append(("_rust_core", (), {}))  # harness pins RUNE_SPINE_CORE=python
+    return cases
+
+
+# ── part 2: clocked battery (sandbox/clock/error cases) ─────────────────
+
+def clocked_cases() -> list[dict]:
+    cases: list[dict] = []
+
+    def add(fn: str, name: str, files: dict, args: list,
+            kwargs: dict | None = None, pin: str | None = None,
+            sandboxed: bool = True, mutate_doc: bool = False,
+            env: dict | None = None):
+        root = _sandbox(files) if sandboxed else None
+        if sandboxed:
+            _point_globals_at(root)
+        saved_pin = os.environ.get("RUNE_SPINE_CORE")
+        saved_env = {k: os.environ.get(k) for k in (env or {})}
+        if pin is not None:
+            os.environ["RUNE_SPINE_CORE"] = pin
+        for k, v in (env or {}).items():
+            os.environ[k] = v
+        try:
+            doc_args = list(args)
+            post_doc = None
+            if mutate_doc:
+                live_doc = copy.deepcopy(args[0])
+                out = run_oracle(fn, args[1:], dict(kwargs or {}),
+                                 live_doc=live_doc)
+                post_doc = live_doc
+            else:
+                out = run_oracle(fn, copy.deepcopy(args), dict(kwargs or {}))
+        finally:
+            if pin is not None:
+                os.environ["RUNE_SPINE_CORE"] = saved_pin or "auto"
+            for k, v in saved_env.items():
+                if v is None:
+                    os.environ.pop(k, None)
+                else:
+                    os.environ[k] = v
+            if sandboxed:
+                _restore_globals()
+        entry = {"fn": fn, "name": name, "files": files,
+                 "args": snap(doc_args), "kwargs": snap(kwargs or {})}
+        entry.update(out)
+        if mutate_doc:
+            entry["post_doc"] = snap(post_doc)
+        cases.append(entry)
+
+    # _module_imports — set→sorted-list conversion, SRC_DIR sandbox
+    for name, module in [
+        ("from_and_plain", "alpha"),
+        ("alias_try_except", "beta"),
+        ("syntax_error_empties", "gamma"),
+        ("no_imports", "plain"),
+        ("empty_file", "empty"),
+        ("semicolons", "semi"),
+        ("paren_continuation", "cont"),
+        ("missing_file_empties", "m_ghost_xyz"),
+    ]:
+        add("_module_imports", name, SRC_FILES, [module])
+
+    # _py_project_meta_surfaces — clock-pinned stamps, doc mutation
+    for name, doc, inputs in [
+        ("full_inputs", fresh_doc(), meta_inputs()),
+        ("empty_inputs", fresh_doc(), {}),
+        ("null_manifest", fresh_doc(), {"config_manifest": None}),
+        ("absent_crate_stays_declared", fresh_doc(), meta_inputs({
+            "crate_dirs": [], "crate_canaries": {}})),
+        ("high_water_keeps_wired", None, meta_inputs()),
+    ]:
+        if doc is None:
+            doc = fresh_doc()
+            doc["entities"].append({
+                "id": "rust-crate:kernel_a", "kind": "rust-crate",
+                "maturity": "WIRED", "tags": ["code"],
+                "identities": {"source": "src/rust/kernel_a"},
+                "evidence_refs": ["earlier#ref"], "observed_at": STAMP,
+            })
+        add("_py_project_meta_surfaces", name, {}, [doc, inputs],
+            mutate_doc=True)
+
+    # _gather_meta_inputs — ROOT/WORKFLOWS_DIR sandbox
+    add("_gather_meta_inputs", "sandbox_tree", meta_tree(), [])
+    add("_gather_meta_inputs", "empty_tree", {}, [])
+    add("_gather_meta_inputs", "malformed_manifest",
+        {"config/umbrellas.json": "{oops"}, [])
+
+    # _rust_core — extra branches live at the END of clocked_cases (they
+    # poison SP_RUST_LIB; spine_loader caches _LIB_PATH at first import).
+
+    # _deployed_source_files — DEPLOYED_LINK sandbox
+    add("_deployed_source_files", "sandbox_deployed", {
+        "deployed/src/rune/alpha.py": ALPHA_SRC,
+        "deployed/src/rust/kernel_a/canary_kernel_a.py": "# canary\n",
+        "deployed/src/rust/kernel_a/lib.rs": "// not a canary\n",
+        "deployed/README.md": "# not a source\n",
+    }, [])
+    add("_deployed_source_files", "absent_link", {}, [])
+
+    # _py_build_from_inputs — UNPINNED (delegation-only stub needs kernel)
+    add("_py_build_from_inputs", "seeded_carry_forward", {},
+        [build_inputs({"existing": seeded_doc()})], pin="auto",
+        sandboxed=False)
+    add("_py_build_from_inputs", "unseeded_manifest", {},
+        [build_inputs()], pin="auto", sandboxed=False)
+    ghost = umbrella_manifest()
+    ghost[GENERATED_VIEW_MARKER] = True
+    add("_py_build_from_inputs", "generated_view_refused", {},
+        [build_inputs({"manifest": ghost})], pin="auto", sandboxed=False)
+    add("_py_build_from_inputs", "kernel_absent_raises", {},
+        [build_inputs()], pin="python", sandboxed=False)
+
+    # _py_ingest_umbrellas_data — pinned pure oracle, clock-pinned stamps
+    manifest = umbrella_manifest()
+    add("_py_ingest_umbrellas_data", "full_manifest", {}, [fresh_doc(), manifest],
+        mutate_doc=True)
+    add("_py_ingest_umbrellas_data", "empty_umbrellas", {},
+        [fresh_doc(), {"umbrellas": {}}], mutate_doc=True)
+    ghost_manifest = {"umbrellas": {}, GENERATED_VIEW_MARKER: True}
+    add("_py_ingest_umbrellas_data", "generated_view_refused", {},
+        [fresh_doc(), ghost_manifest], mutate_doc=True)
+
+    # _gather_drift_inputs — sandbox + registry seam
+    real_registry = spine._canary_registry
+    spine._canary_registry = lambda: ["canary_alpha", "canary_beta"]
+    real_graph = spine._import_graph
+    spine._import_graph = lambda: {
+        "alpha": {"json", "rune"}, "beta": {"sys"}}
+    try:
+        add("_gather_drift_inputs", "sandbox_tree", drift_tree(), [])
+        add("_gather_drift_inputs", "no_pyproject", {
+            "src/rune/alpha.py": ALPHA_SRC}, [])
+        add("_gather_drift_inputs", "unversioned_pyproject", {
+            "pyproject.toml": "[project]\nname = 'rune'\n",
+            "src/rune/alpha.py": ALPHA_SRC}, [])
+    finally:
+        spine._canary_registry = real_registry
+        spine._import_graph = real_graph
+
+    # _py_drift_from_inputs / _py_export_umbrellas / _py_query —
+    # delegation-only stubs: kernel-absent error pinned; happy path is
+    # the kernel's own frozen battery (frozen_spine_oracle.json).
+    add("_py_drift_from_inputs", "kernel_absent_raises", {},
+        [fresh_doc(), {}], pin="python", sandboxed=False)
+    add("_py_export_umbrellas", "kernel_absent_raises", {},
+        [fresh_doc()], pin="python", sandboxed=False)
+    add("_py_query", "kernel_absent_raises", {},
+        [fresh_doc(), "module:alpha"], pin="python", sandboxed=False)
+
+    # _fresh_doc — load path frozen; the absent-store build path is the
+    # zero-dep seam (the Python gather feeds the kernel): under the pin
+    # build() hits the retired stub and raises — the exact frozen error.
+    # The Rust port's injected-`built` happy path is a crate unit test.
+    store_doc = seeded_doc()
+    add("_fresh_doc", "existing_store_wins", {
+        "store/spine/current.json": json.dumps(store_doc),
+    }, [])
+    add("_fresh_doc", "absent_store_kernel_required", {}, [])
+
+    # _rust_core — extra branches the harness pin can't reach. MUST run
+    # LAST and poison spine_loader._LIB_PATH directly (the module is
+    # already imported by the kernel cases and caches _LIB_PATH at first
+    # import; env alone would not reach it).
+    from rune import spine_loader  # noqa: E402 — cached module, path patched below
+
+    saved_lib_path = spine_loader._LIB_PATH
+    spine_loader._LIB_PATH = Path("/nonexistent/librune_spine.so")
+    try:
+        add("_rust_core", "auto_missing_so", {}, [], pin="auto", sandboxed=False)
+        add("_rust_core", "weird_pin_is_auto", {}, [], pin="weird-value",
+            sandboxed=False)
+    finally:
+        spine_loader._LIB_PATH = saved_lib_path
+    # (the auto+present branch returns the loader MODULE — not
+    # JSON-freezable; the Rust battery pins it directly as
+    # Some("spine_loader"))
+
+    return cases
+
+
+def main() -> None:
+    # Part 1 — shared-harness truth.
+    truth = freeze("rune.spine", harness_cases(), "rune.spine")
+    out1 = REPO / "src" / "rust" / "spine" / "frozen_spine_gap_truth.json"
+    write_envelope(FROZEN_NOW, truth, out1)
+    print(f"wrote {out1} ({len(truth)} harness cases)")
+
+    # Part 2 — clocked battery.
+    clocked = clocked_cases()
+    out2 = REPO / "src" / "rust" / "spine" / "frozen_spine_gap_clocked.json"
+    out2.write_text(json.dumps(
+        {"frozen_now": FROZEN_NOW, "cases": clocked},
+        indent=2, ensure_ascii=False) + "\n")
+    print(f"wrote {out2} ({len(clocked)} clocked cases)")
+
+    _restore_globals()
+
+
+if __name__ == "__main__":
+    main()

scripts/walk_contract_spine.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Open + walk the #2479 dichotomy-wave construction contract for spine
+to verifying (with an optional SHA argument: through to closed — the full
+lifecycle runs in ONE process, shared-store clobber hazard). Post-
+machinery-exclusion surface: 25 fns, was 20/25.
+
+GAP-CLOSING PORT: spine (975 lines). Extends src/rust/spine with the
+audit-missing kernels: _module_imports, _gather_meta_inputs,
+_deployed_source_files, _gather_drift_inputs, _fresh_doc (+ machinery
+twins over the proven rune_spine kernels) — name-preserving, IO behind
+injectable seams. Rust fns + doc-envelope FFI (sp_* / free via
+cs_free_string — NEVER re-export) + ctypes battery replay.
+"""
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from rune.construction_interface import ConstructionInterface
+
+iface = ConstructionInterface(builder="factory-gap-sp")
+
+r = iface.submit_need(
+    "Close spine PARTIAL gaps (dichotomy factory, issue #2479 tranche): "
+    "extend src/rust/spine with the 8 audit-missing fns (_py_sorted, "
+    "_module_imports, _py_declared_surfaces, _py_project_meta_surfaces, "
+    "_gather_meta_inputs, _rust_core, _deployed_source_files, "
+    "_py_build_from_inputs) name-preserving, logic 1:1, IO behind "
+    "injectable seams — target >=28/35 fn coverage. Rust fns + doc-envelope "
+    "FFI (sp_* / free via cs_free_string — NEVER re-export) + ctypes loader "
+    "mirroring broker_loader.py + facade delegation with "
+    "RUNE_SPINE_CORE=python pin. Truth frozen via scripts/freeze_truth.py "
+    "(RUNE_SPINE_CORE=python pinned, deep-copied inputs, JSON truth "
+    "embedded VERBATIM in cfg(test) battery); byte-identical verify. "
+    "Phase B language-advantage pass.",
+    origin="builder",
+)
+print("submit_need:", json.dumps(r)[:300])
+if not r.get("success"):
+    print("NEED FAILED — aborting"); sys.exit(1)
+cid = r["ids"]["contract_id"]
+print("contract_id:", cid)
+for phase in ("approved", "building", "verifying"):
+    t = iface.transition(cid, phase)
+    print(f"transition->{phase}:", json.dumps(t)[:200])
+    if not t.get("success"):
+        sys.exit(1)
+
+# Close with SHA evidence when given one (full lifecycle in ONE process —
+# close-time re-walk law: DRAFT → approved → building → verifying → closed).
+if len(sys.argv) > 1:
+    sha = sys.argv[1]
+    c = iface.close(cid, observed_evidence=[{
+        "kind": "port_evidence",
+        "detail": f"spine dichotomy gap-close — {sha} feat(spine): close "
+                  "PARTIAL gaps — dichotomy wave (rebased onto "
+                  "gitea/master, pushed). Audit spine 20/25 -> 25/25 "
+                  "(fn-level 1335 -> 1340): the 5 audit-missing fns "
+                  "(_module_imports, _gather_meta_inputs, "
+                  "_deployed_source_files, _gather_drift_inputs, _fresh_doc) "
+                  "ported name-preserving in src/rust/spine/src/gap.rs with "
+                  "doc-envelope FFI (sp_* symbols, freed via cs_free_string). "
+                  "Battery 5/5 replays the 21-case harness truth (byte-"
+                  "verified against the live oracle) + 35-case clocked "
+                  "battery VERBATIM; live parity 74/74 through the real "
+                  "FFI surface (ctypes vs pinned oracle, incl. fresh_doc "
+                  "save-text cross-check); clippy -D warnings clean; "
+                  "pytest no new failures either pin (15+1 pre-existing at "
+                  "HEAD, retired-oracle class). Divergences caught "
+                  "pre-landing: wall-clock leak in the clocked freeze "
+                  "(structural_store.now passes gmtime() positionally — pin "
+                  "now holds for the stamp format regardless of args), "
+                  "malformed-import SyntaxError branch (empty set, not "
+                  "Err), test_ prefix in drift `tested` stems.",
+    }], scope_files=[
+        "src/rust/spine/src/gap.rs",
+        "src/rust/spine/src/lib.rs",
+        "src/rust/spine/src/branch_battery_gap.rs",
+        "src/rust/spine/frozen_spine_gap_truth.json",
+        "src/rust/spine/frozen_spine_gap_clocked.json",
+        "src/rust/spine/Cargo.toml",
+        "src/rust/spine/Cargo.lock",
+        "scripts/gen_spine_gap_truth.py",
+        "scripts/check_spine_gap_parity.py",
+        "scripts/walk_contract_spine.py",
+    ])
+    print("close:", json.dumps(c)[:300])
+    print("CONTRACT", cid, "closed" if c.get("success") else "CLOSE_FAILED")
+    sys.exit(0 if c.get("success") else 1)
+
+print("CONTRACT", cid, "verifying")

src/rust/spine/Cargo.lock

@@ -32,6 +32,10 @@ dependencies = [
  "proc-macro2",
 ]
 
+[[package]]
+name = "rune_spine"
+version = "0.1.0"
+
 [[package]]
 name = "serde"
 version = "1.0.229"
@@ -79,6 +83,7 @@ dependencies = [
 name = "spine"
 version = "0.1.0"
 dependencies = [
+ "rune_spine",
  "serde",
  "serde_json",
 ]

src/rust/spine/Cargo.toml

@@ -9,3 +9,4 @@ crate-type = ["rlib", "cdylib", "staticlib"]
 [dependencies]
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+rune_spine = { path = "../rune_spine" }

src/rust/spine/frozen_spine_gap_truth.json

@@ -0,0 +1,478 @@
+{
+  "frozen_now": "2026-10-13T08:30:00",
+  "cases": [
+    {
+      "fn": "_py_sorted",
+      "args": [
+        {
+          "b": 1,
+          "a": {
+            "z": 2,
+            "c": 3
+          }
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "a": {
+          "c": 3,
+          "z": 2
+        },
+        "b": 1
+      }
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        [
+          {
+            "a": 2,
+            "b": 1
+          },
+          {
+            "a": 9
+          },
+          {
+            "a": 1,
+            "b": 1
+          }
+        ]
+      ],
+      "kwargs": {},
+      "expected": [
+        {
+          "a": 1,
+          "b": 1
+        },
+        {
+          "a": 2,
+          "b": 1
+        },
+        {
+          "a": 9
+        }
+      ]
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        [
+          3,
+          "x",
+          null,
+          true,
+          1.5,
+          "a"
+        ]
+      ],
+      "kwargs": {},
+      "expected": [
+        "a",
+        "x",
+        1.5,
+        3,
+        null,
+        true
+      ]
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        {
+          "s": "héllo",
+          "i": 7,
+          "f": 2.5,
+          "n": null,
+          "b": false
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "b": false,
+        "f": 2.5,
+        "i": 7,
+        "n": null,
+        "s": "héllo"
+      }
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        {
+          "l": [],
+          "d": {}
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "d": {},
+        "l": []
+      }
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        {
+          "k": [
+            {
+              "z": 1,
+              "a": 2
+            },
+            {
+              "b": []
+            }
+          ]
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "k": [
+          {
+            "a": 2,
+            "z": 1
+          },
+          {
+            "b": []
+          }
+        ]
+      }
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        42
+      ],
+      "kwargs": {},
+      "expected": 42
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        "zzz"
+      ],
+      "kwargs": {},
+      "expected": "zzz"
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        null
+      ],
+      "kwargs": {},
+      "expected": null
+    },
+    {
+      "fn": "_py_sorted",
+      "args": [
+        [
+          {
+            "a": 1
+          },
+          {
+            "a": 1
+          },
+          {
+            "a": 2
+          }
+        ]
+      ],
+      "kwargs": {},
+      "expected": [
+        {
+          "a": 1
+        },
+        {
+          "a": 1
+        },
+        {
+          "a": 2
+        }
+      ]
+    },
+    {
+      "fn": "_py_declared_surfaces",
+      "args": [
+        {
+          "umbrellas": {
+            "cognition": {
+              "description": "thinking parts",
+              "modules": [
+                "alpha"
+              ],
+              "rust_crates": [
+                "kernel_a",
+                "ghost_crate"
+              ],
+              "go_surfaces": [
+                "rune-entry"
+              ],
+              "configs": [
+                "go-version.txt",
+                "phantom.toml"
+              ]
+            },
+            "governance": {
+              "description": "meta"
+            }
+          }
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "rust_crates": [
+          [
+            "kernel_a",
+            "cognition"
+          ],
+          [
+            "ghost_crate",
+            "cognition"
+          ]
+        ],
+        "go_surfaces": [
+          [
+            "rune-entry",
+            "cognition"
+          ]
+        ],
+        "configs": [
+          [
+            "go-version.txt",
+            "cognition"
+          ],
+          [
+            "phantom.toml",
+            "cognition"
+          ]
+        ]
+      }
+    },
+    {
+      "fn": "_py_declared_surfaces",
+      "args": [
+        {
+          "umbrellas": {}
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "rust_crates": [],
+        "go_surfaces": [],
+        "configs": []
+      }
+    },
+    {
+      "fn": "_py_declared_surfaces",
+      "args": [
+        {
+          "umbrellas": {
+            "solo": {
+              "description": "d"
+            }
+          }
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "rust_crates": [],
+        "go_surfaces": [],
+        "configs": []
+      }
+    },
+    {
+      "fn": "_py_declared_surfaces",
+      "args": [
+        {}
+      ],
+      "kwargs": {},
+      "expected": {
+        "rust_crates": [],
+        "go_surfaces": [],
+        "configs": []
+      }
+    },
+    {
+      "fn": "_py_declared_surfaces",
+      "args": [
+        {
+          "umbrellas": {
+            "u": {
+              "rust_crates": [
+                "k"
+              ],
+              "go_surfaces": [
+                "g"
+              ],
+              "configs": [
+                "c"
+              ],
+              "modules": [
+                "m"
+              ],
+              "depends_on": [
+                "d"
+              ],
+              "future": 1
+            }
+          }
+        }
+      ],
+      "kwargs": {},
+      "expected": {
+        "rust_crates": [
+          [
+            "k",
+            "u"
+          ]
+        ],
+        "go_surfaces": [
+          [
+            "g",
+            "u"
+          ]
+        ],
+        "configs": [
+          [
+            "c",
+            "u"
+          ]
+        ]
+      }
+    },
+    {
+      "fn": "_py_entity",
+      "args": [
+        {
+          "entities": [
+            {
+              "id": "module:m_a",
+              "kind": "module",
+              "maturity": "WIRED",
+              "aliases": [
+                "module:m_old"
+              ]
+            },
+            {
+              "id": "capability:c1",
+              "kind": "capability",
+              "maturity": "DECLARED"
+            }
+          ],
+          "relationships": []
+        },
+        "module:m_a"
+      ],
+      "kwargs": {},
+      "expected": {
+        "id": "module:m_a",
+        "kind": "module",
+        "maturity": "WIRED",
+        "aliases": [
+          "module:m_old"
+        ]
+      }
+    },
+    {
+      "fn": "_py_entity",
+      "args": [
+        {
+          "entities": [
+            {
+              "id": "module:m_a",
+              "kind": "module",
+              "maturity": "WIRED",
+              "aliases": [
+                "module:m_old"
+              ]
+            },
+            {
+              "id": "capability:c1",
+              "kind": "capability",
+              "maturity": "DECLARED"
+            }
+          ],
+          "relationships": []
+        },
+        "module:m_old"
+      ],
+      "kwargs": {},
+      "expected": {
+        "id": "module:m_a",
+        "kind": "module",
+        "maturity": "WIRED",
+        "aliases": [
+          "module:m_old"
+        ]
+      }
+    },
+    {
+      "fn": "_py_entity",
+      "args": [
+        {
+          "entities": [
+            {
+              "id": "module:m_a",
+              "kind": "module",
+              "maturity": "WIRED",
+              "aliases": [
+                "module:m_old"
+              ]
+            },
+            {
+              "id": "capability:c1",
+              "kind": "capability",
+              "maturity": "DECLARED"
+            }
+          ],
+          "relationships": []
+        },
+        "module:ghost"
+      ],
+      "kwargs": {},
+      "expected": null
+    },
+    {
+      "fn": "_py_entity",
+      "args": [
+        {
+          "entities": [
+            {
+              "id": "module:m_a",
+              "kind": "module",
+              "maturity": "WIRED",
+              "aliases": [
+                "module:m_old"
+              ]
+            },
+            {
+              "id": "capability:c1",
+              "kind": "capability",
+              "maturity": "DECLARED"
+            }
+          ],
+          "relationships": []
+        },
+        ""
+      ],
+      "kwargs": {},
+      "expected": null
+    },
+    {
+      "fn": "_py_entity",
+      "args": [
+        {
+          "entities": []
+        },
+        "x"
+      ],
+      "kwargs": {},
+      "expected": null
+    },
+    {
+      "fn": "_rust_core",
+      "args": [],
+      "kwargs": {},
+      "expected": null
+    }
+  ]
+}

src/rust/spine/src/branch_battery_gap.rs

@@ -0,0 +1,357 @@
+//! branch_battery_gap.rs — spine gap-closing parity battery (#2479).
+//!
+//! Replays the two frozen truth artifacts VERBATIM (include_str! — never
+//! hand-transcribed):
+//!   * frozen_spine_gap_truth.json — the SHARED freeze_truth harness
+//!     output (byte-verified against the live oracle via
+//!     `scripts/freeze_truth.py --verify ... --module rune.spine`).
+//!   * frozen_spine_gap_clocked.json — sandbox/clock/error cases frozen
+//!     from the live oracle with the process clock pinned and the IO
+//!     roots sandboxed; each case carries its fixture `files` back so
+//!     this battery materializes the identical tree.
+//!
+//! Comparison law (rune_spine facade-battery precedent): key-order
+//! insensitive via store::sorted_deep on BOTH sides, then pyjson::dumps
+//! byte-equality; array order exact. Error cases must carry the oracle's
+//! EXACT message; the one documented transformation is the loader
+//! wrapper: the oracle's delegated-path RuntimeError reads
+//! "sp_project core error: <kernel message>" — the loader adds that
+//! prefix when unwrapping the envelope, so it is stripped before the
+//! kernel-message comparison.
+
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use crate::gap;
+use rune_spine::pyjson::{self, Json};
+use rune_spine::store;
+
+const HARNESS_TRUTH: &str = include_str!("../frozen_spine_gap_truth.json");
+const CLOCKED_TRUTH: &str = include_str!("../frozen_spine_gap_clocked.json");
+
+fn parse(s: &str) -> Json {
+    pyjson::parse(s).unwrap_or_else(|e| panic!("parse failed: {} :: {}", e, s))
+}
+
+fn canon(v: &Json) -> String {
+    pyjson::dumps(&store::sorted_deep(v))
+}
+
+fn expect_field<'a>(case: &'a Json, key: &str) -> &'a Json {
+    case.get(key)
+        .unwrap_or_else(|| panic!("case.{} missing", key))
+}
+
+fn materialize(files: &Json, root: &Path) {
+    if let Some(Json::Obj(pairs)) = Some(files) {
+        for (rel, content) in pairs {
+            let p = root.join(rel);
+            fs::create_dir_all(p.parent().unwrap_or(root))
+                .unwrap_or_else(|e| panic!("mkdir {}: {}", p.display(), e));
+            let text = match content {
+                Json::Str(t) => t.clone(),
+                other => pyjson::dumps(other),
+            };
+            fs::write(&p, text).unwrap_or_else(|e| panic!("write {}: {}", p.display(), e));
+        }
+    }
+}
+
+fn sandbox_dir(idx: usize) -> PathBuf {
+    let d = std::env::temp_dir().join(format!(
+        "spine-gap-battery-{}-{}",
+        std::process::id(),
+        idx
+    ));
+    let _ = fs::remove_dir_all(&d);
+    fs::create_dir_all(&d).expect("sandbox mkdir");
+    d
+}
+
+/// Dispatch one harness case (args = oracle positional args, in order).
+fn replay_harness(case: &Json) -> Result<Json, String> {
+    let fname = store::as_str(expect_field(case, "fn")).to_string();
+    let args = expect_field(case, "args");
+    let items = match args {
+        Json::Arr(items) => items.clone(),
+        _ => vec![],
+    };
+    match fname.as_str() {
+        "_py_sorted" => Ok(gap::_py_sorted(&items[0])),
+        "_py_declared_surfaces" => gap::_py_declared_surfaces(&items[0]),
+        "_py_entity" => Ok(gap::_py_entity(&items[0], store::as_str(&items[1]))),
+        "_rust_core" => Ok(match gap::_rust_core("python", true, true) {
+            Some(name) => Json::Str(name.to_string()),
+            None => Json::Null,
+        }),
+        other => Err(format!("unknown fn in harness truth: {}", other)),
+    }
+}
+
+/// The loader-wrapper prefix the Python loader adds to delegated-path
+/// kernel errors (documented battery law — see module docs).
+fn strip_loader_prefix(msg: &str) -> &str {
+    msg.strip_prefix("sp_project core error: ").unwrap_or(msg)
+}
+
+#[cfg(test)]
+mod truth_battery {
+    use super::*;
+
+    #[test]
+    fn harness_truth_replays_verbatim() {
+        let doc = parse(HARNESS_TRUTH);
+        let cases = doc.get("cases").expect("envelope.cases");
+        let mut n = 0usize;
+        if let Some(Json::Arr(items)) = Some(cases) {
+            for case in items {
+                let fname = store::as_str(expect_field(case, "fn"));
+                let got = replay_harness(case)
+                    .unwrap_or_else(|e| panic!("{} dispatch failed: {}", fname, e));
+                let expect = expect_field(case, "expected");
+                assert_eq!(
+                    canon(&got),
+                    canon(expect),
+                    "case {} parity (canon)",
+                    fname
+                );
+                n += 1;
+            }
+        }
+        assert_eq!(n, 21, "harness case count");
+    }
+
+    #[test]
+    fn clocked_truth_replays_verbatim() {
+        let doc = parse(CLOCKED_TRUTH);
+        let frozen_now = store::as_str(doc.get("frozen_now").expect("frozen_now")).to_string();
+        let stamp = format!("{}Z", frozen_now);
+        let cases = doc.get("cases").expect("cases");
+        let mut n = 0usize;
+        if let Some(Json::Arr(items)) = Some(cases) {
+            for (idx, case) in items.iter().enumerate() {
+                let fname = store::as_str(expect_field(case, "fn")).to_string();
+                let name = store::as_str(expect_field(case, "name")).to_string();
+                let files = case.get("files").cloned().unwrap_or(Json::Obj(vec![]));
+                let root = sandbox_dir(idx);
+                materialize(&files, &root);
+                let args = match expect_field(case, "args") {
+                    Json::Arr(items) => items.clone(),
+                    _ => vec![],
+                };
+                let kwargs = expect_field(case, "kwargs").clone();
+                let _ = kwargs;
+
+                let result: Result<Json, String> = match fname.as_str() {
+                    "_module_imports" => {
+                        let module = store::as_str(&args[0]).to_string();
+                        let src = fs::read_to_string(
+                            root.join("src/rune").join(format!("{}.py", module)),
+                        )
+                        .ok();
+                        gap::_module_imports(&module, src.as_deref())
+                            .map(|names| Json::Arr(names.into_iter().map(Json::Str).collect()))
+                    }
+                    "_gather_meta_inputs" => gap::_gather_meta_inputs(&root, None),
+                    "_deployed_source_files" => Ok(Json::Arr(
+                        gap::_deployed_source_files(&root.join("deployed"))
+                            .into_iter()
+                            .map(Json::Str)
+                            .collect(),
+                    )),
+                    "_rust_core" => {
+                        // both frozen branches: loader importable, .so absent
+                        Ok(match gap::_rust_core("auto", true, false) {
+                            Some(x) => Json::Str(x.to_string()),
+                            None => Json::Null,
+                        })
+                    }
+                    "_py_project_meta_surfaces" => gap::_py_project_meta_surfaces(
+                        &args[0],
+                        &args[1],
+                        &stamp,
+                    )
+                    .map(|out| {
+                        // the frozen expected carries the oracle's return
+                        // (the count int); post_doc carries the mutated doc
+                        Json::Obj(vec![
+                            ("result".into(), out.get("count").cloned().unwrap_or(Json::Null)),
+                            ("doc".into(), out.get("doc").cloned().unwrap_or(Json::Null)),
+                        ])
+                    }),
+                    "_py_ingest_umbrellas_data" => {
+                        gap::_py_ingest_umbrellas_data(&args[0], &args[1], &stamp).map(|out| {
+                            Json::Obj(vec![
+                                ("result".into(), out.get("count").cloned().unwrap_or(Json::Null)),
+                                ("doc".into(), out.get("doc").cloned().unwrap_or(Json::Null)),
+                            ])
+                        })
+                    }
+                    "_py_build_from_inputs" => {
+                        // kernel_absent_raises froze under the python pin
+                        // (delegation-only stub); the other three cases ran
+                        // kernel-live — including generated_view_refused,
+                        // whose kernel error the loader prefix-strip below
+                        // unprefixes.
+                        let core_up = name != "kernel_absent_raises";
+                        gap::_py_build_from_inputs(&args[0], core_up)
+                    }
+                    "_gather_drift_inputs" => {
+                        // the Python-side gathers the generator patched in
+                        // are case fixtures (the registry/graph seam):
+                        let registry = vec!["canary_alpha".to_string(), "canary_beta".to_string()];
+                        let graph = parse(
+                            r#"{"alpha": ["json", "rune"], "beta": ["sys"]}"#,
+                        );
+                        // the oracle's _test_backed_modules strips the
+                        // test_ prefix (p.stem[5:])
+                        let mut tested: Vec<String> = file_stems_under(
+                            &root.join("tests"),
+                            "test_",
+                            ".py",
+                        )
+                        .into_iter()
+                        .map(|stem| stem["test_".len()..].to_string())
+                        .collect();
+                        tested.sort();
+                        let src_stems = readdir_stems(&root.join("src/rune"), ".py");
+                        gap::_gather_drift_inputs(
+                            &root,
+                            &registry,
+                            &graph,
+                            &tested,
+                            &src_stems,
+                            &Json::Null,
+                        )
+                    }
+                    "_py_drift_from_inputs" => gap::_py_drift_from_inputs(&args[0], &args[1], false),
+                    "_py_export_umbrellas" => {
+                        gap::_py_export_umbrellas(&args[0], &stamp, false)
+                    }
+                    "_py_query" => gap::_py_query(&args[0], store::as_str(&args[1]), false),
+                    "_fresh_doc" => gap::_fresh_doc(&root, None),
+                    other => panic!("unknown fn in clocked truth: {}", other),
+                };
+
+                match (result, case.get("error")) {
+                    (Ok(mut got), None) => {
+                        // doc-mutating cases: compare {result, doc} against
+                        // expected + post_doc
+                        if let Some(post) = case.get("post_doc") {
+                            let expect = Json::Obj(vec![
+                                ("result".into(), expect_field(case, "expected").clone()),
+                                ("doc".into(), post.clone()),
+                            ]);
+                            assert_eq!(
+                                canon(&got),
+                                canon(&expect),
+                                "case {}/{} parity (canon)",
+                                fname,
+                                name
+                            );
+                        } else {
+                            if let Json::Obj(ref mut pairs) = got {
+                                if pairs.len() == 2
+                                    && pairs[0].0 == "result"
+                                    && pairs[1].0 == "doc"
+                                    && matches!(case.get("expected"), Some(Json::Num(_)))
+                                {
+                                    // kernel doc-wrappers on non-mutating
+                                    // cases are not expected here
+                                }
+                            }
+                            assert_eq!(
+                                canon(&got),
+                                canon(expect_field(case, "expected")),
+                                "case {}/{} parity (canon)",
+                                fname,
+                                name
+                            );
+                        }
+                    }
+                    (Err(err), Some(err_doc)) => {
+                        let msg = store::as_str(
+                            err_doc.get("message").expect("error.message"),
+                        );
+                        assert_eq!(
+                            err,
+                            strip_loader_prefix(msg),
+                            "case {}/{} error text",
+                            fname,
+                            name
+                        );
+                    }
+                    (other, _) => panic!(
+                        "case {}/{}: unexpected outcome: {:?}",
+                        fname, name, other
+                    ),
+                }
+                let _ = fs::remove_dir_all(&root);
+                n += 1;
+            }
+        }
+        assert_eq!(n, 35, "clocked case count");
+    }
+
+    fn file_stems_under(dir: &Path, prefix: &str, suffix: &str) -> Vec<String> {
+        let mut out = vec![];
+        if let Ok(rd) = fs::read_dir(dir) {
+            for e in rd.flatten() {
+                let name = e.file_name().to_string_lossy().into_owned();
+                if name.len() >= prefix.len() + suffix.len()
+                    && name.starts_with(prefix)
+                    && name.ends_with(suffix)
+                {
+                    out.push(name[..name.len() - suffix.len()].to_string());
+                }
+            }
+        }
+        out
+    }
+
+    fn readdir_stems(dir: &Path, suffix: &str) -> Vec<String> {
+        let mut out = vec![];
+        if let Ok(rd) = fs::read_dir(dir) {
+            for e in rd.flatten() {
+                let name = e.file_name().to_string_lossy().into_owned();
+                if name.ends_with(suffix) {
+                    out.push(name[..name.len() - suffix.len()].to_string());
+                }
+            }
+        }
+        out
+    }
+
+    /// The auto+present branch the frozen truth cannot carry (the loader
+    /// module is unrepresentable over FFI): the verdict is Some.
+    #[test]
+    fn rust_core_auto_present_branch() {
+        assert_eq!(gap::_rust_core("auto", true, true), Some("spine_loader"));
+        assert_eq!(gap::_rust_core("auto", false, true), None);
+        assert_eq!(gap::_rust_core("python", true, true), None);
+    }
+
+    /// The _fresh_doc build-and-save seam the pinned oracle cannot reach
+    /// (build() gathers in Python): an injected built doc is returned AND
+    /// persisted with spine.save's canonical text.
+    #[test]
+    fn fresh_doc_absent_store_builds_and_saves() {
+        let root = sandbox_dir(9001);
+        let built = parse(
+            r#"{"schema_version": 1, "entities": [{"id": "module:alpha", "kind": "module", "maturity": "DECLARED", "tags": [], "identities": {}, "evidence_refs": [], "observed_at": "2026-10-13T08:30:00Z"}], "relationships": [], "observations": [], "generated_at": "2026-10-13T08:30:00Z"}"#,
+        );
+        let got = gap::_fresh_doc(&root, Some(&built)).expect("fresh_doc builds");
+        assert_eq!(canon(&got), canon(&built));
+        let persisted = fs::read_to_string(root.join("store/spine/current.json")).expect("saved");
+        let expect_text = rune_spine::facade::save_text(&Json::Obj(vec![("doc".into(), built)]));
+        if let Ok(Json::Str(text)) = expect_text {
+            assert_eq!(persisted, text, "canonical save text");
+        } else {
+            panic!("save_text failed");
+        }
+        // second call loads the persisted store (load path wins)
+        let again = gap::_fresh_doc(&root, None).expect("reload");
+        assert_eq!(canon(&again), canon(&got));
+        let _ = fs::remove_dir_all(&root);
+    }
+}

src/rust/spine/src/gap.rs

@@ -0,0 +1,704 @@
+//! gap.rs — dichotomy-wave gap-closing kernels (issue #2479 tranche,
+//! contract BUILDER-2026-10-13T08:11:47Z-9b1f32-factory-gap-sp).
+//!
+//! Name-preserving ports of the audit-missing spine.py functions. Two
+//! classes live here:
+//!
+//! REAL kernels (new logic, IO behind injectable seams — the zero-dep
+//! split law: the Python facade gathers what only Python can gather
+//! (imports of rune.canary_native, the import graph) and passes it in;
+//! filesystem probes take an explicit root):
+//!   _module_imports(module, src_text) — the AST-import scan over an
+//!     injected source text (the file read stays Python; None = the
+//!     oracle's OSError branch; a malformed import statement is the
+//!     oracle's SyntaxError branch → empty set).
+//!   _gather_meta_inputs(root, workflows_dir) — the meta-surface gather
+//!     (manifest parse + dir/file probes + glob scans, all explicit).
+//!   _deployed_source_files(root) — the deployed-tree source scan.
+//!     readdir order, NOT sorted — the oracle's glob order is the
+//!     contract (both are readdir sequences on the same directory).
+//!   _gather_drift_inputs(root, registry, graph, tested, src_stems,
+//!     deployed_obs) — assembly + the pyproject version extraction; the
+//!     Python-side gathers (registry/graph/tested/src_stems/deployed_obs)
+//!     are injected seams.
+//!   _fresh_doc(root, built) — load-or-build seam: the store read is
+//!     direct; the absent-store branch takes the caller-gathered build
+//!     result (the oracle's build() gathers in Python) or raises the
+//!     retired-stub error exactly like the pinned oracle.
+//!
+//! MACHINERY twins (the audit excludes _py_*/_core fns as fallback
+//! machinery; these wrap the PROVEN rune_spine kernels — no second
+//! implementation, strangler law):
+//!   _py_sorted → store::sorted_deep (sf_sorted twin)
+//!   _py_declared_surfaces → facade::declared_surfaces
+//!   _py_project_meta_surfaces → facade::project_meta_surfaces_fn
+//!   _py_ingest_umbrellas_data → facade::ingest_umbrellas_data_fn
+//!   _py_entity → store::entity (sf_entity twin)
+//!   _py_build_from_inputs / _py_drift_from_inputs / _py_export_umbrellas
+//!     / _py_query → project/drift/query kernels with the retired-stub
+//!     core-absent error (battery-pinned message)
+//!   _rust_core(pin, loader_importable, so_present) — the availability
+//!     verdict as data (the module itself is unrepresentable over FFI).
+//!
+//! Parity: frozen_spine_gap_truth.json (shared freeze_truth harness,
+//! byte-verified) + frozen_spine_gap_clocked.json (sandbox/clock/error
+//! battery) — both replayed VERBATIM in branch_battery_gap.rs.
+
+use std::fs;
+use std::path::{Path, PathBuf};
+
+use rune_spine::facade;
+use rune_spine::pyjson::{self, Json};
+use rune_spine::store;
+
+pub const KERNEL_NOT_BUILT: &str = "rune_spine kernel not built — run the rune_spine build; the Python oracle was retired";
+
+fn s(text: &str) -> Json {
+    Json::Str(text.to_string())
+}
+
+fn str_arr(items: &[String]) -> Json {
+    Json::Arr(items.iter().map(|x| s(x)).collect())
+}
+
+// ── _module_imports — the AST-import scan over injected source text ───────
+
+/// One physical line reduced to code bytes: comments stripped, string
+/// literal CONTENTS blanked (quotes kept as markers so token boundaries
+/// survive), unterminated triple-quotes spanning lines tracked.
+#[derive(Default)]
+struct LineStripper {
+    in_triple: Option<char>,
+}
+
+impl LineStripper {
+    fn strip(&mut self, line: &str) -> String {
+        let chars: Vec<char> = line.chars().collect();
+        let mut out = String::with_capacity(line.len());
+        let mut i = 0usize;
+        while i < chars.len() {
+            let c = chars[i];
+            if let Some(q) = self.in_triple {
+                // inside a triple-quoted string: look for the closer
+                if c == q && i + 2 < chars.len() + 1 {
+                    let triple: String = chars[i..(i + 3).min(chars.len())].iter().collect();
+                    if triple.len() == 3 && triple.chars().all(|x| x == q) {
+                        self.in_triple = None;
+                        out.push_str(&triple);
+                        i += 3;
+                        continue;
+                    }
+                }
+                out.push(' ');
+                i += 1;
+                continue;
+            }
+            match c {
+                '#' => break, // comment to end of line
+                '"' | '\'' => {
+                    // triple quote opener?
+                    let triple: String = chars[i..(i + 3).min(chars.len())].iter().collect();
+                    if triple.len() == 3 && triple.chars().all(|x| x == c) {
+                        self.in_triple = Some(c);
+                        out.push_str(&triple);
+                        i += 3;
+                        continue;
+                    }
+                    // single-quoted string: blank contents until closer
+                    out.push(c);
+                    i += 1;
+                    while i < chars.len() {
+                        if chars[i] == '\\' && i + 1 < chars.len() {
+                            i += 2;
+                            continue;
+                        }
+                        if chars[i] == c {
+                            out.push(c);
+                            i += 1;
+                            break;
+                        }
+                        out.push(' ');
+                        i += 1;
+                    }
+                }
+                _ => {
+                    out.push(c);
+                    i += 1;
+                }
+            }
+        }
+        out
+    }
+}
+
+fn is_ident_start(c: char) -> bool {
+    c.is_alphabetic() || c == '_'
+}
+
+fn is_ident_char(c: char) -> bool {
+    c.is_alphanumeric() || c == '_'
+}
+
+/// Split source text into logical statements (comment/string-stripped,
+/// bracket- and backslash-continuations joined, `;`-separated).
+fn logical_statements(src: &str) -> Vec<String> {
+    let mut stripper = LineStripper::default();
+    let mut logicals: Vec<String> = vec![];
+    let mut pending = String::new();
+    let mut depth: i32 = 0;
+    for raw in src.lines() {
+        let line = stripper.strip(raw);
+        let mut piece = line.as_str();
+        let mut joins = false;
+        if let Some(stripped) = piece.strip_suffix('\\') {
+            piece = stripped;
+            joins = true;
+        }
+        for ch in piece.chars() {
+            match ch {
+                '(' | '[' | '{' => depth += 1,
+                ')' | ']' | '}' => depth -= 1,
+                _ => {}
+            }
+        }
+        if !pending.is_empty() {
+            pending.push(' ');
+        }
+        pending.push_str(piece.trim_end());
+        if joins || depth > 0 {
+            continue;
+        }
+        logicals.push(std::mem::take(&mut pending));
+    }
+    if !pending.is_empty() {
+        logicals.push(pending);
+    }
+    // split each logical line on ';' at depth 0
+    let mut out: Vec<String> = vec![];
+    for line in logicals {
+        let mut cur = String::new();
+        let mut d: i32 = 0;
+        for ch in line.chars() {
+            match ch {
+                '(' | '[' | '{' => {
+                    d += 1;
+                    cur.push(ch);
+                }
+                ')' | ']' | '}' => {
+                    d -= 1;
+                    cur.push(ch);
+                }
+                ';' if d == 0 => out.push(std::mem::take(&mut cur)),
+                _ => cur.push(ch),
+            }
+        }
+        if !cur.trim().is_empty() {
+            out.push(cur);
+        }
+    }
+    out
+}
+
+/// Parse one dotted identifier starting at `chars[i]`; returns (text, next index).
+fn parse_dotted(chars: &[char], mut i: usize) -> Result<(String, usize), String> {
+    if i >= chars.len() || !is_ident_start(chars[i]) {
+        return Err(format!("expected identifier at {}", i));
+    }
+    let start = i;
+    while i < chars.len() && (is_ident_char(chars[i]) || chars[i] == '.') {
+        i += 1;
+    }
+    // trailing dot is malformed
+    if chars[i - 1] == '.' {
+        return Err("trailing dot in dotted name".to_string());
+    }
+    Ok((chars[start..i].iter().collect(), i))
+}
+
+fn skip_ws(chars: &[char], mut i: usize) -> usize {
+    while i < chars.len() && chars[i].is_whitespace() {
+        i += 1;
+    }
+    i
+}
+
+fn expect_keyword(chars: &[char], i: usize, kw: &str) -> Result<usize, String> {
+    let end = skip_ws(chars, i);
+    let kwc: Vec<char> = kw.chars().collect();
+    if chars.len() >= end + kwc.len() && chars[end..end + kwc.len()] == kwc[..] {
+        let after = end + kwc.len();
+        if after >= chars.len() || !is_ident_char(chars[after]) {
+            return Ok(after);
+        }
+    }
+    Err(format!("expected `{}`", kw))
+}
+
+/// Parse the names after `import` (comma-separated dotted names, optional
+/// `as x` aliases). Any malformed token → Err (the SyntaxError branch).
+fn parse_import_names(chars: &[char], mut i: usize) -> Result<Vec<String>, String> {
+    let mut names = vec![];
+    loop {
+        i = skip_ws(chars, i);
+        let (dotted, next) = parse_dotted(chars, i)?;
+        names.push(dotted);
+        i = skip_ws(chars, next);
+        if i < chars.len() && chars[i] == 'a' {
+            let after = expect_keyword(chars, i, "as")?;
+            i = skip_ws(chars, after);
+            let (_alias, next) = parse_dotted(chars, i)?;
+            i = next;
+        }
+        i = skip_ws(chars, i);
+        if i < chars.len() && chars[i] == ',' {
+            i += 1;
+            continue;
+        }
+        if i < chars.len() {
+            return Err("trailing tokens after import names".to_string());
+        }
+        break;
+    }
+    Ok(names)
+}
+
+/// The spine._module_imports core: top-level imported module names from
+/// injected source text. `src_text=None` mirrors the oracle's OSError
+/// branch; a malformed import statement mirrors its SyntaxError branch
+/// (empty result). The set is returned as a sorted list (FFI law: sets
+/// cross the wire sorted).
+pub fn _module_imports(module: &str, src_text: Option<&str>) -> Result<Vec<String>, String> {
+    let _ = module; // the oracle keys the file read by it; text is injected
+    let text = match src_text {
+        None => return Ok(vec![]),
+        Some(t) => t,
+    };
+    let mut names: Vec<String> = vec![];
+    for stmt in logical_statements(text) {
+        let trimmed = stmt.trim_start();
+        let chars: Vec<char> = trimmed.chars().collect();
+        if chars.is_empty() {
+            continue;
+        }
+        // A malformed import statement mirrors the oracle's SyntaxError
+        // branch: ast.parse refuses the whole file → empty set.
+        if let Ok(after) = expect_keyword(&chars, 0, "import") {
+            let dotted_names = match parse_import_names(&chars, after) {
+                Ok(names) => names,
+                Err(_) => return Ok(vec![]),
+            };
+            for dotted in dotted_names {
+                let top = dotted.split('.').next().unwrap_or("").to_string();
+                if !names.contains(&top) {
+                    names.push(top);
+                }
+            }
+        } else if let Ok(after_from) = expect_keyword(&chars, 0, "from") {
+            // module part: leading dots (relative) then a dotted name
+            let mut i = skip_ws(&chars, after_from);
+            while i < chars.len() && chars[i] == '.' {
+                i += 1;
+            }
+            let parsed: Result<(), String> = if i < chars.len() && is_ident_start(chars[i]) {
+                parse_dotted(&chars, i).and_then(|(module_path, next)| {
+                    expect_keyword(&chars, next, "import").map(|_| module_path)
+                })
+                .map(|module_path| {
+                    // `from x.y import ...` → node.module = "x.y" → "x";
+                    // `from .x import ...` → node.module = "x" → "x"
+                    let top = module_path.split('.').next().unwrap_or("").to_string();
+                    if !names.contains(&top) {
+                        names.push(top);
+                    }
+                })
+            } else {
+                // `from . import x` → node.module = None → skipped by the
+                // oracle; still validate the import keyword + names
+                expect_keyword(&chars, i, "import")
+                    .and_then(|after_import| parse_import_names(&chars, after_import).map(|_| ()))
+            };
+            if parsed.is_err() {
+                return Ok(vec![]); // the oracle's SyntaxError branch
+            }
+        }
+        // non-import statements: not validated (a full Python grammar is
+        // out of scope; the oracle's only observable here is the import set)
+    }
+    names.sort();
+    Ok(names)
+}
+
+// ── _gather_meta_inputs — the meta-surface gather under an explicit root ──
+
+fn read_manifest(root: &Path) -> Option<Json> {
+    let text = fs::read_to_string(root.join("config").join("umbrellas.json")).ok()?;
+    pyjson::parse(&text).ok()
+}
+
+fn dir_entry_names(dir: &Path) -> Vec<String> {
+    let mut names: Vec<String> = match fs::read_dir(dir) {
+        Ok(rd) => rd.filter_map(|e| e.ok()).map(|e| e.file_name().to_string_lossy().into_owned()).collect(),
+        Err(_) => vec![],
+    };
+    names.sort();
+    names
+}
+
+fn glob_match(name: &str, prefix: &str, suffix: &str) -> bool {
+    name.len() >= prefix.len() + suffix.len()
+        && name.starts_with(prefix)
+        && name.ends_with(suffix)
+}
+
+/// The spine._gather_meta_inputs core. `workflows_dir` defaults to
+/// `<root>/.gitea/workflows` (the facade passes the module global so the
+/// monkeypatch seam holds). Canaries and workflows are SORTED (the
+/// oracle's explicit sorted() calls); existence probes are direct.
+pub fn _gather_meta_inputs(root: &Path, workflows_dir: Option<&Path>) -> Result<Json, String> {
+    let manifest = read_manifest(root);
+    // declared surfaces: tolerant parse (the oracle's except → empty)
+    let declared = facade::declared_surfaces(&Json::Obj(vec![(
+        "config_manifest".into(),
+        manifest.clone().unwrap_or(Json::Null),
+    )]))?;
+    let declared_crates: Vec<(String, String)> = match declared.get("rust_crates") {
+        Some(Json::Arr(items)) => items
+            .iter()
+            .filter_map(|it| match it {
+                Json::Arr(pair) if pair.len() == 2 => Some((
+                    store::as_str(&pair[0]).to_string(),
+                    store::as_str(&pair[1]).to_string(),
+                )),
+                _ => None,
+            })
+            .collect(),
+        _ => vec![],
+    };
+    let declared_go: Vec<(String, String)> = match declared.get("go_surfaces") {
+        Some(Json::Arr(items)) => items
+            .iter()
+            .filter_map(|it| match it {
+                Json::Arr(pair) if pair.len() == 2 => Some((
+                    store::as_str(&pair[0]).to_string(),
+                    store::as_str(&pair[1]).to_string(),
+                )),
+                _ => None,
+            })
+            .collect(),
+        _ => vec![],
+    };
+    let declared_cfg: Vec<(String, String)> = match declared.get("configs") {
+        Some(Json::Arr(items)) => items
+            .iter()
+            .filter_map(|it| match it {
+                Json::Arr(pair) if pair.len() == 2 => Some((
+                    store::as_str(&pair[0]).to_string(),
+                    store::as_str(&pair[1]).to_string(),
+                )),
+                _ => None,
+            })
+            .collect(),
+        _ => vec![],
+    };
+
+    let mut crate_dirs: Vec<String> = vec![];
+    for (c, _) in &declared_crates {
+        if root.join("src").join("rust").join(c).is_dir() {
+            crate_dirs.push(c.clone());
+        }
+    }
+    let mut crate_canaries: Vec<(String, Json)> = vec![];
+    for c in &crate_dirs {
+        let dir = root.join("src").join("rust").join(c);
+        let mut stems: Vec<String> = dir_entry_names(&dir)
+            .into_iter()
+            .filter(|n| glob_match(n, "canary_", ".py"))
+            .map(|n| n.trim_end_matches(".py").to_string())
+            .collect();
+        stems.sort();
+        crate_canaries.push((c.clone(), str_arr(&stems)));
+    }
+    let mut go_dirs: Vec<String> = vec![];
+    for (g, _) in &declared_go {
+        if root.join("src").join("go").join(g).is_dir() {
+            go_dirs.push(g.clone());
+        }
+    }
+    let mut config_files: Vec<String> = vec![];
+    for (cfg, _) in &declared_cfg {
+        if root.join(cfg).is_file() {
+            config_files.push(cfg.clone());
+        }
+    }
+    let wf_dir = workflows_dir
+        .map(PathBuf::from)
+        .unwrap_or_else(|| root.join(".gitea").join("workflows"));
+    let mut workflows: Vec<String> = vec![];
+    if wf_dir.is_dir() {
+        workflows = dir_entry_names(&wf_dir)
+            .into_iter()
+            .filter(|n| n.ends_with(".yml"))
+            .collect();
+        workflows.sort();
+    }
+    Ok(Json::Obj(vec![
+        ("config_manifest".into(), manifest.unwrap_or(Json::Null)),
+        ("crate_dirs".into(), str_arr(&crate_dirs)),
+        ("crate_canaries".into(), Json::Obj(crate_canaries)),
+        ("go_dirs".into(), str_arr(&go_dirs)),
+        ("config_files".into(), str_arr(&config_files)),
+        ("workflows".into(), str_arr(&workflows)),
+    ]))
+}
+
+// ── _deployed_source_files — readdir-order source scan ────────────────────
+
+/// The spine._deployed_source_files core: relative source paths under the
+/// deployed link that exist as files. Order is the oracle's glob order —
+/// readdir sequence per directory, NOT sorted (both cores walk the same
+/// directory, so the sequences agree).
+pub fn _deployed_source_files(root: &Path) -> Vec<String> {
+    let mut out: Vec<String> = vec![];
+    let bases = [
+        (root.join("src").join("rune"), "*.py" , false),
+        (root.join("src").join("rust"), "*/canary_*.py", true),
+    ];
+    for (base, pattern, nested) in bases {
+        if !base.is_dir() {
+            continue;
+        }
+        if !nested {
+            for name in readdir_names(&base) {
+                if name.ends_with(".py") {
+                    let p = base.join(&name);
+                    if p.is_file() {
+                        out.push(rel_posix(root, &p));
+                    }
+                }
+            }
+        } else {
+            for sub in readdir_names(&base) {
+                let sub_dir = base.join(&sub);
+                if !sub_dir.is_dir() {
+                    continue;
+                }
+                for name in readdir_names(&sub_dir) {
+                    if glob_match(&name, "canary_", ".py") {
+                        let p = sub_dir.join(&name);
+                        if p.is_file() {
+                            out.push(rel_posix(root, &p));
+                        }
+                    }
+                }
+            }
+        }
+        let _ = pattern;
+    }
+    out
+}
+
+fn readdir_names(dir: &Path) -> Vec<String> {
+    match fs::read_dir(dir) {
+        Ok(rd) => rd
+            .filter_map(|e| e.ok())
+            .map(|e| e.file_name().to_string_lossy().into_owned())
+            .collect(),
+        Err(_) => vec![],
+    }
+}
+
+fn rel_posix(root: &Path, p: &Path) -> String {
+    p.strip_prefix(root).unwrap_or(p).to_string_lossy().replace('\\', "/")
+}
+
+// ── _gather_drift_inputs — assembly + pyproject version extraction ────────
+
+/// The spine._gather_drift_inputs core. registry/graph/tested/src_stems/
+/// deployed_obs are the Python-side gathers (injected seams — the registry
+/// imports rune.canary_native, the graph needs the AST scan); the version
+/// comes from `<root>/pyproject.toml` via the oracle's anchored regex.
+/// graph values and tested are sorted here (the oracle's sorted() calls).
+pub fn _gather_drift_inputs(
+    root: &Path,
+    registry: &[String],
+    graph: &Json,
+    tested: &[String],
+    src_stems: &[String],
+    deployed_obs: &Json,
+) -> Result<Json, String> {
+    let mut sorted_graph: Vec<(String, Json)> = vec![];
+    if let Some(Json::Obj(pairs)) = Some(graph) {
+        for (m, deps) in pairs {
+            let mut list: Vec<String> = store::as_arr(deps).iter().map(|x| store::as_str(x).to_string()).collect();
+            list.sort();
+            sorted_graph.push((m.clone(), str_arr(&list)));
+        }
+    }
+    let mut sorted_tested: Vec<String> = tested.to_vec();
+    sorted_tested.sort();
+    let src_version = pyproject_version(root);
+    Ok(Json::Obj(vec![
+        ("registry".into(), str_arr(registry)),
+        ("graph".into(), Json::Obj(sorted_graph)),
+        ("tested".into(), str_arr(&sorted_tested)),
+        ("src_stems".into(), str_arr(src_stems)),
+        ("deployed_obs".into(), deployed_obs.clone()),
+        ("src_version".into(), s(&src_version)),
+    ]))
+}
+
+/// The oracle's `re.search(r'^version\s*=\s*"([^"]+)"', text, re.M)`:
+/// line-anchored `version` key with a double-quoted value.
+fn pyproject_version(root: &Path) -> String {
+    let text = match fs::read_to_string(root.join("pyproject.toml")) {
+        Ok(t) => t,
+        Err(_) => return String::new(),
+    };
+    for line in text.split('\n') {
+        if !line.starts_with("version") {
+            continue;
+        }
+        let rest = &line["version".len()..];
+        let mut chars = rest.chars().peekable();
+        while matches!(chars.peek(), Some(c) if c.is_whitespace()) {
+            chars.next();
+        }
+        if chars.next() != Some('=') {
+            continue;
+        }
+        while matches!(chars.peek(), Some(c) if c.is_whitespace()) {
+            chars.next();
+        }
+        if chars.next() != Some('"') {
+            continue;
+        }
+        let mut value = String::new();
+        let mut closed = false;
+        for c in chars {
+            if c == '"' {
+                closed = true;
+                break;
+            }
+            value.push(c);
+        }
+        if closed {
+            return value;
+        }
+    }
+    String::new()
+}
+
+// ── _fresh_doc — load-or-build seam ───────────────────────────────────────
+
+/// The spine._fresh_doc core over an explicit root: the store read is
+/// direct (missing/unparseable → absent); the absent-store branch takes
+/// the caller-gathered build result (`built` — the oracle's build()
+/// gathers in Python) and persists it with spine.save's canonical text;
+/// with no injected build the retired-stub error is raised exactly like
+/// the pinned oracle.
+pub fn _fresh_doc(root: &Path, built: Option<&Json>) -> Result<Json, String> {
+    let store_path = root.join("store").join("spine").join("current.json");
+    if store_path.is_file() {
+        if let Ok(text) = fs::read_to_string(&store_path) {
+            if let Ok(doc) = pyjson::parse(&text) {
+                return Ok(doc);
+            }
+        }
+    }
+    match built {
+        Some(doc) => {
+            let text = facade::save_text(&Json::Obj(vec![("doc".into(), doc.clone())]))?;
+            if let Json::Str(body) = text {
+                if let Some(parent) = store_path.parent() {
+                    fs::create_dir_all(parent).map_err(|e| e.to_string())?;
+                }
+                fs::write(&store_path, body).map_err(|e| e.to_string())?;
+            }
+            Ok(doc.clone())
+        }
+        None => Err(KERNEL_NOT_BUILT.to_string()),
+    }
+}
+
+// ── _rust_core — the availability verdict as data ─────────────────────────
+
+/// The spine._rust_core verdict: pin=="python" → None; loader not
+/// importable → None; loader available → Some (represented as the loader
+/// name — the module itself is unrepresentable over FFI).
+pub fn _rust_core(pin: &str, loader_importable: bool, so_present: bool) -> Option<&'static str> {
+    if pin == "python" {
+        return None;
+    }
+    if !loader_importable {
+        return None;
+    }
+    if so_present {
+        Some("spine_loader")
+    } else {
+        None
+    }
+}
+
+// ── machinery twins — thin routes to the PROVEN rune_spine kernels ────────
+
+pub fn _py_sorted(v: &Json) -> Json {
+    store::sorted_deep(v)
+}
+
+pub fn _py_declared_surfaces(manifest: &Json) -> Result<Json, String> {
+    facade::declared_surfaces(&Json::Obj(vec![(
+        "config_manifest".into(),
+        manifest.clone(),
+    )]))
+}
+
+pub fn _py_project_meta_surfaces(doc: &Json, inputs: &Json, now: &str) -> Result<Json, String> {
+    facade::project_meta_surfaces_fn(&Json::Obj(vec![
+        ("doc".into(), doc.clone()),
+        ("inputs".into(), inputs.clone()),
+        ("now".into(), s(now)),
+    ]))
+}
+
+pub fn _py_ingest_umbrellas_data(doc: &Json, data: &Json, now: &str) -> Result<Json, String> {
+    facade::ingest_umbrellas_data_fn(&Json::Obj(vec![
+        ("doc".into(), doc.clone()),
+        ("data".into(), data.clone()),
+        ("now".into(), s(now)),
+    ]))
+}
+
+pub fn _py_entity(doc: &Json, id: &str) -> Json {
+    match store::entity(doc, id) {
+        Some(e) => e.clone(),
+        None => Json::Null,
+    }
+}
+
+pub fn _py_build_from_inputs(inputs: &Json, core_available: bool) -> Result<Json, String> {
+    if !core_available {
+        return Err(KERNEL_NOT_BUILT.to_string());
+    }
+    rune_spine::project::project(inputs)
+}
+
+pub fn _py_drift_from_inputs(doc: &Json, inputs: &Json, core_available: bool) -> Result<Json, String> {
+    if !core_available {
+        return Err(KERNEL_NOT_BUILT.to_string());
+    }
+    Ok(rune_spine::drift::drift(doc, inputs))
+}
+
+pub fn _py_export_umbrellas(doc: &Json, now: &str, core_available: bool) -> Result<Json, String> {
+    if !core_available {
+        return Err(KERNEL_NOT_BUILT.to_string());
+    }
+    rune_spine::query::export_umbrellas(doc, now)
+}
+
+pub fn _py_query(doc: &Json, entity_id: &str, core_available: bool) -> Result<Json, String> {
+    if !core_available {
+        return Err(KERNEL_NOT_BUILT.to_string());
+    }
+    Ok(rune_spine::query::query(doc, entity_id))
+}

src/rust/spine/src/lib.rs

@@ -1,7 +1,317 @@
 //! spine — Rune's self-model spine (W6 deep port).
 //! PARITY port of the core data structures from src/rune/spine.py.
+//!
+//! Dichotomy-wave gap kernels (issue #2479 tranche) live in `gap.rs` and
+//! are exposed as doc-envelope FFI below: every symbol takes ONE JSON
+//! argument string and returns {"ok": true, "value": ...} or
+//! {"ok": false, "err": "..."}. NOTE: no free symbol is exported here —
+//! capability_scoring (same process) owns cs_free_string; callers free
+//! every returned pointer through it (loader ABI law, recall_uplift
+//! precedent). FFI segfault law: the loader pins argtypes + restype
+//! (c_void_p returns — never the default c_int).
+
 use serde::{Deserialize, Serialize};
 
+#[cfg(test)]
+mod branch_battery_gap;
+
+pub mod gap;
+
+use std::ffi::{CStr, CString};
+use std::os::raw::c_char;
+
+use rune_spine::pyjson::{self, Json};
+
+fn read_arg(ptr: *const c_char) -> Result<Json, String> {
+    if ptr.is_null() {
+        return Err("null argument pointer".to_string());
+    }
+    let text = unsafe { CStr::from_ptr(ptr) }
+        .to_str()
+        .map_err(|e| format!("argument is not valid UTF-8: {}", e))?;
+    pyjson::parse(text).map_err(|e| format!("argument JSON parse failed: {}", e))
+}
+
+fn envelope(result: Result<Json, String>) -> *mut c_char {
+    let body = match result {
+        Ok(value) => Json::Obj(vec![
+            ("ok".to_string(), Json::Bool(true)),
+            ("value".to_string(), value),
+        ]),
+        Err(err) => Json::Obj(vec![
+            ("ok".to_string(), Json::Bool(false)),
+            ("err".to_string(), Json::Str(err)),
+        ]),
+    };
+    let out = pyjson::dumps(&body);
+    CString::new(out)
+        .map(|c| c.into_raw())
+        .unwrap_or(std::ptr::null_mut())
+}
+
+fn root_arg(v: &Json, key: &str) -> Result<std::path::PathBuf, String> {
+    match v.get(key).and_then(|x| match x {
+        Json::Str(s) => Some(s.clone()),
+        _ => None,
+    }) {
+        Some(r) if !r.is_empty() => Ok(std::path::PathBuf::from(r)),
+        _ => Err(format!("{} (root path string) is required", key)),
+    }
+}
+
+/// sp_module_imports({"module": str, "src_text": str|null}) -> [names]
+#[no_mangle]
+pub extern "C" fn sp_module_imports(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let module = a.get("module").and_then(|x| match x {
+            Json::Str(s) => Some(s.clone()),
+            _ => None,
+        }).ok_or("arg.module required")?;
+        let src_text = match a.get("src_text") {
+            Some(Json::Str(t)) => Some(t.clone()),
+            _ => None,
+        };
+        let names = gap::_module_imports(&module, src_text.as_deref())?;
+        Ok(Json::Arr(names.into_iter().map(Json::Str).collect()))
+    });
+    envelope(r)
+}
+
+/// sp_gather_meta_inputs({"root": str, "workflows_dir"?: str}) -> inputs
+#[no_mangle]
+pub extern "C" fn sp_gather_meta_inputs(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let root = root_arg(&a, "root")?;
+        let wf = a.get("workflows_dir").and_then(|x| match x {
+            Json::Str(s) => Some(std::path::PathBuf::from(s.clone())),
+            _ => None,
+        });
+        gap::_gather_meta_inputs(&root, wf.as_deref())
+    });
+    envelope(r)
+}
+
+/// sp_deployed_source_files({"root": str}) -> [relative paths]
+#[no_mangle]
+pub extern "C" fn sp_deployed_source_files(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let root = root_arg(&a, "root")?;
+        Ok(Json::Arr(
+            gap::_deployed_source_files(&root).into_iter().map(Json::Str).collect(),
+        ))
+    });
+    envelope(r)
+}
+
+/// sp_gather_drift_inputs({"root","registry","graph","tested","src_stems",
+/// "deployed_obs"}) -> inputs
+#[no_mangle]
+pub extern "C" fn sp_gather_drift_inputs(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let root = root_arg(&a, "root")?;
+        let registry: Vec<String> = a
+            .get("registry")
+            .and_then(|x| match x {
+                Json::Arr(items) => Some(
+                    items
+                        .iter()
+                        .map(|i| rune_spine::store::as_str(i).to_string())
+                        .collect(),
+                ),
+                _ => None,
+            })
+            .ok_or("arg.registry required")?;
+        let graph = a.get("graph").cloned().ok_or("arg.graph required")?;
+        let tested: Vec<String> = a
+            .get("tested")
+            .and_then(|x| match x {
+                Json::Arr(items) => Some(
+                    items
+                        .iter()
+                        .map(|i| rune_spine::store::as_str(i).to_string())
+                        .collect(),
+                ),
+                _ => None,
+            })
+            .ok_or("arg.tested required")?;
+        let src_stems: Vec<String> = a
+            .get("src_stems")
+            .and_then(|x| match x {
+                Json::Arr(items) => Some(
+                    items
+                        .iter()
+                        .map(|i| rune_spine::store::as_str(i).to_string())
+                        .collect(),
+                ),
+                _ => None,
+            })
+            .ok_or("arg.src_stems required")?;
+        let deployed_obs = a.get("deployed_obs").cloned().unwrap_or(Json::Null);
+        gap::_gather_drift_inputs(
+            &root,
+            &registry,
+            &graph,
+            &tested,
+            &src_stems,
+            &deployed_obs,
+        )
+    });
+    envelope(r)
+}
+
+/// sp_fresh_doc({"root": str, "built"?: doc}) -> doc
+#[no_mangle]
+pub extern "C" fn sp_fresh_doc(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let root = root_arg(&a, "root")?;
+        let built = a.get("built").cloned();
+        gap::_fresh_doc(&root, built.as_ref())
+    });
+    envelope(r)
+}
+
+/// sp_rust_core({"pin","loader_importable","so_present"}) -> null|"spine_loader"
+#[no_mangle]
+pub extern "C" fn sp_rust_core(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let pin = a
+            .get("pin")
+            .and_then(|x| match x {
+                Json::Str(s) => Some(s.clone()),
+                _ => None,
+            })
+            .ok_or("arg.pin required")?;
+        let loader_importable = matches!(a.get("loader_importable"), Some(Json::Bool(true)));
+        let so_present = matches!(a.get("so_present"), Some(Json::Bool(true)));
+        Ok(match gap::_rust_core(&pin, loader_importable, so_present) {
+            Some(name) => Json::Str(name.to_string()),
+            None => Json::Null,
+        })
+    });
+    envelope(r)
+}
+
+/// sp_py_sorted({"v": value}) -> normalized value
+#[no_mangle]
+pub extern "C" fn sp_py_sorted(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg)
+        .and_then(|a| a.get("v").cloned().ok_or("arg.v required".to_string()))
+        .map(|v| gap::_py_sorted(&v));
+    envelope(r)
+}
+
+/// sp_py_declared_surfaces({"manifest": manifest}) -> declared
+#[no_mangle]
+pub extern "C" fn sp_py_declared_surfaces(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let manifest = a.get("manifest").cloned().ok_or("arg.manifest required")?;
+        gap::_py_declared_surfaces(&manifest)
+    });
+    envelope(r)
+}
+
+/// sp_py_project_meta_surfaces({"doc","inputs","now"}) -> {"count","doc"}
+#[no_mangle]
+pub extern "C" fn sp_py_project_meta_surfaces(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let doc = a.get("doc").cloned().ok_or("arg.doc required")?;
+        let inputs = a.get("inputs").cloned().ok_or("arg.inputs required")?;
+        let now = a
+            .get("now")
+            .map(rune_spine::store::as_str)
+            .map(|x| x.to_string())
+            .ok_or("arg.now required")?;
+        gap::_py_project_meta_surfaces(&doc, &inputs, &now)
+    });
+    envelope(r)
+}
+
+/// sp_py_ingest_umbrellas_data({"doc","data","now"}) -> {"count","doc"}
+#[no_mangle]
+pub extern "C" fn sp_py_ingest_umbrellas_data(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let doc = a.get("doc").cloned().ok_or("arg.doc required")?;
+        let data = a.get("data").cloned().ok_or("arg.data required")?;
+        let now = a
+            .get("now")
+            .map(rune_spine::store::as_str)
+            .map(|x| x.to_string())
+            .ok_or("arg.now required")?;
+        gap::_py_ingest_umbrellas_data(&doc, &data, &now)
+    });
+    envelope(r)
+}
+
+/// sp_py_entity({"doc","id"}) -> entity|null
+#[no_mangle]
+pub extern "C" fn sp_py_entity(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let doc = a.get("doc").cloned().ok_or("arg.doc required")?;
+        let id = a
+            .get("id")
+            .map(rune_spine::store::as_str)
+            .map(|x| x.to_string())
+            .ok_or("arg.id required")?;
+        Ok(gap::_py_entity(&doc, &id))
+    });
+    envelope(r)
+}
+
+fn core_available(a: &Json) -> bool {
+    matches!(a.get("core_available"), Some(Json::Bool(true)))
+}
+
+/// sp_py_build_from_inputs({"inputs","core_available"}) -> doc
+#[no_mangle]
+pub extern "C" fn sp_py_build_from_inputs(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let inputs = a.get("inputs").cloned().ok_or("arg.inputs required")?;
+        gap::_py_build_from_inputs(&inputs, core_available(&a))
+    });
+    envelope(r)
+}
+
+/// sp_py_drift_from_inputs({"doc","inputs","core_available"}) -> drift list
+#[no_mangle]
+pub extern "C" fn sp_py_drift_from_inputs(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let doc = a.get("doc").cloned().ok_or("arg.doc required")?;
+        let inputs = a.get("inputs").cloned().ok_or("arg.inputs required")?;
+        gap::_py_drift_from_inputs(&doc, &inputs, core_available(&a))
+    });
+    envelope(r)
+}
+
+/// sp_py_export_umbrellas({"doc","now","core_available"}) -> umbrella view
+#[no_mangle]
+pub extern "C" fn sp_py_export_umbrellas(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let doc = a.get("doc").cloned().ok_or("arg.doc required")?;
+        let now = a
+            .get("now")
+            .map(rune_spine::store::as_str)
+            .map(|x| x.to_string())
+            .ok_or("arg.now required")?;
+        gap::_py_export_umbrellas(&doc, &now, core_available(&a))
+    });
+    envelope(r)
+}
+
+/// sp_py_query({"doc","id","core_available"}) -> query result
+#[no_mangle]
+pub extern "C" fn sp_py_query(arg: *const c_char) -> *mut c_char {
+    let r = read_arg(arg).and_then(|a| {
+        let doc = a.get("doc").cloned().ok_or("arg.doc required")?;
+        let id = a
+            .get("id")
+            .map(rune_spine::store::as_str)
+            .map(|x| x.to_string())
+            .ok_or("arg.id required")?;
+        gap::_py_query(&doc, &id, core_available(&a))
+    });
+    envelope(r)
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct SpineEntity {
     pub id: String,