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

FULL parity port of src/rune/capability_ledger.py to the capability_ledger
Rust core (RUST verdict, issue #2479 tranche, contract
BUILDER-2026-10-11T15:50:51Z-bde08d-factory-wave):
- 17 FFI symbols (cl_new/cl_define/cl_get/cl_material_change/cl_move/
  cl_supersede/cl_retire/cl_active/cl_by_component/cl_context_packet/
  cl_save/cl_classify_version_bump/cl_bump_version/cl_attach_spec/
  cl_revise_spec/cl_spec_drift_status/cl_spec_context) via doc-envelope
  law; the ledger doc travels as a sibling envelope key so "value" stays
  oracle-verbatim; free via capability_scoring cs_free_string (no
  re-export); serde_json preserve_order for dict-position parity
- stale partial port replaced (old CapabilityEntry struct had no Python
  counterpart); classify_version_bump/bump_version kept name-compatible
- REAL divergences fixed pre-wiring (battery caught all three):
  MINOR/PATCH bumps must normalize untouched components through int()
  (leading zeros collapse everywhere — '01.02.03' MINOR = '1.3.0');
  bump arithmetic is arbitrary-precision like Python ints (u64/u128
  boundary); SEMVER_RE \\d matches Unicode Nd + int() accepts them
  (full Nd range table, verified against unicodedata)
- py_repr() mirrors Python repr() in error envelopes (str(KeyError) is
  repr'd); note/reason truncation by codepoints (py_slice)
- ctypes loader cl_loader.py mirrors broker_loader.py (LEDGER_RUST_LIB
  override); facade delegation on all 18 public fns with RUNE_LEDGER_CORE
  =python pin and Python-body fallback (kernel errors fall back so the
  proper exception types re-raise)
- battery: 16 sequences / 155 ops frozen from the live oracle
  (frozen_cl_truth.json, replayed VERBATIM in-crate) + save() byte-parity
  pinned (indent=1 sort_keys identical); 500-op live fuzz seed 2479:
  500/500 + final-file-bytes equal
- FIX-FIRST (pre-existing latent regression surfaced by this batch):
  spine unseeded path bypassed the monkeypatchable ingest_umbrellas seam
  once the rune_spine kernel was built — the generated-view refusal fired
  inside the kernel where isolation tests cannot intercept it
  (test_spine_seeds_entry_point red on master with kernel built). build()
  now routes the manifest ingest through the public seam first and hands
  the kernel a carry-attempted doc; the kernel skips its manifest branch
  when existing was provided as an object (true cold path keeps the
  load-bearing refusal). 198 ledger/spine Python tests green.

scripts/gen_capability_ledger_truth.py

@@ -0,0 +1,440 @@
+#!/usr/bin/env python3
+"""Freeze capability_ledger oracle truth for the Rust port battery
+(dichotomy factory batch, #2479 tranche).
+
+Exercises EVERY public function across all branches: CapabilityLedger
+(new/load, save, define defaults + full + duplicate, get prefixed/plain/
+unknown, material_change revisions + generation + source_locations + note
+truncation, move, supersede (plain + prefixed by, note truncation), retire,
+active, by_component, context_packet known/unknown) plus the module-level
+spec machinery (classify_version_bump full 4-bool matrix, bump_version
+valid/invalid/leading-zero/overflow, attach_spec defaults + errors,
+revise_spec classification + proposed-kind governance + unjustified bump +
+KeyError + covers_material_revision, spec_drift_status all outcomes,
+spec_context with/without spec).
+
+The wall clock is FROZEN: every op records the exact _now() value it was
+given, and the kernel receives that same string via the FFI payload — the
+kernel itself never reads a clock.
+
+Output: /tmp/cl_truth.json — replayed VERBATIM by
+src/rust/capability_ledger/src/branch_battery.rs.
+"""
+from __future__ import annotations
+
+import json
+import os
+import sys
+import tempfile
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+os.environ["RUNE_LEDGER_CORE"] = "python"  # freeze the ORACLE, never the kernel
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from rune import capability_ledger as cl  # noqa: E402
+
+# ── deterministic frozen clock ───────────────────────────────────────────
+_T0 = datetime(2026, 1, 1, tzinfo=timezone.utc)
+_tick = [0]
+
+
+def _next_now() -> str:
+    _tick[0] += 1
+    return (_T0 + timedelta(seconds=_tick[0] - 1)).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def _pin_clock(now: str) -> None:
+    """All _now() calls WITHIN one op return the same value (the kernel
+    receives exactly one now per op; live wall-clock is same-second)."""
+    cl._now = lambda n=now: n  # noqa: E731
+
+
+class Seq:
+    """One replayable op sequence against one ledger document."""
+
+    def __init__(self, name: str, sequences: list):
+        self.name = name
+        self.ops: list[dict] = []
+        self.expected: list[dict] = []
+        self.saved_files: dict[str, str] = {}
+        self.ledger: cl.CapabilityLedger | None = None
+        self.tmpdir = tempfile.mkdtemp(prefix=f"cltruth-{name}-")
+        sequences.append(self)
+
+    # -- op plumbing -------------------------------------------------------
+
+    def _call(self, opname: str, **args) -> None:
+        op = {"op": opname, **args}
+        now = _next_now()
+        _pin_clock(now)
+        if opname not in ("spec_context", "classify_version_bump",
+                          "bump_version"):
+            op["now"] = now
+        self.ops.append(op)
+        try:
+            out = getattr(self, f"_{opname}")(op)
+            # oracle returns LIVE references (rec dicts) — snapshot NOW or
+            # later ops mutate the frozen expectation
+            out = json.loads(json.dumps(out))
+            self.expected.append({"ok": True, "value": out})
+        except Exception as e:  # ValueError/SpecError/KeyError — message parity
+            self.expected.append({"ok": False, "err": str(e)})
+
+    def _new(self, op) -> dict:
+        file_text = op.get("file_text")
+        if file_text is not None:
+            (Path(self.tmpdir) / "capability_ledger.json").write_text(file_text)
+        self.ledger = cl.CapabilityLedger(self.tmpdir)
+        return self.ledger.doc
+
+    def _define(self, op) -> dict:
+        kwargs = {k: v for k, v in op.items()
+                  if k not in ("op", "now")}
+        return self.ledger.define(**kwargs)
+
+    def _get(self, op) -> dict | None:
+        return self.ledger.get(op["capability_id"])
+
+    def _material_change(self, op) -> dict:
+        kwargs = {"note": op["note"]}
+        if "generation" in op:
+            kwargs["generation"] = op["generation"]
+        if "source_locations" in op:
+            kwargs["source_locations"] = op["source_locations"]
+        return self.ledger.material_change(op["capability_id"], **kwargs)
+
+    def _move(self, op) -> dict:
+        return self.ledger.move(op["capability_id"], op["new_locations"])
+
+    def _supersede(self, op) -> dict:
+        return self.ledger.supersede(op["capability_id"], op["by"],
+                                     note=op.get("note", ""))
+
+    def _retire(self, op) -> dict:
+        return self.ledger.retire(op["capability_id"],
+                                  note=op.get("note", ""))
+
+    def _active(self, op) -> list:
+        return self.ledger.active()
+
+    def _by_component(self, op) -> list:
+        return self.ledger.by_component(op["component"])
+
+    def _context_packet(self, op) -> dict:
+        return self.ledger.context_packet(op["capability_id"])
+
+    def _save(self, op) -> None:
+        self.ledger.save()
+        self.saved_files["capability_ledger.json"] = (
+            Path(self.tmpdir) / "capability_ledger.json").read_text()
+        return None
+
+    def _classify_version_bump(self, op) -> str:
+        return cl.classify_version_bump(
+            contract_changed=op["contract_changed"],
+            behaviour_added=op["behaviour_added"],
+            consumer_break=op["consumer_break"],
+            implementation_only=op["implementation_only"])
+
+    def _bump_version(self, op) -> str:
+        return cl.bump_version(op["version"], op["kind"])
+
+    def _attach_spec(self, op) -> dict:
+        kwargs = {k: v for k, v in op.items()
+                  if k not in ("op", "now", "capability_id")}
+        return cl.attach_spec(self.ledger, op["capability_id"], **kwargs)
+
+    def _revise_spec(self, op) -> dict:
+        kwargs = {k: v for k, v in op.items()
+                  if k not in ("op", "now", "capability_id")}
+        return cl.revise_spec(self.ledger, op["capability_id"], **kwargs)
+
+    def _spec_drift_status(self, op) -> dict:
+        return cl.spec_drift_status(self.ledger, op["capability_id"])
+
+    def _spec_context(self, op) -> dict:
+        rec = self.ledger.get(op["capability_id"])
+        return cl.spec_context(rec)
+
+
+sequences: list[Seq] = []
+LONG_NOTE = "x" * 260
+LONG_REASON = "r" * 260
+
+
+def frozen_json(doc: dict) -> str:
+    """A saved ledger file EXACTLY as save() writes it (indent=1, sorted)."""
+    return json.dumps(doc, indent=1, sort_keys=True)
+
+
+# ── S1: fresh ledger — defaults, queries, save ───────────────────────────
+s = Seq("fresh-defaults-queries", sequences)
+s._call("new")
+s._call("define", capability_id="auth-token", name="Auth Token",
+        purpose="issue + verify session tokens", owning_project="rune",
+        component="component:auth-core")
+s._call("define", capability_id="cap:prefixed-given", name="Prefixed",
+        purpose="id given with prefix", owning_project="rune",
+        component="component:auth-core")
+s._call("get", capability_id="auth-token")
+s._call("get", capability_id="capability:auth-token")
+s._call("get", capability_id="nope")
+s._call("active")
+s._call("by_component", component="component:auth-core")
+s._call("by_component", component="component:nowhere")
+s._call("context_packet", capability_id="auth-token")
+s._call("context_packet", capability_id="ghost")
+s._call("save")
+
+# ── S2: define with every field provided ─────────────────────────────────
+s = Seq("define-full-fields", sequences)
+s._call("new")
+s._call("define", capability_id="full-rec", name="Full", purpose="all fields",
+        owning_project="rune", component="component:c1",
+        source_locations=["src/a.py", "src/b.py"],
+        contract="contract text", state_touched="store/scaffold",
+        dependencies=["dep-x"], consumers=["consumer-y"],
+        invariants=["inv-1"], behaviour_specs=["bs-1"],
+        constraints="no raw tokens", generation="gen-7")
+s._call("define", capability_id="empty-lists", name="Empty", purpose="empties",
+        owning_project="rune", component="component:c1",
+        source_locations=[], dependencies=[], consumers=[],
+        invariants=[], behaviour_specs=[])
+s._call("get", capability_id="full-rec")
+s._call("get", capability_id="empty-lists")
+s._call("context_packet", capability_id="full-rec")
+
+# ── S3: duplicate define raises ──────────────────────────────────────────
+s = Seq("duplicate-define", sequences)
+s._call("new")
+s._call("define", capability_id="dup", name="D", purpose="first",
+        owning_project="rune", component="component:c")
+s._call("define", capability_id="dup", name="D2", purpose="second",
+        owning_project="rune", component="component:c")
+
+# ── S4: unknown-capability errors for every mutator/query ───────────────
+s = Seq("unknown-cap-errors", sequences)
+s._call("new")
+s._call("material_change", capability_id="ghost", note="n")
+s._call("move", capability_id="ghost", new_locations=["a.py"])
+s._call("supersede", capability_id="ghost", by="other")
+s._call("retire", capability_id="ghost")
+s._call("attach_spec", capability_id="ghost", requirements=["r"],
+        expected_behaviours=["b"], non_goals=["n"], acceptance_criteria=["a"])
+s._call("revise_spec", capability_id="ghost", reason="x")
+s._call("spec_drift_status", capability_id="ghost")
+
+# ── S5: supersede / retire / move lifecycle ──────────────────────────────
+s = Seq("supersede-retire-move", sequences)
+s._call("new")
+s._call("define", capability_id="old-cap", name="Old", purpose="p",
+        owning_project="rune", component="component:c",
+        source_locations=["src/old.py"])
+s._call("define", capability_id="doomed", name="Doomed", purpose="p",
+        owning_project="rune", component="component:c")
+s._call("supersede", capability_id="old-cap", by="new-cap", note="replaced")
+s._call("supersede", capability_id="doomed", by="capability:explicit-prefixed",
+        note=LONG_NOTE)
+s._call("retire", capability_id="doomed", note="dead")
+s._call("retire", capability_id="doomed", note=LONG_NOTE)  # already retired: still appends history
+s._call("active")
+s._call("move", capability_id="old-cap", new_locations=["src/moved/old.py", "src/moved/old2.py"])
+s._call("context_packet", capability_id="old-cap")  # packet still served for superseded
+s._call("by_component", component="component:c")
+
+# ── S6: material_change revision counter + branches ─────────────────────
+s = Seq("material-change-revisions", sequences)
+s._call("new")
+s._call("define", capability_id="rev-cap", name="Rev", purpose="p",
+        owning_project="rune", component="component:c",
+        source_locations=["src/a.py"], constraints="c0")
+s._call("material_change", capability_id="rev-cap", note="first change")
+s._call("material_change", capability_id="rev-cap", note=LONG_NOTE,
+        generation="gen-9")
+s._call("material_change", capability_id="rev-cap", note="third",
+        generation="", source_locations=["src/b.py"])
+s._call("material_change", capability_id="rev-cap", note="fourth",
+        source_locations=[])
+s._call("get", capability_id="rev-cap")
+s._call("context_packet", capability_id="rev-cap")
+
+# ── S7: classify_version_bump — full 4-bool matrix ───────────────────────
+s = Seq("classify-bump-matrix", sequences)
+s._call("new")
+for cc in (False, True):
+    for ba in (False, True):
+        for cb in (False, True):
+            for io in (False, True):
+                s._call("classify_version_bump", contract_changed=cc,
+                        behaviour_added=ba, consumer_break=cb,
+                        implementation_only=io)
+
+# ── S8: bump_version — valid / invalid / zeros / overflow ────────────────
+s = Seq("bump-version-cases", sequences)
+s._call("new")
+for version, kind in [
+    ("0.1.0", "MAJOR"), ("0.1.0", "MINOR"), ("0.1.0", "PATCH"),
+    ("1.2.3", "MAJOR"), ("1.2.3", "MINOR"), ("1.2.3", "PATCH"),
+    ("0.0.9", "PATCH"), ("999.999.999", "MAJOR"), ("01.02.03", "MINOR"),
+    ("007.0.0", "PATCH"),
+    ("18446744073709551615.0.0", "MAJOR"),   # u64 boundary
+    ("99999999999999999999999999.0.0", "PATCH"),  # beyond u128
+    ("1.2", "PATCH"), ("1.2.3.4", "PATCH"), ("abc", "MAJOR"),
+    ("", "PATCH"), ("1..3", "MINOR"), ("-1.2.3", "MAJOR"),
+    ("1.2.x", "PATCH"), (" 1.2.3", "PATCH"), ("1.2.3 ", "PATCH"),
+    ("1.2.3", "BOGUS"), ("1.2.3", ""), ("1.2.3", "major"),
+]:
+    s._call("bump_version", version=version, kind=kind)
+
+# ── S9: attach_spec — defaults + explicit + errors ───────────────────────
+s = Seq("attach-spec-branches", sequences)
+s._call("new")
+s._call("define", capability_id="specd", name="S", purpose="spec'd",
+        owning_project="rune", component="component:c")
+s._call("spec_drift_status", capability_id="specd")  # no spec attached
+s._call("attach_spec", capability_id="specd", requirements=["r1", "r2"],
+        expected_behaviours=["b1"], non_goals=["ng1"],
+        acceptance_criteria=["ac1"])
+s._call("spec_drift_status", capability_id="specd")  # baseline covers rev 0
+s._call("define", capability_id="specd2", name="S2", purpose="x",
+        owning_project="rune", component="component:c")
+s._call("attach_spec", capability_id="specd2", requirements=[],
+        expected_behaviours=[], non_goals=[], acceptance_criteria=[],
+        baseline_version="1.0.0", baseline_reason="stable, evidenced",
+        governance_version_started_at="2025-12-31T23:59:59Z")
+s._call("get", capability_id="specd2")
+# errors
+s._call("define", capability_id="badbase", name="B", purpose="x",
+        owning_project="rune", component="component:c")
+s._call("attach_spec", capability_id="badbase", requirements=[],
+        expected_behaviours=[], non_goals=[], acceptance_criteria=[],
+        baseline_version="1.0")
+s._call("attach_spec", capability_id="badbase", requirements=[],
+        expected_behaviours=[], non_goals=[], acceptance_criteria=[],
+        baseline_version="not-semver")
+s._call("attach_spec", capability_id="ghost", requirements=[],
+        expected_behaviours=[], non_goals=[], acceptance_criteria=[])
+
+# ── S10: revise_spec — classification, governance, drift closure ─────────
+s = Seq("revise-spec-governance", sequences)
+s._call("new")
+s._call("define", capability_id="gov", name="G", purpose="governed",
+        owning_project="rune", component="component:c")
+s._call("attach_spec", capability_id="gov", requirements=["r"],
+        expected_behaviours=["b"], non_goals=["n"], acceptance_criteria=["a"])
+s._call("material_change", capability_id="gov", note="uncovered rev 1")
+s._call("spec_drift_status", capability_id="gov")  # DRIFT: rev 1 > covered 0
+s._call("revise_spec", capability_id="gov", implementation_only=True,
+        reason="internal tidy-up")
+s._call("spec_drift_status", capability_id="gov")  # drift closed
+s._call("material_change", capability_id="gov", note="rev 2")
+s._call("material_change", capability_id="gov", note="rev 3")
+s._call("spec_drift_status", capability_id="gov")  # 2 uncovered
+s._call("revise_spec", capability_id="gov", behaviour_added=True,
+        reason=LONG_REASON, changed_sections=["requirements"],
+        affected_consumers=["consumer-x"],
+        validation_consequences=["retest suite B"])
+s._call("spec_drift_status", capability_id="gov")
+s._call("revise_spec", capability_id="gov", consumer_break=True,
+        reason="breaking contract change")
+s._call("spec_drift_status", capability_id="gov")
+s._call("get", capability_id="gov")
+
+# ── S11: revise_spec — proposed-kind governance branches ─────────────────
+s = Seq("revise-proposed-kind", sequences)
+s._call("new")
+s._call("define", capability_id="pk", name="PK", purpose="p",
+        owning_project="rune", component="component:c")
+s._call("attach_spec", capability_id="pk", requirements=["r"],
+        expected_behaviours=["b"], non_goals=["n"], acceptance_criteria=["a"])
+s._call("revise_spec", capability_id="pk", implementation_only=True,
+        reason="ok patch, proposed matches", proposed_kind="PATCH")
+s._call("revise_spec", capability_id="pk", implementation_only=True,
+        reason="patch proposed but minor earned", proposed_kind="MINOR")
+s._call("revise_spec", capability_id="pk", behaviour_added=True,
+        reason="minor earned, downgraded to patch on purpose",
+        proposed_kind="PATCH")
+s._call("revise_spec", capability_id="pk", implementation_only=True,
+        reason="major proposed for metadata churn — unjustified",
+        proposed_kind="MAJOR")
+s._call("revise_spec", capability_id="pk", implementation_only=True,
+        reason="minor proposed for metadata churn — unjustified",
+        proposed_kind="MINOR")
+s._call("revise_spec", capability_id="pk", implementation_only=True,
+        reason="garbage proposed kind", proposed_kind="HUGE")
+s._call("revise_spec", capability_id="pk", implementation_only=True,
+        reason="empty proposed kind falls back to classify",
+        proposed_kind="")
+s._call("revise_spec", capability_id="pk", reason="no flags at all — churn")
+s._call("get", capability_id="pk")
+
+# ── S12: revise_spec on a capability with no spec ────────────────────────
+s = Seq("revise-no-spec", sequences)
+s._call("new")
+s._call("define", capability_id="bare", name="B", purpose="p",
+        owning_project="rune", component="component:c")
+s._call("revise_spec", capability_id="bare", implementation_only=True,
+        reason="x")
+
+# ── S13: save/reload roundtrip (sorted file bytes pinned) ────────────────
+s = Seq("save-reload-roundtrip", sequences)
+s._call("new")
+s._call("define", capability_id="zzz-last", name="Z", purpose="p",
+        owning_project="rune", component="component:c")
+s._call("define", capability_id="aaa-first", name="A", purpose="p",
+        owning_project="rune", component="component:c")
+s._call("save")
+saved = s.saved_files["capability_ledger.json"]
+s2 = Seq("reload-from-saved-file", sequences)
+s2._call("new", file_text=saved)
+s2._call("active")  # sorted-key order from the file
+s2._call("get", capability_id="aaa-first")
+s2._call("save")
+
+# ── S14: unicode + repr edge cases ───────────────────────────────────────
+s = Seq("unicode-repr-edges", sequences)
+s._call("new")
+s._call("define", capability_id="café-x", name="Café", purpose="rôle",
+        owning_project="rune", component="component:coffeé")
+s._call("context_packet", capability_id="café-x")
+s._call("material_change", capability_id="café-x", note="naïve change → ünïcode ✓")
+s._call("material_change", capability_id="ghost-cap", note="x")
+s._call("bump_version", version="café.1.0", kind="PATCH")
+s._call("context_packet", capability_id="quo'te")
+s._call("context_packet", capability_id='dq"te')
+
+# ── S15: spec_context with and without spec ──────────────────────────────
+s = Seq("spec-context", sequences)
+s._call("new")
+s._call("define", capability_id="ctx-bare", name="CB", purpose="purpose-y",
+        owning_project="rune", component="component:c",
+        contract="contract-y", invariants=["i1"], behaviour_specs=["bs1"])
+s._call("spec_context", capability_id="ctx-bare")  # no spec — empty spec half
+s._call("attach_spec", capability_id="ctx-bare", requirements=["req1"],
+        expected_behaviours=["eb1"], non_goals=["ngo1"],
+        acceptance_criteria=["ac1"])
+s._call("spec_context", capability_id="ctx-bare")
+s._call("material_change", capability_id="ctx-bare", note="m")
+s._call("spec_context", capability_id="ctx-bare")  # last_bump = BASELINE entry
+
+truth = {
+    "frozen": "capability_ledger full public surface + spec machinery",
+    "oracle": "src/rune/capability_ledger.py",
+    "note": "Every op records the exact _now() it was given; the kernel "
+            "receives that string via the FFI payload and never reads a "
+            "clock. saved_files pin save() byte output (Python facade "
+            "serializes with indent=1 sort_keys=True).",
+    "sequences": [
+        {"name": s.name, "ops": s.ops, "expected": s.expected,
+         "saved_files": s.saved_files,
+         "final_doc": s.ledger.doc}
+        for s in sequences
+    ],
+}
+
+out = Path("/tmp/cl_truth.json")
+out.write_text(json.dumps(truth, indent=1, ensure_ascii=False) + "\n")
+n_ops = sum(len(s.ops) for s in sequences)
+print(f"froze {len(sequences)} sequences / {n_ops} ops -> {out}")

src/rune/capability_ledger.py

@@ -15,6 +15,7 @@ the history list.
 """
 from __future__ import annotations
 
+import os
 import time
 from pathlib import Path
 
@@ -28,6 +29,22 @@ def _now() -> str:
     return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
 
 
+def _kernel():
+    """The Rust ledger core (cl_loader) when RUNE_LEDGER_CORE allows it,
+    else None — every public function delegates there first and falls
+    back to the Python body below (same process, same semantics; the
+    kernel returns the oracle output verbatim plus the updated doc)."""
+    if os.environ.get("RUNE_LEDGER_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import cl_loader
+        if cl_loader.available():
+            return cl_loader
+    except Exception:
+        return None
+    return None
+
+
 class CapabilityLedger:
     def __init__(self, store_dir: Path | str):
         self.path = Path(store_dir) / "capability_ledger.json"
@@ -37,13 +54,26 @@ class CapabilityLedger:
         if self.path.is_file():
             import json
             return json.loads(self.path.read_text())
+        k = _kernel()
+        if k is not None:
+            try:
+                return k.new(_now())["value"]
+            except Exception:
+                pass
         return {"schema_version": LEDGER_SCHEMA_VERSION,
                 "capabilities": {}, "history": [],
                 "updated_at": _now()}
 
     def save(self) -> None:
         import json
-        self.doc["updated_at"] = _now()
+        k = _kernel()
+        if k is not None:
+            try:
+                self.doc = k.save(self.doc, _now())["doc"]
+            except Exception:
+                self.doc["updated_at"] = _now()
+        else:
+            self.doc["updated_at"] = _now()
         self.path.parent.mkdir(parents=True, exist_ok=True)
         self.path.write_text(json.dumps(self.doc, indent=1, sort_keys=True))
 
@@ -61,6 +91,23 @@ class CapabilityLedger:
                generation: str = "") -> dict:
         """Create a capability record (auto-invoked by Scaffolding when a
         meaningful capability is created — rule 19Q)."""
+        k = _kernel()
+        if k is not None:
+            try:
+                out = k.define(self.doc, _now(), capability_id=capability_id,
+                               name=name, purpose=purpose,
+                               owning_project=owning_project,
+                               component=component,
+                               source_locations=source_locations,
+                               contract=contract, state_touched=state_touched,
+                               dependencies=dependencies, consumers=consumers,
+                               invariants=invariants,
+                               behaviour_specs=behaviour_specs,
+                               constraints=constraints, generation=generation)
+                self.doc = out["doc"]
+                return out["value"]
+            except Exception:
+                pass  # kernel err (or absent) — Python body re-raises properly
         cid = f"capability:{capability_id}"
         if cid in self.doc["capabilities"]:
             raise ValueError(f"capability exists: {cid}")
@@ -88,6 +135,12 @@ class CapabilityLedger:
         return rec
 
     def get(self, capability_id: str) -> dict | None:
+        k = _kernel()
+        if k is not None:
+            try:
+                return k.get(self.doc, capability_id)["value"]
+            except Exception:
+                pass
         return self.doc["capabilities"].get(
             capability_id if ":" in capability_id
             else f"capability:{capability_id}")
@@ -98,6 +151,16 @@ class CapabilityLedger:
         """Record a material change (auto-invoked after construction —
         rule 19Q.2). History entry keeps traceability; dependents' harden
         state is the behaviour ledger's job (via structural closure)."""
+        k = _kernel()
+        if k is not None:
+            try:
+                out = k.material_change(self.doc, _now(), capability_id,
+                                        note=note, generation=generation,
+                                        source_locations=source_locations)
+                self.doc = out["doc"]
+                return out["value"]
+            except Exception:
+                pass
         rec = self.get(capability_id)
         if rec is None:
             raise ValueError(f"unknown capability {capability_id!r}")
@@ -124,6 +187,14 @@ class CapabilityLedger:
     def move(self, capability_id: str, new_locations: list[str]) -> dict:
         """File move: STABLE IDENTITY SURVIVES (rule 19S-6); the move is
         recorded, history preserved."""
+        k = _kernel()
+        if k is not None:
+            try:
+                out = k.move(self.doc, _now(), capability_id, new_locations)
+                self.doc = out["doc"]
+                return out["value"]
+            except Exception:
+                pass
         rec = self.get(capability_id)
         if rec is None:
             raise ValueError(f"unknown capability {capability_id!r}")
@@ -137,6 +208,15 @@ class CapabilityLedger:
 
     def supersede(self, capability_id: str, by: str, *,
                   note: str = "") -> dict:
+        k = _kernel()
+        if k is not None:
+            try:
+                out = k.supersede(self.doc, _now(), capability_id, by,
+                                  note=note)
+                self.doc = out["doc"]
+                return out["value"]
+            except Exception:
+                pass
         rec = self.get(capability_id)
         if rec is None:
             raise ValueError(f"unknown capability {capability_id!r}")
@@ -150,6 +230,14 @@ class CapabilityLedger:
         return rec
 
     def retire(self, capability_id: str, *, note: str = "") -> dict:
+        k = _kernel()
+        if k is not None:
+            try:
+                out = k.retire(self.doc, _now(), capability_id, note=note)
+                self.doc = out["doc"]
+                return out["value"]
+            except Exception:
+                pass
         rec = self.get(capability_id)
         if rec is None:
             raise ValueError(f"unknown capability {capability_id!r}")
@@ -163,15 +251,33 @@ class CapabilityLedger:
     # -- queries (surgical context, 19E) -----------------------------------
 
     def active(self) -> list[dict]:
+        k = _kernel()
+        if k is not None:
+            try:
+                return k.active(self.doc)["value"]
+            except Exception:
+                pass
         return [r for r in self.doc["capabilities"].values()
                 if r["maturity"] not in ("retired", "superseded")]
 
     def by_component(self, component: str) -> list[dict]:
+        k = _kernel()
+        if k is not None:
+            try:
+                return k.by_component(self.doc, component)["value"]
+            except Exception:
+                pass
         return [r for r in self.active() if r["component"] == component]
 
     def context_packet(self, capability_id: str) -> dict:
         """The LOCAL context packet (19E): everything a fresh builder
         needs for one capability — bounded, structured, no reasoning."""
+        k = _kernel()
+        if k is not None:
+            try:
+                return k.context_packet(self.doc, capability_id)["value"]
+            except Exception:
+                pass
         rec = self.get(capability_id)
         if rec is None:
             return {"error": f"unknown capability {capability_id!r}"}
@@ -227,6 +333,16 @@ def classify_version_bump(*, contract_changed: bool,
                           implementation_only: bool) -> str:
     """§21D-3: derive PATCH/MINOR/MAJOR from the change shape. The model
     may propose; the machinery verifies."""
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.classify_version_bump(
+                contract_changed=contract_changed,
+                behaviour_added=behaviour_added,
+                consumer_break=consumer_break,
+                implementation_only=implementation_only)["value"]
+        except Exception:
+            pass
     if consumer_break:
         return "MAJOR"
     if contract_changed:
@@ -241,6 +357,12 @@ def classify_version_bump(*, contract_changed: bool,
 
 
 def bump_version(version: str, kind: str) -> str:
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.bump_version(version, kind)["value"]
+        except Exception:
+            pass
     m = SEMVER_RE.match(version)
     if not m:
         raise SpecError(f"invalid semver {version!r}")
@@ -267,6 +389,21 @@ def attach_spec(ledger: "CapabilityLedger", capability_id: str, *,
     the starting version is explicit — 1.0.0 only for stable contracts
     with sufficient evidence; otherwise 0.x with a reason."""
     from rune.capability_ledger import LEDGER_SCHEMA_VERSION  # noqa: F401
+    k = _kernel()
+    if k is not None:
+        try:
+            out = k.attach_spec(ledger.doc, _now(), capability_id,
+                                requirements=requirements,
+                                expected_behaviours=expected_behaviours,
+                                non_goals=non_goals,
+                                acceptance_criteria=acceptance_criteria,
+                                baseline_version=baseline_version,
+                                baseline_reason=baseline_reason,
+                                governance_version_started_at=governance_version_started_at)
+            ledger.doc = out["doc"]
+            return out["value"]
+        except Exception:
+            pass
     rec = ledger.get(capability_id)
     if rec is None:
         raise ValueError(f"unknown capability {capability_id!r}")
@@ -307,6 +444,23 @@ def revise_spec(ledger: "CapabilityLedger", capability_id: str, *,
     The model proposes; the machinery classifies and records. Material
     change without a bump is impossible (every revise computes one);
     unjustified bumps are impossible (no-change revise raises)."""
+    k = _kernel()
+    if k is not None:
+        try:
+            out = k.revise_spec(ledger.doc, _now(), capability_id,
+                                contract_changed=contract_changed,
+                                behaviour_added=behaviour_added,
+                                consumer_break=consumer_break,
+                                implementation_only=implementation_only,
+                                reason=reason,
+                                changed_sections=changed_sections,
+                                affected_consumers=affected_consumers,
+                                validation_consequences=validation_consequences,
+                                proposed_kind=proposed_kind)
+            ledger.doc = out["doc"]
+            return out["value"]
+        except Exception:
+            pass
     rec = ledger.get(capability_id)
     if rec is None:
         raise ValueError(f"unknown capability {capability_id!r}")
@@ -357,6 +511,12 @@ def spec_drift_status(ledger: "CapabilityLedger",
         → SPEC_VERSION_DRIFT + durable obligation
 
     Returns {"drift": bool, "obligation": str|None, "detail": ...}."""
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.spec_drift_status(ledger.doc, capability_id)["value"]
+        except Exception:
+            pass
     rec = ledger.get(capability_id)
     if rec is None:
         return {"drift": False, "obligation": None,
@@ -389,6 +549,12 @@ def spec_drift_status(ledger: "CapabilityLedger",
 def spec_context(rec: dict) -> dict:
     """The spec half of a surgical context packet (§21M): WHY + WHAT
     SHOULD EXIST, compact."""
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.spec_context(rec)["value"]
+        except Exception:
+            pass
     s = rec.get("spec") or {}
     return {
         "capability_id": rec["capability_id"],

src/rune/cl_loader.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+"""ctypes loader for the Rust capability_ledger core (dichotomy factory
+batch, #2479 tranche). Mirrors broker_loader.py's contract:
+
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * cl_new({now, file_text?})                          -> doc
+  * cl_define({doc, now, ...define kwargs})            -> {"record", "doc"}
+  * cl_get({doc, capability_id})                       -> record|null
+  * cl_material_change({doc, now, ...})                -> {"record", "doc"}
+  * cl_move({doc, now, capability_id, new_locations})  -> {"record", "doc"}
+  * cl_supersede({doc, now, capability_id, by, note?}) -> {"record", "doc"}
+  * cl_retire({doc, now, capability_id, note?})        -> {"record", "doc"}
+  * cl_active({doc}) / cl_by_component({doc, component}) -> [records]
+  * cl_context_packet({doc, capability_id})            -> packet dict
+  * cl_save({doc, now})                                -> doc (stamped)
+  * cl_classify_version_bump({4 bools})                -> "PATCH"|"MINOR"|"MAJOR"
+  * cl_bump_version({version, kind})                   -> new version
+  * cl_attach_spec({doc, now, ...})                    -> {"spec", "doc"}
+  * cl_revise_spec({doc, now, ...})                    -> {"result", "doc"}
+  * cl_spec_drift_status({doc, capability_id})         -> status dict
+  * cl_spec_context({rec})                             -> packet dict
+
+The doc is the FULL ledger document ({"schema_version", "capabilities",
+"history", "updated_at"}); mutators return the updated doc so the facade
+keeps authoritative state. Free: capability_scoring's cs_free_string
+(shared process) — NEVER re-export. Override the .so with
+LEDGER_RUST_LIB (a FILE path).
+"""
+from __future__ import annotations
+
+import ctypes
+import json
+import os
+from pathlib import Path
+
+_REPO = Path(__file__).resolve().parents[2]
+_CRATE_DIR = _REPO / "src" / "rust" / "capability_ledger"
+_LIB_PATH = Path(
+    os.environ.get("LEDGER_RUST_LIB", _CRATE_DIR / "target" / "release" / "libcapability_ledger.so")
+)
+
+SYMBOLS = (
+    "cl_new", "cl_define", "cl_get", "cl_material_change", "cl_move",
+    "cl_supersede", "cl_retire", "cl_active", "cl_by_component",
+    "cl_context_packet", "cl_save", "cl_classify_version_bump",
+    "cl_bump_version", "cl_attach_spec", "cl_revise_spec",
+    "cl_spec_drift_status", "cl_spec_context",
+)
+
+_lib = None
+
+
+class LedgerCoreUnavailable(FileNotFoundError):
+    """The Rust ledger core is not built. Callers fall back to Python."""
+
+
+def available() -> bool:
+    return _LIB_PATH.is_file()
+
+
+def _load() -> ctypes.CDLL:
+    global _lib
+    if _lib is not None:
+        return _lib
+    if not _LIB_PATH.is_file():
+        raise LedgerCoreUnavailable(str(_LIB_PATH))
+    lib = ctypes.CDLL(str(_LIB_PATH))
+    for sym in SYMBOLS:
+        fn = getattr(lib, sym)
+        fn.argtypes = [ctypes.c_void_p]
+        fn.restype = ctypes.c_void_p
+    # free comes from capability_scoring (same process) if loaded; else our own malloc path
+    try:
+        lib.cs_free_string
+    except AttributeError:
+        scoring = ctypes.CDLL(str(
+            _REPO / "src" / "rust" / "capability_scoring" / "target" / "release" / "libcapability_scoring.so"
+        ))
+        scoring.cs_free_string.argtypes = [ctypes.c_void_p]
+        lib._free = scoring.cs_free_string
+    _lib = lib
+    return lib
+
+
+def _call(sym: str, payload: dict) -> dict:
+    """Returns the FULL success envelope: {"value": ..., "doc": ...} for
+    mutators (the updated ledger doc travels beside the oracle output so
+    the facade can thread self.doc), {"value": ...} for queries. Errors
+    raise RuntimeError(err) — the facade's except-path falls back to the
+    Python body, which re-raises the proper exception type."""
+    lib = _load()
+    data = json.dumps(payload).encode()
+    buf = ctypes.create_string_buffer(data)
+    ptr = getattr(lib, sym)(ctypes.cast(buf, ctypes.c_void_p))
+    if not ptr:
+        raise RuntimeError(f"{sym} returned null")
+    out = ctypes.string_at(ptr).decode()
+    free = getattr(lib, "_free", None)
+    if free is not None:
+        free(ptr)
+    doc = json.loads(out)
+    if not doc.get("ok"):
+        raise RuntimeError(doc.get("err", f"{sym} failed"))
+    return doc
+
+
+def new(now: str, file_text: str | None = None) -> dict:
+    payload: dict = {"now": now}
+    if file_text is not None:
+        payload["file_text"] = file_text
+    return _call("cl_new", payload)
+
+
+def define(doc: dict, now: str, **kwargs) -> dict:
+    return _call("cl_define", {"doc": doc, "now": now, **kwargs})
+
+
+def get(doc: dict, capability_id: str) -> dict | None:
+    return _call("cl_get", {"doc": doc, "capability_id": capability_id})
+
+
+def material_change(doc: dict, now: str, capability_id: str, note: str,
+                    generation: str = "", source_locations=None) -> dict:
+    payload = {"doc": doc, "now": now, "capability_id": capability_id,
+               "note": note, "generation": generation}
+    if source_locations is not None:
+        payload["source_locations"] = source_locations
+    return _call("cl_material_change", payload)
+
+
+def move(doc: dict, now: str, capability_id: str, new_locations: list) -> dict:
+    return _call("cl_move", {"doc": doc, "now": now,
+                             "capability_id": capability_id,
+                             "new_locations": new_locations})
+
+
+def supersede(doc: dict, now: str, capability_id: str, by: str,
+              note: str = "") -> dict:
+    return _call("cl_supersede", {"doc": doc, "now": now,
+                                  "capability_id": capability_id,
+                                  "by": by, "note": note})
+
+
+def retire(doc: dict, now: str, capability_id: str, note: str = "") -> dict:
+    return _call("cl_retire", {"doc": doc, "now": now,
+                               "capability_id": capability_id, "note": note})
+
+
+def active(doc: dict) -> list:
+    return _call("cl_active", {"doc": doc})
+
+
+def by_component(doc: dict, component: str) -> list:
+    return _call("cl_by_component", {"doc": doc, "component": component})
+
+
+def context_packet(doc: dict, capability_id: str) -> dict:
+    return _call("cl_context_packet", {"doc": doc, "capability_id": capability_id})
+
+
+def save(doc: dict, now: str) -> dict:
+    return _call("cl_save", {"doc": doc, "now": now})
+
+
+def classify_version_bump(*, contract_changed: bool, behaviour_added: bool,
+                          consumer_break: bool,
+                          implementation_only: bool) -> str:
+    return _call("cl_classify_version_bump", {
+        "contract_changed": contract_changed,
+        "behaviour_added": behaviour_added,
+        "consumer_break": consumer_break,
+        "implementation_only": implementation_only,
+    })
+
+
+def bump_version(version: str, kind: str) -> str:
+    return _call("cl_bump_version", {"version": version, "kind": kind})
+
+
+def attach_spec(doc: dict, now: str, capability_id: str, **kwargs) -> dict:
+    return _call("cl_attach_spec", {"doc": doc, "now": now,
+                                    "capability_id": capability_id, **kwargs})
+
+
+def revise_spec(doc: dict, now: str, capability_id: str, **kwargs) -> dict:
+    return _call("cl_revise_spec", {"doc": doc, "now": now,
+                                    "capability_id": capability_id, **kwargs})
+
+
+def spec_drift_status(doc: dict, capability_id: str) -> dict:
+    return _call("cl_spec_drift_status", {"doc": doc,
+                                          "capability_id": capability_id})
+
+
+def spec_context(rec: dict) -> dict:
+    return _call("cl_spec_context", {"rec": rec})

src/rune/spine.py

@@ -602,12 +602,22 @@ def build(manifest_path: Path | None = None) -> dict:
     strangler law: the old path stays load-bearing until deletion is proven)."""
     inputs = _gather_build_inputs(manifest_path)
     core = _rust_core()
-    # The unseeded path is the one-time Phase A migration: it stays on the
-    # Python core so the public ingest_umbrellas seam (monkeypatched by
-    # isolation tests) is honored byte-for-byte. Steady-state (seeded)
-    # builds delegate to the Rust core; the parity battery covers BOTH
-    # paths, including an explicit unseeded-ingest parity section.
+    # The unseeded path is the one-time Phase A migration: the public
+    # ingest_umbrellas seam stays authoritative for it (isolation tests
+    # monkeypatch it; the file-driven refusal contract lives there).
+    # Route the manifest ingest through the seam FIRST, then delegate the
+    # rest of the projection with the ingested doc carried as `existing` —
+    # the kernel then takes the carry-forward branch (its own manifest
+    # ingest would bypass the seam and raise the refusal the caller is
+    # entitled to intercept; regression caught by
+    # test_t214_material_closure.test_spine_seeds_entry_point).
     seeded = bool(inputs["existing"]) and any(r["type"] == "belongs_to" for r in inputs["existing"]["relationships"])
+    if not seeded and core is not None:
+        seed_doc = new_doc()
+        mp = inputs.get("manifest_path")
+        ingest_umbrellas(seed_doc, Path(mp) if mp else None)
+        inputs = dict(inputs, existing=seed_doc, manifest=None)
+        seeded = bool(seed_doc["entities"]) or bool(seed_doc["relationships"])
     doc = core.project(inputs) if core is not None and seeded else _py_build_from_inputs(inputs)
     # both cores funnel through the same post-projection passes (the
     # contract-discovery write-back + language placement) — the doc-level

src/rust/capability_ledger/Cargo.lock

@@ -6,10 +6,31 @@ version = 4
 name = "capability_ledger"
 version = "0.1.0"
 dependencies = [
- "serde",
  "serde_json",
 ]
 
+[[package]]
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
+[[package]]
+name = "hashbrown"
+version = "0.17.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+
+[[package]]
+name = "indexmap"
+version = "2.14.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
+dependencies = [
+ "equivalent",
+ "hashbrown",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.18"
@@ -47,7 +68,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba"
 dependencies = [
  "serde_core",
- "serde_derive",
 ]
 
 [[package]]
@@ -76,6 +96,7 @@ version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
+ "indexmap",
  "itoa",
  "memchr",
  "serde",

src/rust/capability_ledger/Cargo.toml

@@ -7,5 +7,4 @@ edition = "2021"
 crate-type = ["rlib", "cdylib", "staticlib"]
 
 [dependencies]
-serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+serde_json = { version = "1", features = ["preserve_order"] }

src/rust/capability_ledger/src/branch_battery.rs

@@ -0,0 +1,163 @@
+//! capability_ledger branch battery — parity vs the frozen Python-oracle
+//! truth (frozen_cl_truth.json, 16 sequences / 155 ops across every
+//! public function and branch, frozen 2026-10-11). Embedded VERBATIM —
+//! never transcribed.
+//!
+//! The battery threads the ledger document exactly like the Python
+//! facade does: mutators return the updated doc, queries don't mutate.
+
+use crate::{
+    op_active, op_attach_spec, op_bump_version, op_by_component,
+    op_classify_version_bump, op_context_packet, op_define, op_get,
+    op_material_change, op_move, op_new, op_retire, op_revise_spec,
+    op_save, op_spec_context, op_spec_drift_status, op_supersede,
+};
+use serde_json::{json, Map, Value};
+
+const TRUTH: &str = include_str!("frozen_cl_truth.json");
+
+fn run_op(doc: &Value, op: &Value) -> (Value, Option<Value>) {
+    let name = op["op"].as_str().unwrap();
+    let mut payload = op.clone();
+    if name != "new" {
+        payload["doc"] = doc.clone();
+    }
+    if name == "spec_context" {
+        // the facade passes the rec it got from get(); resolve it here
+        let cid = op["capability_id"].as_str().unwrap_or("");
+        let key = if cid.contains(':') {
+            cid.to_string()
+        } else {
+            format!("capability:{cid}")
+        };
+        payload["rec"] = doc["capabilities"][&key].clone();
+    }
+    let r = match name {
+        "new" => op_new(&payload),
+        "define" => op_define(&payload),
+        "get" => op_get(&payload),
+        "material_change" => op_material_change(&payload),
+        "move" => op_move(&payload),
+        "supersede" => op_supersede(&payload),
+        "retire" => op_retire(&payload),
+        "active" => op_active(&payload),
+        "by_component" => op_by_component(&payload),
+        "context_packet" => op_context_packet(&payload),
+        "save" => op_save(&payload),
+        "classify_version_bump" => op_classify_version_bump(&payload),
+        "bump_version" => op_bump_version(&payload),
+        "attach_spec" => op_attach_spec(&payload),
+        "revise_spec" => op_revise_spec(&payload),
+        "spec_drift_status" => op_spec_drift_status(&payload),
+        "spec_context" => op_spec_context(&payload),
+        other => panic!("unknown op {other}"),
+    };
+    match r {
+        Ok((value, next)) => {
+            let mut m = Map::new();
+            m.insert("ok".into(), json!(true));
+            m.insert("value".into(), value.clone());
+            if let Some(d) = &next {
+                m.insert("doc".into(), d.clone());
+            }
+            // "new" returns the doc as its value; the battery threads it
+            let next = if name == "new" { Some(value) } else { next };
+            (Value::Object(m), next)
+        }
+        Err(e) => (json!({"ok": false, "err": e}), None),
+    }
+}
+
+#[test]
+fn parity_battery() {
+    let truth: Value = serde_json::from_str(TRUTH).unwrap();
+    for seq in truth["sequences"].as_array().unwrap() {
+        let name = seq["name"].as_str().unwrap();
+        let ops = seq["ops"].as_array().unwrap();
+        let expected = seq["expected"].as_array().unwrap();
+        assert_eq!(ops.len(), expected.len(), "seq {name}: op/expected skew");
+        let mut doc = Value::Null;
+        for (i, (op, exp)) in ops.iter().zip(expected.iter()).enumerate() {
+            let (envelope, next) = run_op(&doc, op);
+            // the frozen contract is {"ok", "value"}; "doc" is kernel
+            // transport for the facade's self.doc threading
+            let mut contract = envelope.clone();
+            contract.as_object_mut().unwrap().remove("doc");
+            assert_eq!(
+                &contract, exp,
+                "seq '{name}' op {i} {:?}",
+                serde_json::to_string(op).unwrap_or_default()
+            );
+            if let Some(d) = next {
+                doc = d;
+            }
+        }
+        assert_eq!(
+            &doc, &seq["final_doc"],
+            "seq '{name}': final document divergence"
+        );
+    }
+}
+
+#[test]
+fn save_bytes_pinned() {
+    // saved_files pin the Python facade's save() bytes (indent=1,
+    // sort_keys=True) — the kernel must return docs that serialize to
+    // exactly them.
+    let truth: Value = serde_json::from_str(TRUTH).unwrap();
+    for seq in truth["sequences"].as_array().unwrap() {
+        if seq["saved_files"].as_object().unwrap().is_empty() {
+            continue;
+        }
+        let mut doc = Value::Null;
+        for op in seq["ops"].as_array().unwrap() {
+            let (_env, next) = run_op(&doc, op);
+            if let Some(d) = next {
+                doc = d;
+            }
+        }
+        let sorted = sort_keys(&doc);
+        assert_eq!(
+            serde_json::to_string_pretty(&sorted)
+                .map(pretty_to_py_indent)
+                .ok(),
+            seq["saved_files"]["capability_ledger.json"].as_str().map(str::to_string),
+            "seq '{}': save() byte parity",
+            seq["name"].as_str().unwrap()
+        );
+    }
+}
+
+/// Rebuild an object with sorted keys (Python sort_keys=True semantics).
+fn sort_keys(v: &Value) -> Value {
+    match v {
+        Value::Object(m) => {
+            let mut keys: Vec<&String> = m.keys().collect();
+            keys.sort();
+            Value::Object(
+                keys.into_iter()
+                    .map(|k| (k.clone(), sort_keys(&m[k])))
+                    .collect(),
+            )
+        }
+        Value::Array(a) => Value::Array(a.iter().map(sort_keys).collect()),
+        other => other.clone(),
+    }
+}
+
+/// serde_json pretty-prints with 2-space indent — Python's indent=1 uses
+/// one space per level. Convert line by line.
+fn pretty_to_py_indent(s: String) -> String {
+    let mut out = String::with_capacity(s.len());
+    for line in s.lines() {
+        let indent = line.len() - line.trim_start().len();
+        for _ in 0..indent / 2 {
+            out.push(' ');
+        }
+        out.push_str(line.trim_start());
+        out.push('\n');
+    }
+    // Python's json.dumps emits no trailing newline
+    out.truncate(out.trim_end_matches('\n').len());
+    out
+}

src/rust/capability_ledger/src/lib.rs

@@ -1,68 +1,175 @@
-//! capability_ledger — capability tracking ledger (W6).
-use serde::{Deserialize, Serialize};
-use std::collections::BTreeMap;
+//! capability_ledger — compact versioned Capability Ledgers (T19 §19D):
+//! local re-entry context packets. A fresh LLM rehydrates the meaning of
+//! a target area after compaction/context loss/model swap/long absence
+//! WITHOUT rereading the repository.
+//!
+//! Identity rule: capability IDs are STABLE across file moves (name-
+//! based, not path-based); history is preserved on supersede/retire.
+//!
+//! FULL parity port of src/rune/capability_ledger.py (dichotomy factory
+//! batch, #2479 tranche). The ledger DOCUMENT travels across the
+//! boundary: mutators receive the current doc and return the updated doc
+//! alongside the op result — the kernel is pure, the Python facade owns
+//! file I/O and serialization (save() bytes: indent=1, sort_keys=True).
+//! Every op receives the exact `_now()` string the facade captured; the
+//! kernel never reads a clock.
+//!
+//! FFI (doc-envelope law): every symbol takes one JSON payload pointer
+//! and returns {"ok": true, "value": ...} | {"ok": false, "err": ...}.
+//! Free: capability_scoring's cs_free_string (shared process) — this
+//! crate deliberately does NOT re-export a free symbol.
+//!
+//! Key-ordering law: the oracle's dicts are insertion-ordered; the kernel
+//! parses and emits JSON with serde_json's preserve_order (IndexMap) so
+//! dict-position semantics (new key appends, existing key updates in
+//! place) match Python exactly.
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct CapabilityEntry {
-    pub name: String,
-    pub provider: String,
-    pub registered_at: String,
-    pub active: bool,
+use serde_json::{json, Map, Value};
+use std::ffi::{CStr, CString};
+use std::os::raw::c_char;
+
+/// Ops return (oracle-output, optional updated doc); the doc travels as a
+/// sibling "doc" key in the success envelope so "value" stays verbatim.
+pub type OpResult = Result<(Value, Option<Value>), String>;
+
+const LEDGER_SCHEMA_VERSION: i64 = 1;
+const SPEC_SCHEMA_VERSION: i64 = 1;
+
+/// Python's `\d` in SEMVER_RE matches ANY Unicode Nd (decimal digit)
+/// codepoint, and int() accepts them — so must we. Nd runs are contiguous
+/// with digit value = codepoint - run start (verified against
+/// unicodedata across the whole BMP+ planes).
+const ND_RANGES: &[(u32, u32)] = &[
+    (0x30, 0x39), (0x660, 0x669), (0x6F0, 0x6F9), (0x7C0, 0x7C9),
+    (0x966, 0x96F), (0x9E6, 0x9EF), (0xA66, 0xA6F), (0xAE6, 0xAEF),
+    (0xB66, 0xB6F), (0xBE6, 0xBEF), (0xC66, 0xC6F), (0xCE6, 0xCEF),
+    (0xD66, 0xD6F), (0xDE6, 0xDEF), (0xE50, 0xE59), (0xED0, 0xED9),
+    (0xF20, 0xF29), (0x1040, 0x1049), (0x1090, 0x1099), (0x17E0, 0x17E9),
+    (0x1810, 0x1819), (0x1946, 0x194F), (0x19D0, 0x19D9), (0x1A80, 0x1A89),
+    (0x1A90, 0x1A99), (0x1B50, 0x1B59), (0x1BB0, 0x1BB9), (0x1C40, 0x1C49),
+    (0x1C50, 0x1C59), (0xA620, 0xA629), (0xA8D0, 0xA8D9), (0xA900, 0xA909),
+    (0xA9D0, 0xA9D9), (0xA9F0, 0xA9F9), (0xAA50, 0xAA59), (0xABF0, 0xABF9),
+    (0xFF10, 0xFF19), (0x104A0, 0x104A9), (0x10D30, 0x10D39),
+    (0x10D40, 0x10D49), (0x11066, 0x1106F), (0x110F0, 0x110F9),
+    (0x11136, 0x1113F), (0x111D0, 0x111D9), (0x112F0, 0x112F9),
+    (0x11450, 0x11459), (0x114D0, 0x114D9), (0x11650, 0x11659),
+    (0x116C0, 0x116C9), (0x116D0, 0x116D9), (0x116DA, 0x116E3),
+    (0x11730, 0x11739), (0x118E0, 0x118E9), (0x11950, 0x11959),
+    (0x11BF0, 0x11BF9), (0x11C50, 0x11C59), (0x11D50, 0x11D59),
+    (0x11DA0, 0x11DA9), (0x11F50, 0x11F59), (0x16130, 0x16139),
+    (0x16A60, 0x16A69), (0x16AC0, 0x16AC9), (0x16B50, 0x16B59),
+    (0x16D70, 0x16D79), (0x1CCF0, 0x1CCF9), (0x1D7CE, 0x1D7D7),
+    (0x1D7D8, 0x1D7E1), (0x1D7E2, 0x1D7EB), (0x1D7EC, 0x1D7F5),
+    (0x1D7F6, 0x1D7FF), (0x1E140, 0x1E149), (0x1E2F0, 0x1E2F9),
+    (0x1E4F0, 0x1E4F9), (0x1E5F1, 0x1E5FA), (0x1E950, 0x1E959),
+    (0x1FBF0, 0x1FBF9),
+];
+
+fn nd_digit(c: char) -> Option<u32> {
+    let cp = c as u32;
+    let idx = ND_RANGES
+        .binary_search_by(|&(lo, hi)| {
+            if cp < lo {
+                std::cmp::Ordering::Greater
+            } else if cp > hi {
+                std::cmp::Ordering::Less
+            } else {
+                std::cmp::Ordering::Equal
+            }
+        })
+        .ok()?;
+    Some(cp - ND_RANGES[idx].0)
 }
 
-pub struct CapabilityLedger {
-    pub capabilities: BTreeMap<String, CapabilityEntry>,
-}
-
-impl CapabilityLedger {
-    pub fn new() -> Self {
-        CapabilityLedger {
-            capabilities: BTreeMap::new(),
+/// Python repr() of a str: quote preference (' unless the string contains
+/// ' and not "), standard one-char escapes, \xNN for ASCII control.
+fn py_repr(s: &str) -> String {
+    let dq = s.contains('\'') && !s.contains('"');
+    let quote = if dq { '"' } else { '\'' };
+    let mut out = String::with_capacity(s.len() + 2);
+    out.push(quote);
+    for c in s.chars() {
+        match c {
+            '\\' => out.push_str("\\\\"),
+            '\n' => out.push_str("\\n"),
+            '\r' => out.push_str("\\r"),
+            '\t' => out.push_str("\\t"),
+            '\x07' => out.push_str("\\a"),
+            '\x08' => out.push_str("\\b"),
+            '\x0b' => out.push_str("\\v"),
+            '\x0c' => out.push_str("\\f"),
+            c if c == quote => {
+                out.push('\\');
+                out.push(c);
+            }
+            c if (c as u32) < 0x20 || (c as u32) == 0x7f => {
+                out.push_str(&format!("\\x{:02x}", c as u32));
+            }
+            c => out.push(c),
         }
     }
-    pub fn register(&mut self, name: &str, provider: &str) {
-        self.capabilities.insert(
-            name.to_string(),
-            CapabilityEntry {
-                name: name.to_string(),
-                provider: provider.to_string(),
-                registered_at: now(),
-                active: true,
-            },
-        );
-    }
+    out.push(quote);
+    out
 }
 
-impl Default for CapabilityLedger {
-    fn default() -> Self {
-        Self::new()
-    }
+/// Python str slicing `s[:n]` — by code points, not bytes.
+fn py_slice(s: &str, n: usize) -> String {
+    s.chars().take(n).collect()
 }
 
-fn now() -> String {
-    format!(
-        "{}",
-        std::time::SystemTime::now()
-            .duration_since(std::time::UNIX_EPOCH)
-            .unwrap()
-            .as_secs()
-    )
+fn is_semver(version: &str) -> bool {
+    let parts: Vec<&str> = version.split('.').collect();
+    parts.len() == 3
+        && parts
+            .iter()
+            .all(|p| !p.is_empty() && p.chars().all(|c| nd_digit(c).is_some()))
 }
 
-/// classify_version_bump — §21D-3: derive PATCH/MINOR/MAJOR from the
-/// change shape (oracle: capability_ledger.classify_version_bump).
-/// Precedence: consumer_break > contract_changed > behaviour_added >
-/// implementation_only; all-false is SpecError (metadata churn never bumps).
-pub fn classify_version_bump(
-    contract_changed: bool,
-    behaviour_added: bool,
-    consumer_break: bool,
-    implementation_only: bool,
-) -> Result<&'static str, String> {
-    if consumer_break {
-        return Ok("MAJOR");
+/// int() semantics on a digit string: leading zeros collapse ("007" →
+/// "7", "000" → "0").
+fn norm_component(part: &str) -> String {
+    let digits: Vec<u32> = part.chars().map(|c| nd_digit(c).unwrap()).collect();
+    let first_nz = digits
+        .iter()
+        .position(|&d| d != 0)
+        .unwrap_or(digits.len() - 1);
+    digits[first_nz..]
+        .iter()
+        .map(|d| char::from_digit(*d, 10).unwrap())
+        .collect()
+}
+
+/// Decimal +1 on a digit string with int() semantics ("007" → "8").
+/// Unbounded — no overflow, exactly like Python's arbitrary-precision ints.
+fn bump_component(part: &str) -> String {
+    let digits: Vec<u32> = norm_component(part)
+        .chars()
+        .map(|c| nd_digit(c).unwrap())
+        .collect();
+    let mut ds = digits;
+    let mut i = ds.len();
+    loop {
+        i -= 1;
+        if ds[i] == 9 {
+            ds[i] = 0;
+            if i == 0 {
+                ds.insert(0, 1);
+                break;
+            }
+        } else {
+            ds[i] += 1;
+            break;
+        }
     }
-    if contract_changed {
+    ds.iter()
+        .map(|d| char::from_digit(*d, 10).unwrap())
+        .collect()
+}
+
+fn classify_raw(contract_changed: bool, behaviour_added: bool,
+                consumer_break: bool, implementation_only: bool)
+    -> Result<&'static str, String> {
+    if consumer_break || contract_changed {
         return Ok("MAJOR");
     }
     if behaviour_added {
@@ -71,68 +178,580 @@ pub fn classify_version_bump(
     if implementation_only {
         return Ok("PATCH");
     }
-    Err("no version-worthy change: metadata churn does not bump".to_string())
+    Err("no version-worthy change: metadata churn does not bump".into())
 }
 
-/// bump_version — strict ^(\d+)\.(\d+)\.(\d+)$ (oracle SEMVER_RE);
-/// MAJOR resets minor+patch, MINOR resets patch. Errors mirror SpecError.
-pub fn bump_version(version: &str, kind: &str) -> Result<String, String> {
-    let parts: Vec<&str> = version.split('.').collect();
-    if parts.len() != 3
-        || parts
-            .iter()
-            .any(|p| p.is_empty() || !p.bytes().all(|b| b.is_ascii_digit()))
-    {
-        return Err(format!("invalid semver {:?}", version));
-    }
-    let nums: Vec<u64> = parts.iter().map(|p| p.parse().unwrap()).collect();
+fn bump_order(kind: &str) -> Result<u8, String> {
     match kind {
-        "MAJOR" => Ok(format!("{}.0.0", nums[0] + 1)),
-        "MINOR" => Ok(format!("{}.{}.0", nums[0], nums[1] + 1)),
-        "PATCH" => Ok(format!("{}.{}.{}", nums[0], nums[1], nums[2] + 1)),
-        _ => Err(format!("unknown bump kind {:?}", kind)),
+        "PATCH" => Ok(0),
+        "MINOR" => Ok(1),
+        "MAJOR" => Ok(2),
+        // str(KeyError) is repr(key)
+        _ => Err(py_repr(kind)),
     }
 }
 
+fn bump_version_str(version: &str, kind: &str) -> Result<String, String> {
+    let parts: Vec<&str> = version.split('.').collect();
+    if !is_semver(version) {
+        return Err(format!("invalid semver {}", py_repr(version)));
+    }
+    match kind {
+        "MAJOR" => Ok(format!("{}.0.0", bump_component(parts[0]))),
+        "MINOR" => Ok(format!("{}.{}.0", norm_component(parts[0]), bump_component(parts[1]))),
+        "PATCH" => Ok(format!(
+            "{}.{}.{}",
+            norm_component(parts[0]),
+            norm_component(parts[1]),
+            bump_component(parts[2])
+        )),
+        _ => Err(format!("unknown bump kind {}", py_repr(kind))),
+    }
+}
+
+// ── doc helpers ──────────────────────────────────────────────────────────
+
+fn resolve_key(_doc: &Value, capability_id: &str) -> String {
+    if capability_id.contains(':') {
+        capability_id.to_string()
+    } else {
+        format!("capability:{capability_id}")
+    }
+}
+
+fn ledger_get<'a>(doc: &'a Value, capability_id: &str) -> Option<&'a Value> {
+    let key = resolve_key(doc, capability_id);
+    doc.get("capabilities").and_then(|c| c.get(&key))
+}
+
+fn ledger_take(doc: &mut Value, capability_id: &str)
+    -> Result<(String, Value), String> {
+    let key = resolve_key(doc, capability_id);
+    match doc.get("capabilities").and_then(|c| c.get(&key)) {
+        Some(rec) => Ok((key, rec.clone())),
+        None => Err(format!("unknown capability {}", py_repr(capability_id))),
+    }
+}
+
+fn put_rec(doc: &mut Value, key: &str, rec: Value) {
+    doc["capabilities"][key] = rec;
+}
+
+fn push_history(doc: &mut Value, entry: Value) {
+    doc["history"].as_array_mut().unwrap().push(entry);
+}
+
+fn arg_str<'a>(a: &'a Value, k: &str, default: &'a str) -> &'a str {
+    a.get(k).and_then(|v| v.as_str()).unwrap_or(default)
+}
+
+fn arg_arr(a: &Value, k: &str) -> Value {
+    a.get(k)
+        .and_then(|v| v.as_array())
+        .cloned()
+        .map(Value::Array)
+        .unwrap_or_else(|| json!([]))
+}
+
+fn arg_bool(a: &Value, k: &str) -> bool {
+    a.get(k).and_then(|v| v.as_bool()).unwrap_or(false)
+}
+
+// ── ops (the oracle, one fn per public Python function) ──────────────────
+
+pub fn op_new(a: &Value) -> OpResult {
+    let now = arg_str(a, "now", "");
+    let doc = match a.get("file_text").and_then(|v| v.as_str()) {
+        // _load: the file IS the doc (Python facade read it; we parse it)
+        Some(t) => serde_json::from_str(t).map_err(|e| e.to_string())?,
+        None => json!({
+            "schema_version": LEDGER_SCHEMA_VERSION,
+            "capabilities": {},
+            "history": [],
+            "updated_at": now,
+        }),
+    };
+    // the doc IS the value here; the battery threads it from "value"
+    Ok((doc, None))
+}
+
+pub fn op_define(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    let now = arg_str(a, "now", "");
+    let cid = format!("capability:{}", arg_str(a, "capability_id", ""));
+    if ledger_get(&doc, &cid).is_some() {
+        return Err(format!("capability exists: {cid}"));
+    }
+    let rec = json!({
+        "capability_id": cid,
+        "name": arg_str(a, "name", ""),
+        "purpose": arg_str(a, "purpose", ""),
+        "owning_project": arg_str(a, "owning_project", ""),
+        "component": arg_str(a, "component", ""),
+        "source_locations": arg_arr(a, "source_locations"),
+        "contract": arg_str(a, "contract", ""),
+        "state_touched": arg_str(a, "state_touched", ""),
+        "dependencies": arg_arr(a, "dependencies"),
+        "consumers": arg_arr(a, "consumers"),
+        "invariants": arg_arr(a, "invariants"),
+        "behaviour_specs": arg_arr(a, "behaviour_specs"),
+        "maturity": "implemented",
+        "constraints": arg_str(a, "constraints", ""),
+        "generation": arg_str(a, "generation", ""),
+        "limitations": [],
+        "created_at": now,
+        "updated_at": now,
+    });
+    put_rec(&mut doc, &cid, rec.clone());
+    Ok((rec, Some(doc)))
+}
+
+pub fn op_get(a: &Value) -> OpResult {
+    let doc = &a["doc"];
+    let cid = arg_str(a, "capability_id", "");
+    Ok((ledger_get(doc, cid).cloned().unwrap_or(Value::Null), None))
+}
+
+pub fn op_material_change(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    let now = arg_str(a, "now", "");
+    let (key, mut rec) =
+        ledger_take(&mut doc, arg_str(a, "capability_id", ""))?;
+    // T21.1 §21.1D: deterministic revision counter, not wall-clock ordering
+    let rev = rec
+        .get("material_revision")
+        .and_then(|v| v.as_i64())
+        .unwrap_or(0)
+        + 1;
+    rec["material_revision"] = json!(rev);
+    push_history(
+        &mut doc,
+        json!({
+            "capability_id": rec["capability_id"],
+            "at": now,
+            "event": "material_change",
+            "note": py_slice(arg_str(a, "note", ""), 200),
+            "generation": arg_str(a, "generation", ""),
+            "prior_maturity": rec["maturity"],
+            "material_revision": rev,
+        }),
+    );
+    rec["updated_at"] = json!(now);
+    let generation = arg_str(a, "generation", "");
+    if !generation.is_empty() {
+        rec["generation"] = json!(generation);
+    }
+    if let Some(sl) = a.get("source_locations") {
+        if !sl.is_null() {
+            rec["source_locations"] = sl.clone();
+        }
+    }
+    put_rec(&mut doc, &key, rec.clone());
+    Ok((rec, Some(doc)))
+}
+
+pub fn op_move(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    let now = arg_str(a, "now", "");
+    let (key, mut rec) =
+        ledger_take(&mut doc, arg_str(a, "capability_id", ""))?;
+    let old = rec["source_locations"].clone();
+    let new_locations = a.get("new_locations").cloned().unwrap_or(json!([]));
+    rec["source_locations"] = new_locations.clone();
+    rec["updated_at"] = json!(now);
+    push_history(
+        &mut doc,
+        json!({
+            "capability_id": rec["capability_id"],
+            "at": now,
+            "event": "moved",
+            "from": old,
+            "to": new_locations,
+        }),
+    );
+    put_rec(&mut doc, &key, rec.clone());
+    Ok((rec, Some(doc)))
+}
+
+pub fn op_supersede(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    let now = arg_str(a, "now", "");
+    let (key, mut rec) =
+        ledger_take(&mut doc, arg_str(a, "capability_id", ""))?;
+    let by_in = arg_str(a, "by", "");
+    let by = if by_in.contains(':') {
+        by_in.to_string()
+    } else {
+        format!("capability:{by_in}")
+    };
+    rec["maturity"] = json!("superseded");
+    rec["superseded_by"] = json!(by);
+    rec["updated_at"] = json!(now);
+    push_history(
+        &mut doc,
+        json!({
+            "capability_id": rec["capability_id"],
+            "at": now,
+            "event": "superseded",
+            "by": rec["superseded_by"],
+            "note": py_slice(arg_str(a, "note", ""), 160),
+        }),
+    );
+    put_rec(&mut doc, &key, rec.clone());
+    Ok((rec, Some(doc)))
+}
+
+pub fn op_retire(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    let now = arg_str(a, "now", "");
+    let (key, mut rec) =
+        ledger_take(&mut doc, arg_str(a, "capability_id", ""))?;
+    rec["maturity"] = json!("retired");
+    rec["updated_at"] = json!(now);
+    push_history(
+        &mut doc,
+        json!({
+            "capability_id": rec["capability_id"],
+            "at": now,
+            "event": "retired",
+            "note": py_slice(arg_str(a, "note", ""), 160),
+        }),
+    );
+    put_rec(&mut doc, &key, rec.clone());
+    Ok((rec, Some(doc)))
+}
+
+fn is_active(rec: &Value) -> bool {
+    let m = rec.get("maturity").and_then(|v| v.as_str()).unwrap_or("");
+    m != "retired" && m != "superseded"
+}
+
+pub fn op_active(a: &Value) -> OpResult {
+    let caps = a["doc"]
+        .get("capabilities")
+        .and_then(|c| c.as_object())
+        .cloned()
+        .unwrap_or_else(Map::new);
+    Ok((
+        Value::Array(caps.values().filter(|r| is_active(r)).cloned().collect()),
+        None,
+    ))
+}
+
+pub fn op_by_component(a: &Value) -> OpResult {
+    let component = arg_str(a, "component", "");
+    let caps = a["doc"]
+        .get("capabilities")
+        .and_then(|c| c.as_object())
+        .cloned()
+        .unwrap_or_else(Map::new);
+    Ok((
+        Value::Array(
+            caps.values()
+                .filter(|r| is_active(r))
+                .filter(|r| {
+                    r.get("component").and_then(|v| v.as_str()) == Some(component)
+                })
+                .cloned()
+                .collect(),
+        ),
+        None,
+    ))
+}
+
+pub fn op_context_packet(a: &Value) -> OpResult {
+    let doc = &a["doc"];
+    let cid = arg_str(a, "capability_id", "");
+    let Some(rec) = ledger_get(doc, cid) else {
+        return Ok((json!({"error": format!("unknown capability {}", py_repr(cid))}), None));
+    };
+    let g = |k: &str| rec.get(k).cloned().unwrap_or(Value::Null);
+    Ok((json!({
+        "schema_version": LEDGER_SCHEMA_VERSION,
+        "capability_id": g("capability_id"),
+        "name": g("name"),
+        "purpose": g("purpose"),
+        "component": g("component"),
+        "source_locations": g("source_locations"),
+        "contract": g("contract"),
+        "state_touched": g("state_touched"),
+        "dependencies": g("dependencies"),
+        "consumers": g("consumers"),
+        "invariants": g("invariants"),
+        "behaviour_specs": g("behaviour_specs"),
+        "maturity": g("maturity"),
+        "constraints": g("constraints"),
+        "generation": g("generation"),
+        "limitations": g("limitations"),
+    }), None))
+}
+
+pub fn op_save(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    doc["updated_at"] = json!(arg_str(a, "now", ""));
+    // oracle save() returns None; the stamped doc travels as the sibling
+    Ok((Value::Null, Some(doc)))
+}
+
+pub fn op_classify_version_bump(a: &Value) -> OpResult {
+    Ok((
+        json!(classify_raw(
+            arg_bool(a, "contract_changed"),
+            arg_bool(a, "behaviour_added"),
+            arg_bool(a, "consumer_break"),
+            arg_bool(a, "implementation_only"),
+        )?),
+        None,
+    ))
+}
+
+pub fn op_bump_version(a: &Value) -> OpResult {
+    Ok((
+        json!(bump_version_str(arg_str(a, "version", ""), arg_str(a, "kind", ""))?),
+        None,
+    ))
+}
+
+pub fn op_attach_spec(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    let now = arg_str(a, "now", "");
+    let (key, mut rec) =
+        ledger_take(&mut doc, arg_str(a, "capability_id", ""))?;
+    let baseline_version = arg_str(a, "baseline_version", "0.1.0").to_string();
+    if !is_semver(&baseline_version) {
+        return Err(format!(
+            "invalid baseline semver {}",
+            py_repr(&baseline_version)
+        ));
+    }
+    let baseline_reason = arg_str(a, "baseline_reason", "initial governed baseline");
+    let gvs = arg_str(a, "governance_version_started_at", now);
+    let spec = json!({
+        "schema_version": SPEC_SCHEMA_VERSION,
+        "version": baseline_version,
+        "governance_version_started_at": gvs,
+        "baseline_reason": baseline_reason,
+        "requirements": arg_arr(a, "requirements"),
+        "expected_behaviours": arg_arr(a, "expected_behaviours"),
+        "non_goals": arg_arr(a, "non_goals"),
+        "acceptance_criteria": arg_arr(a, "acceptance_criteria"),
+        "version_history": [{
+            "at": now,
+            "version": baseline_version,
+            "kind": "BASELINE",
+            "reason": baseline_reason,
+            "changed_sections": [],
+            "affected_consumers": [],
+            "validation_consequences": [],
+        }],
+    });
+    rec["spec"] = spec.clone();
+    rec["updated_at"] = json!(now);
+    put_rec(&mut doc, &key, rec);
+    Ok((spec, Some(doc)))
+}
+
+pub fn op_revise_spec(a: &Value) -> OpResult {
+    let mut doc = a.get("doc").cloned().unwrap_or(Value::Null);
+    let now = arg_str(a, "now", "");
+    let (key, mut rec) =
+        ledger_take(&mut doc, arg_str(a, "capability_id", ""))?;
+    let cid = arg_str(a, "capability_id", "");
+    let mut spec = match rec.get("spec") {
+        Some(s) if !s.is_null() => s.clone(),
+        _ => {
+            return Err(format!("{cid} has no spec — attach one first"));
+        }
+    };
+    // the machinery classifies; the model may only PROPOSE (never upgrade)
+    let actual = classify_raw(
+        arg_bool(a, "contract_changed"),
+        arg_bool(a, "behaviour_added"),
+        arg_bool(a, "consumer_break"),
+        arg_bool(a, "implementation_only"),
+    )?;
+    let proposed = arg_str(a, "proposed_kind", "");
+    let kind = if proposed.is_empty() {
+        actual.to_string()
+    } else {
+        if bump_order(proposed)? > bump_order(actual)? {
+            return Err(format!(
+                "proposed {proposed} exceeds change shape {actual} — unjustified bump"
+            ));
+        }
+        proposed.to_string()
+    };
+    let old = spec["version"].as_str().unwrap_or("").to_string();
+    let new = bump_version_str(&old, &kind)?;
+    spec["version"] = json!(new);
+    spec["version_history"]
+        .as_array_mut()
+        .unwrap()
+        .push(json!({
+            "at": now,
+            "version": new,
+            "kind": kind,
+            "reason": py_slice(arg_str(a, "reason", ""), 200),
+            "changed_sections": arg_arr(a, "changed_sections"),
+            "affected_consumers": arg_arr(a, "affected_consumers"),
+            "validation_consequences": arg_arr(a, "validation_consequences"),
+            // the governed decision COVERS all material revisions up to
+            // and including the current one — drift closes deterministically
+            "covers_material_revision": rec
+                .get("material_revision")
+                .and_then(|v| v.as_i64())
+                .unwrap_or(0),
+        }));
+    rec["spec"] = spec;
+    rec["updated_at"] = json!(now);
+    put_rec(&mut doc, &key, rec);
+    Ok((
+        json!({"old": old, "new": new, "kind": kind, "reason": arg_str(a, "reason", "")}),
+        Some(doc),
+    ))
+}
+
+pub fn op_spec_drift_status(a: &Value) -> OpResult {
+    let doc = &a["doc"];
+    let cid = arg_str(a, "capability_id", "");
+    let Some(rec) = ledger_get(doc, cid) else {
+        return Ok((json!({
+            "drift": false, "obligation": Value::Null,
+            "detail": format!("unknown capability {}", py_repr(cid)),
+        }), None));
+    };
+    let Some(spec) = rec.get("spec").filter(|s| !s.is_null()) else {
+        return Ok((json!({
+            "drift": false, "obligation": Value::Null,
+            "detail": "no spec attached — nothing to drift from",
+        }), None));
+    };
+    let hist = spec.get("version_history").and_then(|v| v.as_array());
+    let covered = match hist {
+        Some(h) if !h.is_empty() => h
+            .last()
+            .and_then(|e| e.get("covers_material_revision"))
+            .and_then(|v| v.as_i64())
+            .unwrap_or(0),
+        _ => 0,
+    };
+    let current = rec
+        .get("material_revision")
+        .and_then(|v| v.as_i64())
+        .unwrap_or(0);
+    let detail = json!({
+        "material_revision": current,
+        "governed_revision": covered,
+        "spec_version": spec.get("version").cloned().unwrap_or(Value::Null),
+    });
+    if current > covered {
+        Ok((json!({
+            "drift": true,
+            "obligation": format!(
+                "SPEC_VERSION_DRIFT: {cid} has {} uncovered material revision(s) \
+                 (material rev {current} > governed rev {covered}) — \
+                 run revise_spec (PATCH/MINOR/MAJOR) to close",
+                current - covered
+            ),
+            "detail": detail,
+        }), None))
+    } else {
+        Ok((json!({"drift": false, "obligation": Value::Null, "detail": detail}), None))
+    }
+}
+
+pub fn op_spec_context(a: &Value) -> OpResult {
+    let rec = match a.get("rec") {
+        Some(r) if !r.is_null() => r.clone(),
+        _ => return Err("spec_context: rec is null".into()),
+    };
+    let s = rec
+        .get("spec")
+        .filter(|s| !s.is_null())
+        .cloned()
+        .unwrap_or_else(|| json!({}));
+    let gk = |v: &Value, k: &str, d: Value| v.get(k).cloned().unwrap_or(d);
+    let last_bump = match s.get("version_history").and_then(|v| v.as_array()) {
+        Some(h) if !h.is_empty() => h.last().cloned().unwrap(),
+        _ => json!({}),
+    };
+    Ok((
+        json!({
+            "capability_id": rec.get("capability_id").cloned().unwrap_or(Value::Null),
+            "version": gk(&s, "version", Value::Null),
+            "purpose": rec.get("purpose").cloned().unwrap_or(Value::Null),
+            "requirements": gk(&s, "requirements", json!([])),
+            "expected_behaviours": gk(&s, "expected_behaviours", json!([])),
+            "non_goals": gk(&s, "non_goals", json!([])),
+            "acceptance_criteria": gk(&s, "acceptance_criteria", json!([])),
+            "contract": rec.get("contract").cloned().unwrap_or(Value::Null),
+            "invariants": gk(&rec, "invariants", json!([])),
+            "behaviour_specs": gk(&rec, "behaviour_specs", json!([])),
+            "last_bump": last_bump,
+        }),
+        None,
+    ))
+}
+
+// ── FFI (doc-envelope law) ───────────────────────────────────────────────
+
+fn read_arg(ptr: *const c_char) -> Result<Value, String> {
+    if ptr.is_null() {
+        return Err("null argument pointer".into());
+    }
+    let x = unsafe { CStr::from_ptr(ptr) }.to_str().map_err(|e| e.to_string())?;
+    serde_json::from_str(x).map_err(|e| e.to_string())
+}
+
+fn envelope(r: OpResult) -> *mut c_char {
+    let v = match r {
+        Ok((value, doc)) => {
+            let mut m = Map::new();
+            m.insert("ok".into(), json!(true));
+            m.insert("value".into(), value);
+            if let Some(d) = doc {
+                m.insert("doc".into(), d);
+            }
+            Value::Object(m)
+        }
+        Err(e) => json!({"ok": false, "err": e}),
+    };
+    CString::new(serde_json::to_string(&v).unwrap_or_default())
+        .map(|c| c.into_raw())
+        .unwrap_or(std::ptr::null_mut())
+}
+
+macro_rules! ffi_op {
+    ($name:ident, $op:path) => {
+        #[no_mangle]
+        pub extern "C" fn $name(arg: *const c_char) -> *mut c_char {
+            envelope(read_arg(arg).and_then(|a| $op(&a)))
+        }
+    };
+}
+
+ffi_op!(cl_new, op_new);
+ffi_op!(cl_define, op_define);
+ffi_op!(cl_get, op_get);
+ffi_op!(cl_material_change, op_material_change);
+ffi_op!(cl_move, op_move);
+ffi_op!(cl_supersede, op_supersede);
+ffi_op!(cl_retire, op_retire);
+ffi_op!(cl_active, op_active);
+ffi_op!(cl_by_component, op_by_component);
+ffi_op!(cl_context_packet, op_context_packet);
+ffi_op!(cl_save, op_save);
+ffi_op!(cl_classify_version_bump, op_classify_version_bump);
+ffi_op!(cl_bump_version, op_bump_version);
+ffi_op!(cl_attach_spec, op_attach_spec);
+ffi_op!(cl_revise_spec, op_revise_spec);
+ffi_op!(cl_spec_drift_status, op_spec_drift_status);
+ffi_op!(cl_spec_context, op_spec_context);
+
+// NOTE: no cl_free_string export — sibling crates sharing the Python
+// process already export identical free symbols; a duplicate breaks the
+// cdylib link. Callers free through capability_scoring's cs_free_string.
+
+#[path = "branch_battery.rs"]
 #[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn version_bump_parity() {
-        // batch 132 (oracle-probed): precedence + strict-semver + errors
-        assert_eq!(
-            classify_version_bump(false, false, true, false),
-            Ok("MAJOR")
-        );
-        assert_eq!(
-            classify_version_bump(true, false, false, false),
-            Ok("MAJOR")
-        );
-        assert_eq!(
-            classify_version_bump(false, true, false, false),
-            Ok("MINOR")
-        );
-        assert_eq!(
-            classify_version_bump(false, false, false, true),
-            Ok("PATCH")
-        );
-        assert!(classify_version_bump(false, false, false, false).is_err());
-        assert_eq!(bump_version("1.2.3", "MAJOR"), Ok("2.0.0".to_string()));
-        assert_eq!(bump_version("1.2.3", "MINOR"), Ok("1.3.0".to_string()));
-        assert_eq!(bump_version("1.2.3", "PATCH"), Ok("1.2.4".to_string()));
-        assert_eq!(bump_version("0.0.9", "PATCH"), Ok("0.0.10".to_string()));
-        assert!(bump_version("1.2", "PATCH").is_err());
-        assert!(bump_version("abc", "MAJOR").is_err());
-        assert!(bump_version("1.2.3.4", "PATCH").is_err());
-        assert!(bump_version("", "PATCH").is_err());
-        assert!(bump_version("1.2.3", "BOGUS").is_err());
-    }
-
-    #[test]
-    fn test_register() {
-        let mut ledger = CapabilityLedger::new();
-        ledger.register("test_cap", "test_provider");
-        assert!(ledger.capabilities.contains_key("test_cap"));
-    }
-}
+mod branch_battery;

src/rust/rune_spine/src/project.rs

@@ -379,7 +379,13 @@ pub fn project(inputs: &Json) -> Result<Json, String> {
         }
     }
     let manifest = obj_get(inputs, "manifest").cloned().unwrap_or(Json::Null);
-    if !seeded {
+    // `existing` provided as an object means carry-forward was attempted by
+    // the caller (the facade routes the unseeded manifest ingest through the
+    // public ingest_umbrellas seam first — isolation tests monkeypatch it).
+    // Only the TRUE cold path (no existing at all) ingests the manifest
+    // here, where the generated-view refusal is load-bearing.
+    let carry_attempted = matches!(existing, Json::Obj(_));
+    if !seeded && !carry_attempted {
         match &manifest {
             Json::Null => return Err("manifest unreadable and store unseeded".to_string()),
             m => {