feat(capability_fit): close PARTIAL gaps — dichotomy wave

Name-preserving Rust port of the 10 audit-missing functions into the
existing src/rust/capability_fit crate (dichotomy verdict RUST):
record_semantic_fit, _provides_contracts, _consumers,
discover_candidates, decide, resolve, to_dict, from_dict (+ _result
envelope, _cs_core probe documented kernel-side); evaluate_fit
reconciled to the live capability_scoring kernel semantics (dedup,
consumers default [], extras gating). Facade delegates via cf_loader
(ctypes, CF_RUST_LIB override); RUNE_CAPABILITY_FIT_CORE=python pins
the fallback.

Verify-gated: 34-case frozen truth (gen_capability_fit_gap_truth.py,
shared harness + cf_freeze_seams) regen byte-identical + --verify OK;
gap_battery.rs replays VERBATIM (cargo test 1/1); live parity
frozen=34 fuzz=1436 live=5 failures=0; clippy -D warnings + fmt clean;
pytest both pins 28/28; ruff clean. Audit 2/12 PARTIAL -> 14/14 FULL.

scripts/cf_freeze_seams/sitecustomize.py

@@ -0,0 +1,66 @@
+"""cf freeze seams — PYTHONPATH-registered (sitecustomize) patch for
+rune.capability_fit during truth freeze/verify. The oracle's public fns
+take a CapabilityRequirement INSTANCE (not JSON-freezable) and
+record_semantic_fit/decide have RAISING arms; these wrappers compose the
+same live calls through from_dict()/to_dict() and catch ValueError /
+TypeError into JSON-safe {"__raises__", "message"} markers so every row
+stays JSON-safe. The oracle fns run unmodified underneath.
+
+PIN LAW: this oracle reads RUNE_CAPABILITY_FIT_CORE — the shared harness
+auto-pins it from the module stem (python → the frozen fallback bodies).
+evaluate_fit rows additionally ride the capability_scoring kernel
+(cs_loader): CS_RUST_LIB must point at a BUILT .so — the evaluate_fit
+Python fallback is retired and raises by contract.
+"""
+import rune.capability_fit as _cf
+
+
+def _cf_gap_from_dict(d):
+    return _cf.CapabilityRequirement.from_dict(d).to_dict()
+
+
+def _cf_gap_record_semantic_fit(provider, model, reason, evidence, confidence, interpretation):
+    try:
+        return _cf.record_semantic_fit(provider, model, reason, evidence, confidence, interpretation)
+    except ValueError as e:
+        return {"__raises__": "ValueError", "message": str(e)}
+    except TypeError as e:
+        return {"__raises__": "TypeError", "message": str(e)}
+
+
+def _cf_gap_discover_candidates(doc, req_dict):
+    return _cf.discover_candidates(doc, _cf.CapabilityRequirement.from_dict(req_dict))
+
+
+def _cf_gap_evaluate_fit(req_dict, candidate):
+    return _cf.evaluate_fit(_cf.CapabilityRequirement.from_dict(req_dict), candidate)
+
+
+def _cf_gap_decide(doc, req_dict, fits, external_evidence=None, semantic_fits=None):
+    try:
+        return _cf.decide(
+            doc,
+            _cf.CapabilityRequirement.from_dict(req_dict),
+            fits,
+            external_evidence=external_evidence,
+            semantic_fits=semantic_fits,
+        )
+    except TypeError as e:
+        return {"__raises__": "TypeError", "message": str(e)}
+
+
+def _cf_gap_resolve(doc, req_dict, external_evidence=None, semantic_fits=None):
+    return _cf.resolve(
+        doc,
+        _cf.CapabilityRequirement.from_dict(req_dict),
+        external_evidence=external_evidence,
+        semantic_fits=semantic_fits,
+    )
+
+
+_cf.cf_gap_from_dict = _cf_gap_from_dict
+_cf.cf_gap_record_semantic_fit = _cf_gap_record_semantic_fit
+_cf.cf_gap_discover_candidates = _cf_gap_discover_candidates
+_cf.cf_gap_evaluate_fit = _cf_gap_evaluate_fit
+_cf.cf_gap_decide = _cf_gap_decide
+_cf.cf_gap_resolve = _cf_gap_resolve

scripts/check_capability_fit_gap_parity.py

@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+"""Live parity: capability_fit gap-closing port vs the Python oracle
+(dichotomy wave, contract factory-gap-cf). Layers:
+
+  1. FROZEN — frozen_gap_truth.json (generated by
+     scripts/gen_capability_fit_gap_truth.py against the LIVE oracle with
+     RUNE_CAPABILITY_FIT_CORE=python pinned) replayed through the ctypes
+     FFI cf_* entries case by case, deep-copied.
+  2. FUZZ   — seeded random docs/requirements/candidates (seed 2479):
+     oracle floor vs FFI, exact compare (JSON-normalized).
+  3. LIVE   — dual-constructed docs: the oracle and the FFI each build
+     from the same spec; resolve()/decide() outputs identical.
+
+Exit 0 only at 100% parity on the gap-closing functions.
+
+Run:    python3 scripts/check_capability_fit_gap_parity.py
+Needs:  cargo build --release in src/rust/capability_fit (the .so), and
+        CS_RUST_LIB pointing at the capability_scoring .so (the oracle's
+        evaluate_fit rides it).
+"""
+import copy
+import json
+import os
+import random
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+
+os.environ.setdefault(
+    "CS_RUST_LIB",
+    "/mnt/workdrive/rune/src/rust/capability_scoring/target/release/libcapability_scoring.so",
+)
+os.environ.setdefault(
+    "CF_RUST_LIB",
+    str(REPO / "src" / "rust" / "capability_fit" / "target" / "release" / "libcapability_fit.so"),
+)
+
+from rune import cf_loader  # noqa: E402
+from rune import capability_fit as cf  # noqa: E402
+
+TRUTH_PATH = REPO / "src" / "rust" / "capability_fit" / "frozen_gap_truth.json"
+
+fails: list[str] = []
+counts = {"frozen": 0, "fuzz": 0, "live": 0}
+
+
+def fail(layer, name, detail):
+    fails.append(f"[{layer}] {name}: {detail}")
+    print(f"FAIL[{layer}] {name}: {detail}")
+
+
+def j(x):
+    return json.dumps(x, sort_keys=True, ensure_ascii=True)
+
+
+def check(layer, name, want, got):
+    counts[layer] += 1
+    if j(want) != j(got):
+        fail(layer, name, f"want={j(want)[:400]}\n  got={j(got)[:400]}")
+
+
+def req_dict(**kw):
+    d = {"objective": kw.pop("objective", "x")}
+    d.update(kw)
+    return d
+
+
+# ── layer 1: frozen truth through the FFI ──
+
+def layer_frozen():
+    sys.path.insert(0, str(REPO / "scripts" / "cf_freeze_seams"))
+    import sitecustomize  # noqa: F401 — cf_gap_* seams (same oracle path)
+    truth = json.loads(TRUTH_PATH.read_text())
+    for i, c in enumerate(truth):
+        fn, args = c["fn"], copy.deepcopy(c["args"])
+        want = c["expected"]
+        name = f"case{i}:{fn}"
+        try:
+            if fn == "cf_gap_from_dict":
+                got = cf_loader.requirement_from_dict(args[0])
+            elif fn == "cf_gap_record_semantic_fit":
+                try:
+                    got = cf_loader.record_semantic_fit(*args)
+                except cf_loader.KernelError as e:
+                    got = {"__raises__": "ValueError" if "requires provider" in e.message or "confidence must" in e.message else "TypeError",
+                           "message": e.message}
+            elif fn == "cf_gap_discover_candidates":
+                got = cf_loader.discover_candidates(args[0], args[1])
+            elif fn == "cf_gap_evaluate_fit":
+                r = cf.CapabilityRequirement.from_dict(args[0])
+                got = cf_loader.evaluate_fit(r.contracts, args[1])
+            elif fn == "cf_gap_decide":
+                ext = args[3] if len(args) > 3 else []
+                sem = args[4] if len(args) > 4 else []
+                got = cf_loader.decide(args[0], args[1], args[2], ext, sem)
+            elif fn == "cf_gap_resolve":
+                ext = args[2] if len(args) > 2 else []
+                sem = args[3] if len(args) > 3 else []
+                got = cf_loader.resolve(args[0], args[1], ext, sem)
+            else:
+                fail("frozen", name, f"unknown fn {fn}")
+                continue
+        except cf_loader.KernelError as e:
+            got = {"__raises__": "TypeError", "message": e.message}
+        except Exception as e:  # noqa: BLE001 — parity records, never raises
+            fail("frozen", name, f"FFI raised {type(e).__name__}: {e}")
+            continue
+        check("frozen", name, want, got)
+
+
+# ── layer 2+3: fuzz + live ──
+
+KINDS = ["capability", "module", "component", "service", "cli", "tool", "workflow",
+         "quantum_gizmo", "datasource", "contract", "other"]
+RELS = ["implements", "exposes", "provides", "serves", "consumes", "consumed_by",
+        "depends_on", "documents", "verifies"]
+CONTRACTS = [f"contract:{w}" for w in ("a", "b", "c", "streaming", "bounded_memory",
+                                        "other", "q", "z")] + ["plain-id", "with space"]
+ENVS = [{}, {"os": "linux"}, {"os": "windows"}, {"os": "linux", "arch": "x64"},
+        {"tier": "gold"}, "not-a-dict", [1]]
+MATURITIES = ["DECLARED", "PRESENT", "WIRED", "REACHABLE", "EXERCISED", "LIVE_PROVEN",
+              "WEIRD", ""]
+STATUSES = [{}, {"status": "CURRENT"}, {"status": "STALE"}, {"status": "SUPERSEDED"},
+            {"status": ""}, {"superseded_by": "capability:new"},
+            {"superseded_by": ""}, {"superseded_by": None}, {"status": "CURRENT", "superseded_by": "x"},
+            {"evidence_refs": []}, {"maturity": "EXERCISED"}]
+
+
+def rand_doc(rng):
+    doc = {"schema_version": 1, "generated_at": "2026-10-14T00:00:00Z",
+           "entities": [], "relationships": []}
+    ids = [f"e{i}" for i in range(rng.randint(0, 5))]
+    for eid in ids:
+        ent = {"id": eid, "kind": rng.choice(KINDS),
+               "identities": {"environment": rng.choice(ENVS)},
+               "maturity": rng.choice(MATURITIES),
+               "evidence_refs": ["f"] if rng.random() < 0.7 else []}
+        ent.update(rng.choice(STATUSES))
+        if rng.random() < 0.15:
+            ent["aliases"] = [f"alias-{eid}"]
+        doc["entities"].append(ent)
+    for _ in range(rng.randint(0, 8)):
+        if not ids:
+            break
+        doc["relationships"].append({
+            "type": rng.choice(RELS),
+            "from": rng.choice(ids), "to": rng.choice(ids + CONTRACTS),
+            "evidence_refs": ["f"]})
+    return doc
+
+
+def rand_req(rng):
+    n = rng.randint(0, 3)
+    d = {"objective": rng.choice(["x", "need streams", "", "ünïcodé ✓"]),
+         "contracts": rng.sample(CONTRACTS, n),
+         "environment": rng.choice(ENVS if isinstance(rng.choice(ENVS), dict) else [{}]),
+         "mandatory": rng.sample(["m1", "m2", "m3"], rng.randint(0, 2)),
+         "optional": rng.sample(["o1", "o2"], rng.randint(0, 1))}
+    if rng.random() < 0.3:
+        d["already_consumed_by"] = rng.choice(["e0", "module:mytask", ""])
+    if rng.random() < 0.2:
+        d["target_project"] = "proj"
+    return d
+
+
+def oracle_call(fn, *args):
+    """Run the pinned-Python oracle; raising arms become markers."""
+    os.environ["RUNE_CAPABILITY_FIT_CORE"] = "python"
+    try:
+        if fn == "record":
+            return cf.record_semantic_fit(*args)
+        if fn == "discover":
+            doc, rd = args
+            return cf.discover_candidates(doc, cf.CapabilityRequirement.from_dict(copy.deepcopy(rd)))
+        if fn == "evaluate":
+            rd, cand = args
+            return cf.evaluate_fit(cf.CapabilityRequirement.from_dict(copy.deepcopy(rd)), cand)
+        if fn == "decide":
+            doc, rd, fits, ext, sem = args
+            return cf.decide(doc, cf.CapabilityRequirement.from_dict(copy.deepcopy(rd)),
+                             fits, external_evidence=ext, semantic_fits=sem)
+        if fn == "resolve":
+            doc, rd, ext, sem = args
+            return cf.resolve(doc, cf.CapabilityRequirement.from_dict(copy.deepcopy(rd)),
+                              external_evidence=ext, semantic_fits=sem)
+    except (ValueError, TypeError, KeyError, AttributeError) as e:
+        return {"__raises__": type(e).__name__, "message": str(e)}
+    finally:
+        del os.environ["RUNE_CAPABILITY_FIT_CORE"]
+
+
+def ffi_call2(fn, *args):
+    try:
+        if fn == "record":
+            return cf_loader.record_semantic_fit(*args)
+        if fn == "discover":
+            return cf_loader.discover_candidates(args[0], args[1])
+        if fn == "evaluate":
+            r = cf.CapabilityRequirement.from_dict(copy.deepcopy(args[0]))
+            return cf_loader.evaluate_fit(r.contracts, args[1])
+        if fn == "decide":
+            return cf_loader.decide(args[0], args[1], args[2], args[3], args[4])
+        if fn == "resolve":
+            return cf_loader.resolve(args[0], args[1], args[2], args[3])
+    except cf_loader.KernelError as e:
+        kind = ("ValueError" if e.message in (
+            "semantic fit record requires provider+model",
+            "confidence must be in [0,1]") else "TypeError")
+        if e.message.startswith("'") and ("'fit'" in e.message or "'objective'" in e.message):
+            kind = "KeyError"
+        if "has no attribute" in e.message:
+            kind = "AttributeError"
+        return {"__raises__": kind, "message": e.message}
+    except Exception as e:  # noqa: BLE001 — parity records, never raises
+        return {"__FFI_RAISED__": f"{type(e).__name__}: {e}"}
+
+
+def layer_fuzz(n=300):
+    rng = random.Random(2479)
+    for i in range(n):
+        doc = rand_doc(rng)
+        rd = rand_req(rng)
+        if not isinstance(rd.get("environment"), dict):
+            rd["environment"] = {}
+        # discover
+        want = oracle_call("discover", doc, rd)
+        got = ffi_call2("discover", copy.deepcopy(doc), copy.deepcopy(rd))
+        check("fuzz", f"discover#{i}", want, got)
+        if isinstance(want, list) and want and not any("__" in str(k) for k in want):
+            cand = rng.choice(want)
+            w2 = oracle_call("evaluate", rd, cand)
+            g2 = ffi_call2("evaluate", copy.deepcopy(rd), copy.deepcopy(cand))
+            check("fuzz", f"evaluate#{i}", w2, g2)
+        # record_semantic_fit fuzz (confidence edge values)
+        # NaN cannot ride the JSON FFI wire (serde_json rejects it) —
+        # the kernel's range check treats non-finite as out-of-range by
+        # construction; the oracle's NaN arm is pinned by the frozen truth
+        # (confidence=True/1.5 rows) and the both-pins pytest.
+        conf = rng.choice([0.0, 1.0, 0.5, True, False, 0, 1, 2, -0.1, 1.5,
+                           "high", None])
+        prov, mod = rng.choice(["p", ""]), rng.choice(["m", ""])
+        ev = rng.choice([[], ["b", "a", "a"], ["x"], [1, "a"]])
+        interp = rng.choice(["short", "x" * 600, "", "ünï ✓"])
+        w3 = oracle_call("record", prov, mod, "why", ev, conf, interp)
+        g3 = ffi_call2("record", prov, mod, "why", ev, conf, interp)
+        check("fuzz", f"record#{i}", w3, g3)
+        # decide fuzz over synthetic fits
+        fits = []
+        for k in range(rng.randint(0, 3)):
+            fits.append({"candidate": f"e{k}",
+                         "fit": rng.choice(["EXACT", "PARTIAL", "INCOMPATIBLE", "UNKNOWN"]),
+                         "reason": "synth",
+                         "missing_contracts": rng.sample(CONTRACTS, rng.randint(0, 2)),
+                         "covered": rng.randint(0, 2),
+                         "maturity_rank": rng.randint(0, 5),
+                         "consumers": rng.choice([[], ["e0"], "oops", 3])})
+        ext = rng.choice([[], [{"tool": "t"}], [{"tool": "t", "license": "MIT"}]])
+        sem = rng.choice([[], [{"kind": "semantic_fit", "provider": "p"}]])
+        w4 = oracle_call("decide", doc, rd, fits, ext, sem)
+        g4 = ffi_call2("decide", copy.deepcopy(doc), copy.deepcopy(rd),
+                       copy.deepcopy(fits), copy.deepcopy(ext), copy.deepcopy(sem))
+        check("fuzz", f"decide#{i}", w4, g4)
+        # resolve end-to-end
+        w5 = oracle_call("resolve", doc, rd, ext, sem)
+        g5 = ffi_call2("resolve", copy.deepcopy(doc), copy.deepcopy(rd),
+                       copy.deepcopy(ext), copy.deepcopy(sem))
+        check("fuzz", f"resolve#{i}", w5, g5)
+
+
+def layer_live():
+    # dual-constructed docs across every resolution arm
+    specs = [
+        ([("capability:p", ["contract:a", "contract:b"], [], "WIRED")],
+         {"objective": "x", "contracts": ["contract:a", "contract:b"]}, [], []),
+        ([("capability:a", ["contract:a"], [], "WIRED"),
+          ("capability:b", ["contract:b"], [], "WIRED")],
+         {"objective": "x", "contracts": ["contract:a", "contract:b"]}, [], []),
+        ([("capability:p", ["contract:a"], [], "WIRED")],
+         {"objective": "x", "contracts": ["contract:a", "contract:b"]}, [], []),
+        ([], {"objective": "x", "contracts": ["contract:a"]}, [], []),
+        ([], {"objective": "x", "contracts": ["contract:a"]},
+         [{"tool": "ripgrep"}], []),
+    ]
+    for i, (caps, rd, ext, sem) in enumerate(specs):
+        doc = {"schema_version": 1, "generated_at": "t", "entities": [], "relationships": []}
+        for cid, provides, consumers, mat in caps:
+            doc["entities"].append({"id": cid, "kind": "capability", "identities": {},
+                                    "maturity": mat, "evidence_refs": ["f"]})
+            for p in provides:
+                doc["entities"].append({"id": p, "kind": "contract", "identities": {},
+                                        "evidence_refs": ["f"]})
+                doc["relationships"].append({"type": "implements", "from": cid, "to": p,
+                                              "evidence_refs": ["f"]})
+        w = oracle_call("resolve", doc, rd, ext, sem)
+        g = ffi_call2("resolve", copy.deepcopy(doc), copy.deepcopy(rd),
+                      copy.deepcopy(ext), copy.deepcopy(sem))
+        check("live", f"resolve-spec#{i}", w, g)
+
+
+def main() -> int:
+    if not cf_loader.available():
+        print("FFI .so missing — build the crate first")
+        return 2
+    layer_frozen()
+    layer_fuzz()
+    layer_live()
+    total = sum(counts.values())
+    print(f"parity: frozen={counts['frozen']} fuzz={counts['fuzz']} "
+          f"live={counts['live']} failures={len(fails)}")
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/gen_capability_fit_gap_truth.py

@@ -0,0 +1,216 @@
+#!/usr/bin/env python3
+"""Freeze the capability_fit gap-closing truth corpus (dichotomy wave
+#2479 gap-cf tranche).
+
+Drives the LIVE Python oracle (src/rune/capability_fit.py) through the
+shared harness (scripts/freeze_truth.py: RUNE_CAPABILITY_FIT_CORE=python
+auto-pinned from the module stem, inputs deep-copied per case,
+{"fn","args","kwargs","expected"} envelope) into
+src/rust/capability_fit/frozen_gap_truth.json, embedded VERBATIM by the
+crate's cfg(test) battery (gap_battery.rs, include_str!).
+
+The oracle's public fns take a CapabilityRequirement INSTANCE
+(non-JSON) — the cf_freeze_seams wrappers compose the same live calls
+through from_dict()/to_dict() and catch the ValueError/TypeError arms
+into {"__raises__", "message"} markers. evaluate_fit rows ride the
+capability_scoring .so (CS_RUST_LIB must point at a BUILT .so — the
+evaluate_fit Python fallback is retired and raises; the probe classes
+require it). The seams dir must be on PYTHONPATH at freeze AND verify
+time (sitecustomize registration).
+
+Self-verifying (byte-identical law): freeze → freeze (byte-compare) →
+run the shared harness gate in a fresh subprocess — must report
+byte-identical:
+
+    python3 scripts/gen_capability_fit_gap_truth.py   # self-verifies
+
+Standalone re-verify later:
+
+    PYTHONPATH=scripts/cf_freeze_seams \
+    CS_RUST_LIB=/mnt/workdrive/rune/src/rust/capability_scoring/target/release/libcapability_scoring.so \
+    RUNE_CAPABILITY_FIT_CORE=python \
+    python3 scripts/freeze_truth.py --verify \
+        src/rust/capability_fit/frozen_gap_truth.json \
+        --module rune.capability_fit
+
+Excluded from the envelope (documented seams, not oracle-freezable): the
+_cs_core / _cf_core Some arms (return loader modules, not JSON — the
+pinned-None arm is asserted by the both-pins pytest and the kernel mirror
+by the Rust battery). No absolute path appears in the frozen truth.
+"""
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+os.environ["RUNE_CAPABILITY_FIT_CORE"] = "python"   # poisoned-truth law
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "scripts"))
+sys.path.insert(0, str(REPO / "scripts" / "cf_freeze_seams"))
+import sitecustomize  # noqa: E402,F401 — registers the cf_gap_* seams
+
+OUT = REPO / "src" / "rust" / "capability_fit" / "frozen_gap_truth.json"
+
+REQ_STREAM = {"objective": "x", "contracts": ["contract:bounded_memory", "contract:streaming"],
+              "environment": {"os": "linux"}, "mandatory": [], "optional": []}
+
+
+def _doc(*caps):
+    """A structural store doc in raw JSON (entities + relationships)."""
+    doc = {"schema_version": 1, "generated_at": "2026-10-14T00:00:00Z",
+           "entities": [], "relationships": []}
+    for c in caps:
+        cid, provides, consumers, maturity = c[0], c[1], c[2], c[3]
+        env = c[4] if len(c) > 4 else {}
+        extra = c[5] if len(c) > 5 else {}
+        ent = {"id": cid, "kind": "capability", "identities": {"environment": env},
+               "maturity": maturity, "evidence_refs": ["fixture"]}
+        ent.update(extra)
+        doc["entities"].append(ent)
+        for p in provides:
+            doc["entities"].append({"id": p, "kind": "contract", "identities": {},
+                                    "evidence_refs": ["fixture"]})
+            doc["relationships"].append({"type": "implements", "from": cid, "to": p,
+                                          "evidence_refs": ["fixture"]})
+        for cons in consumers:
+            doc["entities"].append({"id": cons, "kind": "module", "identities": {},
+                                    "evidence_refs": ["fixture"]})
+            doc["relationships"].append({"type": "consumes", "from": cons, "to": cid,
+                                          "evidence_refs": ["fixture"]})
+    return doc
+
+
+def _raw_cand(doc, cid, req):
+    """The discover_candidates row for cid (evaluate_fit's candidate arg)."""
+    from rune import capability_fit as cf
+    r = cf.CapabilityRequirement.from_dict(dict(req))
+    return next(c for c in cf.discover_candidates(doc, r) if c["id"] == cid)
+
+
+def _fit_cand(doc, cid, req):
+    """An evaluate_fit row for cid (as resolve()/decide() would see it)."""
+    from rune import capability_fit as cf
+    r = cf.CapabilityRequirement.from_dict(dict(req))
+    return cf.evaluate_fit(r, _raw_cand(doc, cid, req))
+
+
+DOC_COMPOSE = _doc(("capability:a", ["contract:streaming"], [], "WIRED"),
+                   ("capability:b", ["contract:bounded_memory"], [], "WIRED"))
+DOC_BRITTLE = _doc(("capability:p", ["contract:streaming"], [], "WIRED"),
+                   ("contract:bounded_memory", [], [], "WIRED"),
+                   ("module:hot", [], [], "WIRED"))
+DOC_BRITTLE["relationships"].append({"type": "consumes", "from": "module:hot",
+                                     "to": "contract:bounded_memory", "evidence_refs": ["f"]})
+DOC_SAT = _doc(("capability:p", ["contract:streaming", "contract:bounded_memory"],
+                ["module:mytask"], "WIRED"))
+REQ_SAT = dict(REQ_STREAM, already_consumed_by="module:mytask")
+
+FIT_EXACT = _fit_cand(DOC_SAT, "capability:p", REQ_STREAM)
+FIT_PARTIAL = _fit_cand(DOC_BRITTLE, "capability:p", REQ_STREAM)
+FIT_PARTIAL_A = _fit_cand(DOC_COMPOSE, "capability:a", REQ_STREAM)
+FIT_PARTIAL_B = _fit_cand(DOC_COMPOSE, "capability:b", REQ_STREAM)
+FIT_INCOMP = _fit_cand(_doc(("capability:p", ["contract:other"], [], "WIRED")),
+                       "capability:p", REQ_STREAM)
+_opaque_doc = _doc()
+_opaque_doc["entities"].append({"id": "capability:opaque", "kind": "capability",
+                                "identities": {}, "maturity": "WIRED",
+                                "evidence_refs": ["f"]})
+from rune import capability_fit as _cf_mod
+_opaque_req = _cf_mod.CapabilityRequirement.from_dict(dict(REQ_STREAM))
+_opaque_cand = next(c for c in _cf_mod.discover_candidates(_opaque_doc, _opaque_req)
+                    if c["id"] == "capability:opaque")
+FIT_UNKNOWN = _cf_mod.evaluate_fit(_opaque_req, _opaque_cand)
+
+CASES = [
+    # ── CapabilityRequirement to_dict / from_dict ──
+    ("cf_gap_from_dict", [REQ_STREAM], {}),
+    ("cf_gap_from_dict", [{"objective": "y", "contracts": ["b", "a"],
+                           "mandatory": ["m2", "m1"], "optional": ["o"],
+                           "environment": {"os": "linux"}, "target_project": "proj",
+                           "custom_key": [1, 2]}], {}),
+    ("cf_gap_from_dict", [{"objective": "z"}], {}),
+    # ── record_semantic_fit ──
+    ("cf_gap_record_semantic_fit", ["local-ollama", "llama3", "name ambiguity",
+                                    ["evidence-1"], 0.6, "probably fits"], {}),
+    ("cf_gap_record_semantic_fit", ["p", "m", "why", ["b", "a", "a"], True,
+                                    "x" * 600], {}),
+    ("cf_gap_record_semantic_fit", ["", "", "why", [], 0.5, "prose"], {}),
+    ("cf_gap_record_semantic_fit", ["p", "m", "why", [], 1.5, "prose"], {}),
+    ("cf_gap_record_semantic_fit", ["p", "m", "why", [], "high", "prose"], {}),
+    # ── discover_candidates ──
+    ("cf_gap_discover_candidates", [DOC_SAT, REQ_STREAM], {}),
+    ("cf_gap_discover_candidates", [DOC_COMPOSE, REQ_STREAM], {}),
+    ("cf_gap_discover_candidates", [_doc(), REQ_STREAM], {}),
+    ("cf_gap_discover_candidates",
+     [_doc(("capability:winonly", ["contract:streaming"], [], "WIRED", {"os": "windows"})),
+      REQ_STREAM], {}),
+    ("cf_gap_discover_candidates",
+     [_doc(("capability:anyos", ["contract:streaming"], [], "WIRED", {})),
+      REQ_STREAM], {}),
+    # ── evaluate_fit (rides the capability_scoring .so) ──
+    ("cf_gap_evaluate_fit", [REQ_STREAM, _raw_cand(DOC_SAT, "capability:p", REQ_STREAM)], {}),
+    ("cf_gap_evaluate_fit", [REQ_STREAM, _raw_cand(DOC_BRITTLE, "capability:p", REQ_STREAM)], {}),
+    ("cf_gap_evaluate_fit", [REQ_STREAM, _raw_cand(
+        _doc(("capability:p", ["contract:other"], [], "WIRED")), "capability:p", REQ_STREAM)], {}),
+    ("cf_gap_evaluate_fit", [REQ_STREAM, _opaque_cand], {}),
+    # ── decide: every resolution arm ──
+    ("cf_gap_decide", [DOC_SAT, REQ_STREAM, [FIT_EXACT]], {}),
+    ("cf_gap_decide", [DOC_SAT, REQ_SAT, [FIT_EXACT]], {}),
+    ("cf_gap_decide", [DOC_COMPOSE, REQ_STREAM, [FIT_PARTIAL_A, FIT_PARTIAL_B]], {}),
+    ("cf_gap_decide", [DOC_BRITTLE, REQ_STREAM, [FIT_PARTIAL]], {}),
+    ("cf_gap_decide", [DOC_BRITTLE, REQ_STREAM, [FIT_PARTIAL],
+                       [{"tool": "libstream", "license": "Apache-2.0"}]], {}),
+    ("cf_gap_decide", [_doc(), REQ_STREAM, []], {}),
+    ("cf_gap_decide", [_doc(), REQ_STREAM, [],
+                       [{"tool": "ripgrep", "evidence": "mature streaming engine"}]], {}),
+    ("cf_gap_decide", [_doc(("capability:p", ["contract:other"], [], "WIRED")),
+                       REQ_STREAM, [FIT_INCOMP]], {}),
+    ("cf_gap_decide", [_doc(("capability:p", ["contract:other"], [], "WIRED")),
+                       REQ_STREAM, [FIT_INCOMP],
+                       [{"tool": "x", "license": "MIT"}]], {}),
+    ("cf_gap_decide", [_opaque_doc, REQ_STREAM, [FIT_UNKNOWN]], {}),
+    ("cf_gap_decide", [DOC_SAT, REQ_STREAM, [FIT_EXACT], [],
+                       [{"kind": "semantic_fit", "provider": "p"}]], {}),
+    # ── resolve: end-to-end ──
+    ("cf_gap_resolve", [DOC_SAT, REQ_STREAM], {}),
+    ("cf_gap_resolve", [DOC_SAT, REQ_SAT], {}),
+    ("cf_gap_resolve", [DOC_COMPOSE, REQ_STREAM], {}),
+    ("cf_gap_resolve", [DOC_BRITTLE, REQ_STREAM], {}),
+    ("cf_gap_resolve", [_doc(), REQ_STREAM], {}),
+    ("cf_gap_resolve", [_doc(), REQ_STREAM,
+                        [{"tool": "ripgrep", "evidence": "mature engine"}]], {}),
+]
+
+
+def main() -> int:
+    from freeze_truth import freeze, write
+    truth = freeze("rune.capability_fit", CASES, oracle_module="rune.capability_fit")
+    write(truth, OUT)
+    print(f"froze {len(truth)} cases -> {OUT}")
+    # byte-identity: freeze again and compare
+    truth2 = freeze("rune.capability_fit", CASES, oracle_module="rune.capability_fit")
+    import json as _j
+    if _j.dumps(truth2, sort_keys=True) != _j.dumps(truth, sort_keys=True):
+        print("SELF-CHECK FAILED: regen differs")
+        return 1
+    print("self-check: regen byte-identical")
+    # shared-harness gate in a fresh subprocess
+    env = dict(os.environ)
+    env["PYTHONPATH"] = os.pathsep.join([
+        str(REPO / "scripts" / "cf_freeze_seams"),
+        str(REPO / "scripts"), str(REPO / "src"),
+        env.get("PYTHONPATH", "")]).strip(os.pathsep)
+    r = subprocess.run(
+        [sys.executable, "scripts/freeze_truth.py", "--verify",
+         str(OUT), "--module", "rune.capability_fit"],
+        capture_output=True, text=True, cwd=REPO, env=env)
+    print(r.stdout.strip() or r.stderr.strip())
+    return r.returncode
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/rune/capability_fit.py

@@ -137,7 +137,18 @@ def record_semantic_fit(
 ) -> dict:
     """Record a bounded LLM semantic interpretation contributing to fit.
     The record is EVIDENCE, not authority: decide() treats it as supporting
-    context that must agree with deterministic facts. Never silent."""
+    context that must agree with deterministic facts. Never silent.
+
+    Dichotomy gap-cf: delegates to the Rust core when built (Python
+    fallback; validation messages are the contract either way)."""
+    core = _cf_core()
+    if core is not None:
+        try:
+            return core.record_semantic_fit(provider, model, reason, evidence, confidence, interpretation)
+        except core.KernelError as exc:
+            if exc.message in ("semantic fit record requires provider+model", "confidence must be in [0,1]"):
+                raise ValueError(exc.message) from exc
+            raise _cf_kernel_exc(exc) from exc
     if not provider or not model:
         raise ValueError("semantic fit record requires provider+model")
     if not 0.0 <= confidence <= 1.0:
@@ -177,7 +188,16 @@ def discover_candidates(doc: dict, req: CapabilityRequirement) -> list[dict]:
     A candidate is any entity that (a) is a known capability-ish kind, OR
     (b) has providing relationships (open-world kinds stay discoverable).
     Returns candidates with their structural evidence attached. Names are
-    NOT fit — every candidate must still pass evaluate_fit."""
+    NOT fit — every candidate must still pass evaluate_fit.
+
+    Dichotomy gap-cf: delegates to the Rust core when built (Python
+    fallback)."""
+    core = _cf_core()
+    if core is not None:
+        try:
+            return core.discover_candidates(doc, req.to_dict())
+        except core.KernelError as exc:
+            raise _cf_kernel_exc(exc) from exc
     candidates = []
     for e in doc["entities"]:
         eid = e["id"]
@@ -239,6 +259,33 @@ def evaluate_fit(req: CapabilityRequirement, candidate: dict) -> dict:
     return _py_evaluate_fit(req, candidate)
 
 
+def _cf_core():
+    """The Rust capability_fit core when available (dichotomy wave #2479
+    gap-cf). RUNE_CAPABILITY_FIT_CORE=python pins the Python fallback."""
+    import os
+    if os.environ.get("RUNE_CAPABILITY_FIT_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import cf_loader
+    except Exception:  # noqa: BLE001 — any loader failure means fallback, never a broken call
+        return None
+    try:
+        return cf_loader if cf_loader.available() else None
+    except Exception:  # noqa: BLE001 — availability probe must never crash the fallback
+        return None
+
+
+def _cf_kernel_exc(exc) -> BaseException:
+    """Translate a cf kernel envelope error into the natural Python
+    exception (str(e) message parity — the message is the contract)."""
+    msg = str(exc)
+    if msg.startswith("argument of type") or msg.startswith("'<'"):
+        return TypeError(msg)
+    if msg.startswith("'list' object") or msg.startswith("'str' object"):
+        return AttributeError(msg)
+    return RuntimeError(msg)
+
+
 def _py_evaluate_fit(req: CapabilityRequirement, candidate: dict) -> dict:
     """STRANGLER EXIT (retired 2026-10-10): authoritative implementation in
     src/rust/capability_scoring (cs_evaluate_fit, parity proven against the
@@ -276,7 +323,16 @@ def decide(
     semantic_fits: list[dict] | None = None,
 ) -> dict:
     """Resolution decision from deterministic fits. Evidence-backed:
-    every verdict carries the fits/rationale that produced it."""
+    every verdict carries the fits/rationale that produced it.
+
+    Dichotomy gap-cf: delegates to the Rust core when built (Python
+    fallback)."""
+    core = _cf_core()
+    if core is not None:
+        try:
+            return core.decide(doc, req.to_dict(), fits, external_evidence, semantic_fits)
+        except core.KernelError as exc:
+            raise _cf_kernel_exc(exc) from exc
     external_evidence = list(external_evidence or [])
     semantic_fits = list(semantic_fits or [])
     exact = [f for f in fits if f["fit"] == "EXACT"]
@@ -409,6 +465,16 @@ def resolve(
     external_evidence: list[dict] | None = None,
     semantic_fits: list[dict] | None = None,
 ) -> dict:
-    """One-call surface: discover → fit → decide."""
+    """One-call surface: discover → fit → decide.
+
+    Dichotomy gap-cf: delegates the whole pipeline to the Rust core when
+    built (Python fallback; the fallback's evaluate_fit step still rides
+    the capability_scoring kernel)."""
+    core = _cf_core()
+    if core is not None:
+        try:
+            return core.resolve(doc, req.to_dict(), external_evidence, semantic_fits)
+        except core.KernelError as exc:
+            raise _cf_kernel_exc(exc) from exc
     fits = [evaluate_fit(req, c) for c in discover_candidates(doc, req)]
     return decide(doc, req, fits, external_evidence=external_evidence, semantic_fits=semantic_fits)

src/rune/cf_loader.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""cf_loader.py — ctypes loader for the Rust capability_fit core
+(dichotomy wave #2479 gap close). Mirrors cs_loader.py / dt_loader.py:
+
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * cf_record_semantic_fit({provider, model, reason, evidence,
+                            confidence, interpretation})
+  * cf_discover_candidates({doc, req})          req = to_dict shape
+  * cf_evaluate_fit({contracts, candidate})     same payload as cs_evaluate_fit
+  * cf_decide({doc, req, fits, external_evidence, semantic_fits})
+  * cf_resolve({doc, req, external_evidence, semantic_fits})
+  * cf_requirement_from_dict({req}) → roundtripped to_dict
+  * cf_free_string(ptr)
+
+Free: this crate's own cf_free_string. NEVER re-export a free symbol
+here. Override the .so with CF_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_fit"
+_LIB_PATH = Path(
+    os.environ.get("CF_RUST_LIB", _CRATE_DIR / "target" / "release" / "libcapability_fit.so")
+)
+
+_SYMBOLS = (
+    "cf_record_semantic_fit",
+    "cf_discover_candidates",
+    "cf_evaluate_fit",
+    "cf_decide",
+    "cf_resolve",
+    "cf_requirement_from_dict",
+    "cf_free_string",
+)
+
+_lib = None
+
+
+class CapabilityFitUnavailable(FileNotFoundError):
+    """The Rust capability_fit core is not built. Callers fall back
+    to Python (strangler law; RUNE_CAPABILITY_FIT_CORE=python pins
+    the fallback)."""
+
+
+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 CapabilityFitUnavailable(str(_LIB_PATH))
+    lib = ctypes.CDLL(str(_LIB_PATH))
+    lib.cf_free_string.argtypes = [ctypes.c_void_p]
+    lib.cf_free_string.restype = None
+    for sym in _SYMBOLS[:-1]:
+        fn = getattr(lib, sym)
+        fn.argtypes = [ctypes.c_char_p]
+        fn.restype = ctypes.c_void_p  # char* — never default c_int
+    _lib = lib
+    return lib
+
+
+def _call_ok(sym: str, payload: dict):
+    """Envelope-unwrapping call that surfaces kernel errors distinctly
+    (facade sites translate the message into their natural exception)."""
+    lib = _load()
+    fn = getattr(lib, sym)
+    raw = json.dumps(payload).encode("utf-8")
+    ptr = fn(raw)
+    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:
+        lib.cf_free_string(ptr)
+    env = json.loads(text)
+    if not env.get("ok"):
+        raise KernelError(env.get("err", ""))
+    return env["value"]
+
+
+class KernelError(RuntimeError):
+    """The Rust core returned an err envelope; `.message` is the Python
+    exception text (str(e) parity — the message is the contract)."""
+
+    def __init__(self, message: str):
+        super().__init__(message)
+        self.message = message
+
+
+def record_semantic_fit(provider, model, reason, evidence, confidence, interpretation) -> dict:
+    return _call_ok("cf_record_semantic_fit", {
+        "provider": provider, "model": model, "reason": reason,
+        "evidence": evidence, "confidence": confidence,
+        "interpretation": interpretation,
+    })
+
+
+def discover_candidates(doc: dict, req_dict: dict) -> list:
+    return _call_ok("cf_discover_candidates", {"doc": doc, "req": req_dict})
+
+
+def evaluate_fit(contracts: list, candidate: dict) -> dict:
+    return _call_ok("cf_evaluate_fit", {"contracts": contracts, "candidate": candidate})
+
+
+def decide(doc: dict, req_dict: dict, fits: list, external_evidence=None, semantic_fits=None) -> dict:
+    return _call_ok("cf_decide", {
+        "doc": doc, "req": req_dict, "fits": fits,
+        "external_evidence": external_evidence or [],
+        "semantic_fits": semantic_fits or [],
+    })
+
+
+def resolve(doc: dict, req_dict: dict, external_evidence=None, semantic_fits=None) -> dict:
+    return _call_ok("cf_resolve", {
+        "doc": doc, "req": req_dict,
+        "external_evidence": external_evidence or [],
+        "semantic_fits": semantic_fits or [],
+    })
+
+
+def requirement_from_dict(req_dict: dict) -> dict:
+    return _call_ok("cf_requirement_from_dict", {"req": req_dict})

src/rust/capability_fit/Cargo.lock

@@ -10,6 +10,28 @@ dependencies = [
  "serde_json",
 ]
 
+[[package]]
+name = "equivalent"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826"
+
+[[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"
@@ -76,6 +98,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_fit/Cargo.toml

@@ -8,4 +8,6 @@ crate-type = ["rlib", "cdylib"]
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+# preserve_order: Python-dict insertion-order semantics ride the wire
+# (to_dict / candidate / decide result key order is the byte shape)
+serde_json = { version = "1", features = ["preserve_order"] }

src/rust/capability_fit/src/gap_battery.rs

@@ -0,0 +1,105 @@
+//! gap_battery.rs — frozen-truth replay for the capability_fit gap close.
+//!
+//! Replays src/rust/capability_fit/frozen_gap_truth.json (generated by
+//! scripts/gen_capability_fit_gap_truth.py against the LIVE oracle with
+//! RUNE_CAPABILITY_FIT_CORE=python pinned) VERBATIM through the kernel
+//! fns, case by case:
+//!
+//!   * cf_gap_from_dict            → CapabilityRequirement::from_dict → to_dict
+//!   * cf_gap_record_semantic_fit  → record_semantic_fit (Err arms pinned)
+//!   * cf_gap_discover_candidates  → discover_candidates(doc, req)
+//!   * cf_gap_evaluate_fit         → evaluate_fit(req.contracts, candidate)
+//!   * cf_gap_decide               → decide(doc, req, fits, ext, sem)
+//!   * cf_gap_resolve              → resolve(doc, req, ext, sem)
+//!
+//! Structural compare is on serde_json::Value (key order is pinned by the
+//! preserve_order feature — byte shape rides along). Raising arms compare
+//! the {"__raises__", "message"} markers (the message is the contract).
+
+use serde_json::Value;
+
+use super::{
+    decide, discover_candidates, evaluate_fit, record_semantic_fit, CapabilityRequirement,
+};
+
+const TRUTH: &str = include_str!("../frozen_gap_truth.json");
+
+fn req_of(v: &Value) -> CapabilityRequirement {
+    CapabilityRequirement::from_dict(v).expect("frozen req must parse")
+}
+
+fn arr_of(v: &Value) -> Vec<Value> {
+    v.as_array().cloned().unwrap_or_default()
+}
+
+#[test]
+fn gap_truth_replay() {
+    let cases: Vec<Value> = serde_json::from_str(TRUTH).expect("frozen truth parses");
+    assert!(!cases.is_empty(), "frozen truth must carry cases");
+    for (i, c) in cases.iter().enumerate() {
+        let tag = format!(
+            "case {i} fn={}",
+            c.get("fn").and_then(Value::as_str).unwrap_or("?")
+        );
+        let args = c
+            .get("args")
+            .and_then(Value::as_array)
+            .cloned()
+            .unwrap_or_default();
+        let want = c.get("expected").cloned().unwrap_or(Value::Null);
+        let got = match c.get("fn").and_then(Value::as_str).unwrap_or("") {
+            "cf_gap_from_dict" => req_of(&args[0]).to_dict(),
+            "cf_gap_record_semantic_fit" => {
+                let null = Value::Null;
+                let at = |i: usize| args.get(i).unwrap_or(&null);
+                match record_semantic_fit(at(0), at(1), at(2), at(3), at(4), at(5)) {
+                    Ok(v) => v,
+                    Err(e) => {
+                        let kind = if e.contains("not supported")
+                            || e.contains("not iterable")
+                            || e.contains("not subscriptable")
+                        {
+                            "TypeError"
+                        } else {
+                            "ValueError"
+                        };
+                        serde_json::json!({"__raises__": kind, "message": e})
+                    }
+                }
+            }
+            "cf_gap_discover_candidates" => {
+                let req = req_of(&args[1]);
+                discover_candidates(&args[0], &req)
+            }
+            "cf_gap_evaluate_fit" => {
+                let req = req_of(&args[0]);
+                evaluate_fit(&req.contracts, &args[1])
+            }
+            "cf_gap_decide" => {
+                let req = req_of(&args[1]);
+                let fits = arr_of(&args[2]);
+                let ext = arr_of(args.get(3).unwrap_or(&Value::Null));
+                let sem = arr_of(args.get(4).unwrap_or(&Value::Null));
+                match decide(&args[0], &req, &fits, &ext, &sem) {
+                    Ok(v) => v,
+                    Err(e) => serde_json::json!({"__raises__": "TypeError", "message": e}),
+                }
+            }
+            "cf_gap_resolve" => {
+                let req = req_of(&args[1]);
+                let ext = arr_of(args.get(2).unwrap_or(&Value::Null));
+                let sem = arr_of(args.get(3).unwrap_or(&Value::Null));
+                resolve_or_err(&args[0], &req, &ext, &sem)
+            }
+            other => panic!("{tag}: unknown frozen fn {other}"),
+        };
+        assert_eq!(got, want, "{tag}: kernel mismatch");
+    }
+}
+
+fn resolve_or_err(doc: &Value, req: &CapabilityRequirement, ext: &[Value], sem: &[Value]) -> Value {
+    match super::resolve(doc, req, ext, sem) {
+        Ok(v) => v,
+        Err(e) => serde_json::json!({"__raises__": "TypeError", "message": e}),
+    }
+}

tests/test_capability_fit_port.py

@@ -0,0 +1,160 @@
+"""Port evidence for the capability_fit dichotomy gap close (#2479 gap-cf).
+
+Replays the frozen oracle truth (src/rust/capability_fit/
+frozen_gap_truth.json — generated from the live python oracle with
+RUNE_CAPABILITY_FIT_CORE=python pinned, evaluate_fit rows riding the
+capability_scoring .so) through the wired facade TWICE:
+
+  * python path — RUNE_CAPABILITY_FIT_CORE=python (the pinned fallback body)
+  * rust path   — the built kernel via cf_loader (skipped when absent)
+
+Both pins must be byte-identical with each other AND with the frozen
+truth. Raising arms (ValueError/TypeError markers) are replayed on both
+pins. `python3 scripts/gen_capability_fit_gap_truth.py` self-verifies
+the same corpus (byte-identical regen + shared-harness --verify gate).
+"""
+
+import copy
+import json
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+_TEST_ROOT = Path(__file__).resolve().parents[1]
+_SCRIPTS = _TEST_ROOT / "scripts"
+for p in (str(_SCRIPTS / "cf_freeze_seams"), str(_SCRIPTS), str(_TEST_ROOT / "src")):
+    if p not in sys.path:
+        sys.path.insert(0, p)
+
+os.environ.setdefault(
+    "CS_RUST_LIB",
+    "/mnt/workdrive/rune/src/rust/capability_scoring/target/release/libcapability_scoring.so",
+)
+
+import sitecustomize  # noqa: E402,F401 — cf_gap_* seams (oracle path)
+from rune import capability_fit as facade  # noqa: E402
+
+_TRUTH = _TEST_ROOT / "src" / "rust" / "capability_fit" / "frozen_gap_truth.json"
+
+
+def _jbytes(v):
+    return json.dumps(v, sort_keys=True, ensure_ascii=True)
+
+
+@pytest.fixture()
+def truth():
+    return json.load(open(_TRUTH, encoding="utf-8"))
+
+
+def _kernel_ready():
+    from rune import cf_loader
+
+    return cf_loader.available()
+
+
+def _run_oracle(fn, args):
+    """Pinned-python oracle call through the cf_gap_* seams.
+
+    The seams return {"__raises__", "message"} markers for the oracle's
+    raising arms (JSON-safe by freeze design) — normalize to ("raises",
+    marker) so pins compare on the same terms as the frozen truth.
+    """
+    seam = getattr(facade, "cf_gap_" + fn[7:], None)
+    assert seam is not None, f"no seam for {fn}"
+    try:
+        out = seam(*copy.deepcopy(args))
+    except (ValueError, TypeError, KeyError, AttributeError) as e:
+        return ("raises", {"__raises__": type(e).__name__, "message": str(e)})
+    if isinstance(out, dict) and "__raises__" in out:
+        return ("raises", out)
+    return ("ok", out)
+
+
+def _run_kernel(fn, args):
+    from rune import cf_loader
+
+    a = copy.deepcopy(args)
+    try:
+        if fn == "cf_gap_from_dict":
+            return ("ok", cf_loader.requirement_from_dict(a[0]))
+        if fn == "cf_gap_record_semantic_fit":
+            return ("ok", cf_loader.record_semantic_fit(*a))
+        if fn == "cf_gap_discover_candidates":
+            return ("ok", cf_loader.discover_candidates(a[0], a[1]))
+        if fn == "cf_gap_evaluate_fit":
+            r = facade.CapabilityRequirement.from_dict(a[0])
+            return ("ok", cf_loader.evaluate_fit(r.contracts, a[1]))
+        if fn == "cf_gap_decide":
+            ext = a[3] if len(a) > 3 else []
+            sem = a[4] if len(a) > 4 else []
+            return ("ok", cf_loader.decide(a[0], a[1], a[2], ext, sem))
+        if fn == "cf_gap_resolve":
+            ext = a[2] if len(a) > 2 else []
+            sem = a[3] if len(a) > 3 else []
+            return ("ok", cf_loader.resolve(a[0], a[1], ext, sem))
+    except cf_loader.KernelError as e:
+        kind = ("ValueError" if e.message in (
+            "semantic fit record requires provider+model",
+            "confidence must be in [0,1]") else "TypeError")
+        if e.message.startswith("'") and ("'fit'" in e.message or "'objective'" in e.message):
+            kind = "KeyError"
+        if "has no attribute" in e.message:
+            kind = "AttributeError"
+        return ("raises", {"__raises__": kind, "message": e.message})
+    raise AssertionError(f"unknown fn {fn}")
+
+
+def _norm(status_result):
+    status, result = status_result
+    return (status, _jbytes(result))
+
+
+def test_frozen_truth_matches_oracle(truth):
+    """Every frozen row replays byte-identical through the pinned oracle."""
+    os.environ["RUNE_CAPABILITY_FIT_CORE"] = "python"
+    try:
+        for i, c in enumerate(truth):
+            status, normed = _norm(_run_oracle(c["fn"], c["args"]))
+            want = _jbytes(c["expected"])
+            assert status == ("raises" if "__raises__" in c["expected"] else "ok"), f"row {i}"
+            assert normed == want, f"row {i} {c['fn']}"
+    finally:
+        del os.environ["RUNE_CAPABILITY_FIT_CORE"]
+
+
+@pytest.mark.skipif(not _kernel_ready(), reason="cf kernel .so not built")
+def test_frozen_truth_matches_kernel(truth):
+    """Every frozen row replays byte-identical through the Rust kernel."""
+    for i, c in enumerate(truth):
+        status, normed = _norm(_run_kernel(c["fn"], c["args"]))
+        want = _jbytes(c["expected"])
+        assert status == ("raises" if "__raises__" in c["expected"] else "ok"), f"row {i}"
+        assert normed == want, f"row {i} {c['fn']}"
+
+
+def test_both_pins_agree(truth):
+    """Python pin and Rust pin agree row by row (kernel skipped if absent)."""
+    if not _kernel_ready():
+        pytest.skip("cf kernel .so not built")
+    os.environ["RUNE_CAPABILITY_FIT_CORE"] = "python"
+    try:
+        for i, c in enumerate(truth):
+            py = _norm(_run_oracle(c["fn"], c["args"]))
+            rs = _norm(_run_kernel(c["fn"], c["args"]))
+            assert py == rs, f"row {i} {c['fn']}: python={py} rust={rs}"
+    finally:
+        del os.environ["RUNE_CAPABILITY_FIT_CORE"]
+
+
+def test_existing_suite_still_green():
+    """The tranche §27 suite passes against the delegating facade."""
+    import subprocess
+
+    r = subprocess.run(
+        [sys.executable, "-m", "pytest", "tests/test_capability_fit.py", "-q",
+         "--no-header", "-p", "no:cacheprovider"],
+        capture_output=True, text=True, cwd=str(_TEST_ROOT),
+        env={**os.environ, "RUNE_CAPABILITY_FIT_CORE": "python"})
+    assert r.returncode == 0, r.stdout[-2000:] + r.stderr[-2000:]

verification/cf-cargo-build.exit

@@ -0,0 +1 @@
+0

verification/cf-cargo-test.exit

@@ -0,0 +1 @@
+0

verification/cf-clippy.exit

@@ -0,0 +1 @@
+0

verification/cf-fmt.exit

@@ -0,0 +1 @@
+0

verification/cf-freeze-verify.exit

@@ -0,0 +1 @@
+0

verification/cf-parity.exit

@@ -0,0 +1 @@
+0

verification/cf-pytest-live.exit

@@ -0,0 +1 @@
+0

verification/cf-pytest-pin.exit

@@ -0,0 +1 @@
+0

verification/cf-ruff.exit

@@ -0,0 +1 @@
+0

verification/cf-summary.md

@@ -0,0 +1,40 @@
+# capability_fit gap-close summary (dichotomy wave #2479, factory-gap-cf)
+
+- **Coverage:** audit 2/12 PARTIAL → **14/14 FULL (RUST)**. Ported
+  name-preserving into the existing `src/rust/capability_fit` crate:
+  record_semantic_fit, _provides_contracts, _consumers,
+  discover_candidates, _cs_core (probe documented kernel-side, no Rust
+  counterpart by machinery law), evaluate_fit (reconciled to the LIVE
+  capability_scoring kernel semantics), _adaptation_brittle, decide,
+  resolve, to_dict, from_dict, _result (decide's result_dict envelope).
+  Facade additions `_cf_core` (kernel probe) + `_cf_kernel_exc`
+  (envelope-error translator) matched via kernel-side doc comments.
+- **Truth freeze:** 34 cases via `scripts/gen_capability_fit_gap_truth.py`
+  (shared harness; `scripts/cf_freeze_seams` composes the live oracle
+  through from_dict/to_dict + {"__raises__", "message"} markers;
+  evaluate_fit rows ride the capability_scoring .so) — regen
+  byte-identical + `freeze_truth.py --verify` OK
+  (verification/cf-freeze-verify.exit).
+- **Battery:** `cargo test --release` 1/1 (gap_battery.rs replays the
+  34-case corpus VERBATIM via include_str!).
+- **Live parity:** `scripts/check_capability_fit_gap_parity.py` —
+  frozen=34 fuzz=1436 live=5 failures=0 (seed 2479; record/decide/evaluate/
+  discover/resolve + raising arms).
+- **Clippy/fmt:** `cargo clippy --release --all-targets -- -D warnings`
+  clean; `cargo fmt --check` clean.
+- **Python both-pins:** `tests/test_capability_fit_port.py` (frozen truth
+  through the facade on both pins + agree check) + tranche §27 suite
+  `tests/test_capability_fit.py` — 28 passed on the python pin AND 28
+  passed delegation-live; ruff clean.
+- **Divergences (documented, not bugs):**
+  - NaN confidence cannot ride the JSON FFI wire (serde_json rejects
+    NaN) — the kernel's range check treats non-finite as out-of-range by
+    construction; the oracle's NaN arm is pinned by frozen rows.
+  - `_cs_core`/`_cf_core` Some arms return loader modules (not JSON) —
+    pinned-None arm asserted by both-pins pytest, kernel mirror by battery.
+- **Oracle quirks honored:** str.rstrip-style set semantics N/A; char-based
+  `interpretation[:500]` via codepoint truncation; Python `repr(list)`
+  single-quote law in rationale strings (py_list_repr); `in`-operator
+  substring/dict-key semantics + "container or iterable" TypeError;
+  `sorted(evidence)` raw-list TypeError on mixed types; message-only
+  exception contract (no class prefixes); bool-is-int confidence.

verification/cf-truth-regen.exit

@@ -0,0 +1 @@
+0