feat(contracts): close PARTIAL gaps — dichotomy wave

Name-preserving 1:1 port of the 7 audit-missing functions (_ct_core,
_resolve_and_project, _run_contract_python, capture_signatures,
save_signatures, load_signatures, main as contracts_main) into
src/rust/contracts with IO behind injectable seams (CoreProbe /
Invoker / Signaturer / MainIO); 8 new FFI kernels (ct_gate,
ct_resolve, ct_run_python, ct_run_all, ct_capture, ct_save, ct_load,
ct_main) wired in contracts_loader.

Frozen truth: src/rust/contracts/src/ct_gap_truth.json (127 rows,
freeze byte-identical + harness --verify 23/23 clean); battery
src/rust/contracts/src/gap_battery.rs 19/19 green; live parity
replay failures=0 (facade 30, kernel 15); clippy -D warnings clean;
Python both pins green. Audit: contracts 11/11 FULL.

One documented eager/lazy divergence (projection-crash row): crash on
the delegated path, PASS on the lazy oracle path — pinned both ways.

scripts/ct_rig.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""ct_rig.py — deterministic fixture modules for the contracts gap-close
+frozen truth (dichotomy wave, #2479 tranche).
+
+The contracts oracle resolves modules with a bare ``__import__``; two of
+its branches need modules no stdlib provides:
+
+* import-time crash (``except Exception -> crash`` on import): see
+  ``ct_rig_badimport.py`` (raises RuntimeError at import).
+* projection-time crash (``_project`` raising on a hostile result):
+  ``make_bad_len`` returns an object whose ``__len__`` raises, so
+  ``hasattr(result, "__len__")`` is True but ``len(result)`` explodes.
+
+``make_ok`` / ``ATTR_VALUE`` cover the plain found paths without depending
+on any third-party module. Everything here is pure and deterministic —
+safe to import from both the freeze generator and ``--verify`` (both run
+with ``scripts/`` on ``sys.path``).
+"""
+
+ATTR_VALUE = "ct-rig-stable-value"
+
+
+class _BadLen:
+    def __len__(self):
+        raise RuntimeError("boom")
+
+
+def make_bad_len():
+    """Return an object that breaks ``_project``'s ``len()`` call."""
+    return _BadLen()
+
+
+def make_ok():
+    """Return a plain JSON-safe dict (the found-path control case)."""
+    return {"ok": True, "n": 3}
+
+
+def make_none():
+    """Return None (the expect_not_none-fail control case)."""
+    return None
+
+
+def raise_on_call():
+    """Raise on call (the call-crash control case)."""
+    raise ValueError("rig call failure")

scripts/ct_rig_badimport.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+"""ct_rig_badimport.py — import-crash fixture for the contracts gap-close
+frozen truth (dichotomy wave, #2479 tranche).
+
+Importing this module raises a non-ImportError, exercising the oracle's
+``except Exception -> crash`` import branch in ``_resolve_and_project``
+(and ``_run_contract_python``'s outer catch). Deterministic by
+construction. See ``ct_rig.py`` for the companion fixtures.
+"""
+
+raise RuntimeError("rig import failure")

scripts/gen_contracts_gap_truth.py

@@ -0,0 +1,584 @@
+#!/usr/bin/env python3
+"""gen_contracts_gap_truth.py — freeze/replay parity truth for the contracts
+gap-close (dichotomy wave, #2479 tranche).
+
+Freeze mode (default): exercises the LIVE Python oracle
+(src/rune/contracts.py — the 7 functions the audit names as missing:
+_ct_core, _resolve_and_project, _run_contract_python, capture_signatures,
+save_signatures, load_signatures, main, plus run_contract/run_all_contracts
+companions) and records the truth into
+src/rust/contracts/src/ct_gap_truth.json.
+
+The file has two sections:
+
+* ``cases`` — shared-harness schema, re-verifiable with
+  ``PYTHONPATH=src python3 scripts/freeze_truth.py --verify
+  src/rust/contracts/src/ct_gap_truth.json --module rune.contracts``.
+  Contract-taking fns cross via the harness ``recv`` form: the receiver
+  spec rebuilds a live ``Contract`` (cls __new__ + attrs, no IO) and the
+  fn is called as an unbound-style first argument. ``expect_type`` stays
+  None because a type object is not JSON-safe — the type-check branches
+  are covered by the pre-existing 58-case ``ct_truth.json`` battery,
+  replayed through the new ``_run_contract_python`` Rust kernel.
+  ``check_signature_drift()`` reads the real signatures file (no args) so
+  it stays out of the harness set (its compare kernel is already frozen
+  in ct_truth.json's drift_cases).
+* ``extra`` — oracle-derived rows the harness cannot round-trip
+  (``ContractResult`` dataclasses are not JSON-native; ``save/load`` touch
+  the signatures file; ``main`` drives argv/stdout/exit-code):
+  ``result_cases`` (run_contract / _run_contract_python asdicts),
+  ``run_all`` (42 asdicts), ``probes`` (loader-state -> gate bool),
+  ``file_cases`` (per-case deterministic dirs, no paths in expectations),
+  ``main_cases`` (argv/stdout/exit-code). Embedded VERBATIM in the
+  ``gap_battery`` alongside ``cases``.
+
+Replay mode (--replay): re-runs the SAME scenarios with the Rust core
+delegation ACTIVE and asserts facade-level and kernel-level parity.
+
+Poisoned-truth law: freeze pins RUNE_CONTRACTS_CORE=python BEFORE any
+oracle import (the shared harness also pins it per module stem).
+Mutation law: case inputs are deep-copied BEFORE they reach the oracle.
+Deterministic-dirs law: file/main cases redirect SIGNATURES_FILE into
+/tmp/ct-gap-freeze/<case>/ (fixed names — identical at freeze, --verify
+and --replay); expectations never embed the paths.
+"""
+import copy
+import dataclasses
+import io
+import json
+import os
+import sys
+from contextlib import redirect_stdout
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TRUTH_PATH = os.path.join(REPO, "src", "rust", "contracts",
+                          "src", "ct_gap_truth.json")
+FREEZE_ROOT = "/tmp/ct-gap-freeze"
+REPLAY = "--replay" in sys.argv
+
+sys.path.insert(0, os.path.join(REPO, "scripts"))
+sys.path.insert(0, os.path.join(REPO, "src"))
+from freeze_truth import freeze  # noqa: E402
+
+if not REPLAY:
+    os.environ["RUNE_CONTRACTS_CORE"] = "python"  # poisoned-truth law
+else:
+    os.environ.pop("RUNE_CONTRACTS_CORE", None)  # live-parity law
+
+sys.path.insert(0, os.path.join(REPO, "src", "rune"))
+import contracts as oracle_mod  # noqa: E402
+
+
+def contract_dict(name, module, function, args=(), kwargs=None, **kw):
+    """A JSON-safe Contract field dict (exact dataclass field set)."""
+    return {
+        "name": name,
+        "module": module,
+        "function": function,
+        "args": list(args),
+        "kwargs": dict(kwargs or {}),
+        "expect_type": None,
+        "expect_keys": kw.get("expect_keys"),
+        "expect_len": kw.get("expect_len"),
+        "expect_not_none": kw.get("expect_not_none", False),
+        "expect_callable": kw.get("expect_callable", False),
+        "skip_on_import_error": kw.get("skip_on_import_error", True),
+    }
+
+
+def recv_case(fn, contract, args=()):
+    """Harness case in recv form: {"cls": "Contract", <field>: value}
+    rebuilds a live Contract receiver; the fn is called fn(contract, *args).
+    The harness records the result AND the post-call receiver attrs —
+    contracts are never mutated, so attrs echo the inputs."""
+    recv = {"cls": "Contract"}
+    recv.update(copy.deepcopy(contract))
+    return {"fn": fn, "recv": recv, "args": list(args), "kwargs": {}}
+
+
+def live_contract(d):
+    """Rebuild the live Contract from a JSON-safe field dict."""
+    return oracle_mod.Contract(
+        name=d["name"], module=d["module"], function=d["function"],
+        args=tuple(d["args"]), kwargs=dict(d["kwargs"]),
+        expect_type=None, expect_keys=d["expect_keys"],
+        expect_len=d["expect_len"], expect_not_none=d["expect_not_none"],
+        expect_callable=d["expect_callable"],
+        skip_on_import_error=d["skip_on_import_error"],
+    )
+
+
+# ── branch inventory ───────────────────────────────────────────────────────
+# _resolve_and_project + run_contract/_run_contract_python share this
+# inventory: every phase branch (found / import_error / import-crash /
+# not_found / call-crash / project-crash), the expect_callable
+# short-circuit (callable and non-callable), the plain-attribute path,
+# and each expectation branch (not-none, keys ok/missing, len
+# ok/mismatch). expect_type stays None (JSON law).
+RESOLVE_CASES = [
+    ("rp-found-dict", contract_dict("rp found", "ct_rig", "make_ok")),
+    ("rp-found-none-notnone-fail", contract_dict(
+        "rp none", "ct_rig", "make_none", expect_not_none=True)),
+    ("rp-found-str-attr", contract_dict(
+        "rp attr", "ct_rig", "ATTR_VALUE")),
+    ("rp-call-crash", contract_dict("rp crash", "ct_rig", "raise_on_call")),
+    ("rp-project-crash", contract_dict("rp proj", "ct_rig", "make_bad_len")),
+    ("rp-import-error", contract_dict(
+        "rp import", "definitely_missing_module_xyz", "whatever")),
+    ("rp-import-crash", contract_dict(
+        "rp badimport", "ct_rig_badimport", "whatever")),
+    ("rp-not-found", contract_dict(
+        "rp notfound", "ct_rig", "no_such_fn_xyz")),
+    ("rp-expect-callable", contract_dict(
+        "rp callable", "ct_rig", "make_ok", expect_callable=True)),
+    ("rp-expect-callable-not", contract_dict(
+        "rp notcallable", "ct_rig", "ATTR_VALUE", expect_callable=True)),
+    ("rp-keys-ok", contract_dict(
+        "rp keys", "ct_rig", "make_ok", expect_keys=["ok"])),
+    ("rp-keys-missing", contract_dict(
+        "rp keys miss", "ct_rig", "make_ok", expect_keys=["nope"])),
+    ("rp-len-ok", contract_dict(
+        "rp len", "ct_rig", "make_ok", expect_len=2)),
+    ("rp-len-mismatch", contract_dict(
+        "rp len miss", "ct_rig", "make_ok", expect_len=9)),
+]
+
+# _project: (contract, JSON-safe result) — every projection branch
+# (dict/empty-dict/list/str/None/int-unsized/bool). expect_type None →
+# type_ok is always null here; the True/False branches live in the
+# 58-case ct_truth.json battery.
+PROJECT_CASES = [
+    ("pj-dict", contract_dict("pj", "ct_rig", "make_ok"),
+     {"ok": True, "n": 3}),
+    ("pj-empty-dict", contract_dict("pj", "ct_rig", "make_ok"), {}),
+    ("pj-list", contract_dict("pj", "ct_rig", "make_ok"), [1, 2, 3]),
+    ("pj-str", contract_dict("pj", "ct_rig", "make_ok"), "hello"),
+    ("pj-none", contract_dict("pj", "ct_rig", "make_ok"), None),
+    ("pj-int-unsized", contract_dict("pj", "ct_rig", "make_ok"), 42),
+    ("pj-true", contract_dict("pj", "ct_rig", "make_ok"), True),
+]
+
+
+def freeze_harness_cases():
+    """Cases for the shared harness (JSON-safe in/out, --verify clean)."""
+    entries = [("_ct_core", [], {})]
+    for _tag, c in RESOLVE_CASES:
+        entries.append(recv_case("_resolve_and_project", c))
+    for _tag, c, result in PROJECT_CASES:
+        entries.append(recv_case("_project", c, [result]))
+    entries.append(("capture_signatures", [], {}))
+    return freeze("rune.contracts", entries, oracle_module="rune.contracts")
+
+
+def case_dir(name):
+    d = os.path.join(FREEZE_ROOT, name)
+    os.makedirs(d, exist_ok=True)
+    return d
+
+
+def seed_drifted(d):
+    """Write a saved-sigs file that drifts from the live capture on
+    exactly one key (the alphabetically first)."""
+    cur = oracle_mod.capture_signatures()
+    key = sorted(cur)[0]
+    saved = {key: "(old_sig)", sorted(cur)[1]: cur[sorted(cur)[1]]}
+    oracle_mod.SIGNATURES_FILE = os.path.join(d, "sigs.json")
+    with open(oracle_mod.SIGNATURES_FILE, "w") as f:
+        json.dump(saved, f, indent=2)
+    return key, cur[key]
+
+
+def freeze_extra():
+    """Oracle-derived rows the harness cannot round-trip (dataclass
+    returns, file IO, argv/stdout). Per-case deterministic dirs; no paths
+    in expectations."""
+    extra = {}
+
+    # result_cases: run_contract + _run_contract_python asdicts over the
+    # resolve inventory (pinned core: run_contract == oracle path), plus
+    # the skip=False import-error fail branch.
+    result_cases = []
+    for tag, c in RESOLVE_CASES:
+        for fn_name in ("run_contract", "_run_contract_python"):
+            fn = getattr(oracle_mod, fn_name)
+            got = fn(live_contract(copy.deepcopy(c)))
+            result_cases.append({
+                "name": f"{fn_name}:{tag}",
+                "contract": copy.deepcopy(c),
+                "fn": fn_name,
+                "expected": dataclasses.asdict(copy.deepcopy(got)),
+            })
+    noskip = contract_dict("rp import no-skip",
+                           "definitely_missing_module_xyz", "whatever",
+                           skip_on_import_error=False)
+    for fn_name in ("run_contract", "_run_contract_python"):
+        fn = getattr(oracle_mod, fn_name)
+        got = fn(live_contract(copy.deepcopy(noskip)))
+        result_cases.append({
+            "name": f"{fn_name}:rp-import-noskip-fail",
+            "contract": copy.deepcopy(noskip),
+            "fn": fn_name,
+            "expected": dataclasses.asdict(copy.deepcopy(got)),
+        })
+    extra["result_cases"] = result_cases
+
+    # delegated_cases: the facade's DELEGATED run_contract outputs (Rust
+    # kernel over the resolution). Recorded by calling the loader kernel
+    # directly (the env pin only gates loader.core(), not the FFI). For
+    # every resolve row the kernel verdict is frozen next to the oracle
+    # verdict — one known path divergence: a result whose PROJECTION
+    # raises (rp-project-crash) is a crash on the delegated path (eager
+    # projection) but a PASS on the pure-Python path (lazy checks). Both
+    # behaviors are oracle-shipped; the battery replays both.
+    from rune import contracts_loader  # noqa: E402
+    delegated = []
+    delegated_inventory = [(t, c) for t, c in RESOLVE_CASES] + [
+        ("rp-import-noskip-fail", noskip)]
+    for tag, c in delegated_inventory:
+        live = live_contract(copy.deepcopy(c))
+        phase, err, call, proj = oracle_mod._resolve_and_project(
+            copy.deepcopy(live))
+        out = contracts_loader.verdict({
+            "name": c["name"], "module": c["module"],
+            "function": c["function"], "expect_type": None,
+            "expect_keys": c["expect_keys"], "expect_len": c["expect_len"],
+            "expect_not_none": c["expect_not_none"],
+            "expect_callable": c["expect_callable"],
+            "skip_on_import_error": c["skip_on_import_error"],
+            "phase": phase, "phase_error": err, "callable": call,
+            "projection": proj})
+        delegated.append({
+            "name": f"run_contract:{tag}",
+            "contract": copy.deepcopy(c),
+            "resolution": [phase, err, call, proj],
+            "expected": out,
+        })
+    extra["delegated_cases"] = delegated
+
+    # run_all: the full 42-result order-preserving run, plus the
+    # resolution inputs (spec fields + _resolve_and_project outcome) that
+    # produced them — the battery replays run_all_contracts over these.
+    # Two rows are env-pinned (sibling crates' .so build state decides
+    # their error text); they are frozen as captured in this checkout.
+    run_all = [
+        dataclasses.asdict(copy.deepcopy(r))
+        for r in oracle_mod.run_all_contracts()
+    ]
+    run_all_inputs = []
+    for c in oracle_mod.CONTRACTS:
+        phase, err, call, proj = oracle_mod._resolve_and_project(c)
+        run_all_inputs.append({
+            "name": c.name, "module": c.module, "function": c.function,
+            "expect_type": (c.expect_type.__name__
+                            if c.expect_type is not None else None),
+            "expect_keys": c.expect_keys, "expect_len": c.expect_len,
+            "expect_not_none": c.expect_not_none,
+            "expect_callable": c.expect_callable,
+            "skip_on_import_error": c.skip_on_import_error,
+            "phase": phase, "phase_error": err, "callable": call,
+            "projection": proj,
+        })
+    assert len(run_all) == len(run_all_inputs) == len(oracle_mod.CONTRACTS)
+    extra["run_all"] = run_all
+    extra["run_all_inputs"] = run_all_inputs
+
+    # probes: loader states -> gate bool (module objects are not
+    # JSON-safe; the recorded bool is `result is not None` from live calls).
+    real_loader = oracle_mod.contracts_loader
+
+    class _NoCore:
+        def core(self):
+            return None
+
+    class _RaisingCore:
+        def core(self):
+            raise RuntimeError("rig core failure")
+
+    class _LiveCore:
+        def core(self):
+            return object()
+
+    probes = []
+    for tag, loader in (("probe:loader-none", None),
+                        ("probe:core-none", _NoCore()),
+                        ("probe:core-raises", _RaisingCore()),
+                        ("probe:core-live", _LiveCore())):
+        oracle_mod.contracts_loader = loader
+        try:
+            got = oracle_mod._ct_core()
+            probes.append({"name": tag, "expected": got is not None})
+        finally:
+            oracle_mod.contracts_loader = real_loader
+    extra["probes"] = probes
+
+    # file_cases: save/load through per-case dirs (SIGNATURES_FILE seam).
+    real_file = oracle_mod.SIGNATURES_FILE
+    file_cases = []
+    sig_sets = {
+        "sigset:two": {"m.f": "(x)", "n.g": "(a, b=None)"},
+        "sigset:empty": {},
+        "sigset:unicode": {"m.f": "(s='héllo → world')"},
+    }
+    try:
+        for tag, sigs in sig_sets.items():
+            d = case_dir(tag.replace(":", "-"))
+            oracle_mod.SIGNATURES_FILE = os.path.join(d, "sigs.json")
+            if os.path.exists(oracle_mod.SIGNATURES_FILE):
+                os.remove(oracle_mod.SIGNATURES_FILE)
+            oracle_mod.save_signatures(copy.deepcopy(sigs))
+            with open(oracle_mod.SIGNATURES_FILE) as f:
+                content = f.read()
+            loaded = oracle_mod.load_signatures()
+            file_cases.append({"name": tag, "sigs": copy.deepcopy(sigs),
+                               "content": content,
+                               "expected": copy.deepcopy(loaded)})
+        # load-missing -> {} (no file on disk).
+        d = case_dir("sigset-missing")
+        oracle_mod.SIGNATURES_FILE = os.path.join(d, "absent.json")
+        if os.path.exists(oracle_mod.SIGNATURES_FILE):
+            os.remove(oracle_mod.SIGNATURES_FILE)
+        file_cases.append({"name": "sigset:missing",
+                           "sigs": {},
+                           "content": None,
+                           "expected": oracle_mod.load_signatures()})
+        # load-corrupt -> {} (suppressed exception branch).
+        d = case_dir("sigset-corrupt")
+        oracle_mod.SIGNATURES_FILE = os.path.join(d, "corrupt.json")
+        with open(oracle_mod.SIGNATURES_FILE, "w") as f:
+            f.write("{not json")
+        file_cases.append({"name": "sigset:corrupt",
+                           "sigs": {},
+                           "content": "{not json",
+                           "expected": oracle_mod.load_signatures()})
+    finally:
+        oracle_mod.SIGNATURES_FILE = real_file
+    extra["file_cases"] = file_cases
+
+    # main_cases: argv dispatch (stdout captured, SystemExit caught,
+    # signatures file redirected per case). Call-time prints from live
+    # contract modules interleave into `test` output (deterministic in
+    # this env; Python-side IO — the battery splits at the first result
+    # line). The two -drift variants pre-seed a drifted saved file so the
+    # drift-printing branches are frozen too.
+    real_argv = sys.argv
+    main_cases = []
+    try:
+        for tag, argv in (("main:usage", ["contracts.py"]),
+                          ("main:test", ["contracts.py", "test"]),
+                          ("main:signatures", ["contracts.py", "signatures"]),
+                          ("main:drift", ["contracts.py", "drift"]),
+                          ("main:bogus", ["contracts.py", "bogus"])):
+            d = case_dir(tag.replace(":", "-"))
+            oracle_mod.SIGNATURES_FILE = os.path.join(d, "sigs.json")
+            if os.path.exists(oracle_mod.SIGNATURES_FILE):
+                os.remove(oracle_mod.SIGNATURES_FILE)
+            sys.argv = list(argv)
+            buf = io.StringIO()
+            code = 0
+            with redirect_stdout(buf):
+                try:
+                    oracle_mod.main()
+                except SystemExit as e:
+                    code = e.code if isinstance(e.code, int) else 1
+            main_cases.append({"name": tag, "argv": list(argv),
+                               "exit_code": code,
+                               "expected": buf.getvalue()})
+        for tag, argv in (("main:test-drift", ["contracts.py", "test"]),
+                          ("main:drift-drift", ["contracts.py", "drift"])):
+            d = case_dir(tag.replace(":", "-"))
+            key, cur_sig = seed_drifted(d)
+            sys.argv = list(argv)
+            buf = io.StringIO()
+            code = 0
+            with redirect_stdout(buf):
+                try:
+                    oracle_mod.main()
+                except SystemExit as e:
+                    code = e.code if isinstance(e.code, int) else 1
+            main_cases.append({
+                "name": tag, "argv": list(argv), "exit_code": code,
+                "drift_key": key, "drift_line": f"{key}: (old_sig) → {cur_sig}",
+                "expected": buf.getvalue()})
+    finally:
+        sys.argv = real_argv
+        oracle_mod.SIGNATURES_FILE = real_file
+    extra["main_cases"] = main_cases
+    return extra
+
+
+def canon(x):
+    return json.dumps(x, sort_keys=True, default=str)
+
+
+def replay():
+    """Live parity: facade (delegation active) + kernel (loader direct)
+    against the frozen file."""
+    with open(TRUTH_PATH) as f:
+        doc = json.load(f)
+    failures = 0
+
+    # Warm import order (mirrors freeze: run_all ran first, so every
+    # contract module is in sys.modules before main_cases capture stdout).
+    oracle_mod.run_all_contracts()
+
+    # Facade replay A: harness-case fns through the live oracle must
+    # reproduce the frozen expectations (recv rebuilt by hand — the
+    # harness does this at --verify time). _ct_core is pin/env-dependent
+    # (the probes section covers its branch logic).
+    for c in doc["cases"]:
+        if c["fn"] == "_ct_core":
+            continue
+        fn = getattr(oracle_mod, c["fn"])
+        if "recv" in c:
+            got = fn(live_contract(c["recv"]), *copy.deepcopy(c["args"]))
+            want = c["expected"]["result"]
+        else:
+            got = fn(*copy.deepcopy(c["args"]), **copy.deepcopy(c["kwargs"]))
+            want = c["expected"]
+        if canon(got) != canon(want):
+            failures += 1
+            print(f"HARNESS DIVERGE {c['fn']} {c['recv'].get('name', c['args'])}"
+                  f"\n  got:  {canon(got)[:300]}\n  want: {canon(want)[:300]}")
+
+    # Facade replay B: run_contract/_run_contract_python over the resolve
+    # inventory must reproduce the frozen result_cases (frozen on the
+    # pinned python path — run_contract rows replay with the pin set so
+    # the facade takes the oracle path; the delegated path is replayed
+    # kernel-side below).
+    for rc in doc["extra"]["result_cases"]:
+        fn = getattr(oracle_mod, rc["fn"])
+        if rc["fn"] == "run_contract":
+            os.environ["RUNE_CONTRACTS_CORE"] = "python"
+        try:
+            got = dataclasses.asdict(
+                fn(live_contract(copy.deepcopy(rc["contract"]))))
+        finally:
+            os.environ.pop("RUNE_CONTRACTS_CORE", None)
+        if canon(got) != canon(rc["expected"]):
+            failures += 1
+            print(f"FACADE DIVERGE {rc['name']}"
+                  f"\n  got:  {canon(got)[:300]}"
+                  f"\n  want: {canon(rc['expected'])[:300]}")
+
+    # Facade replay C: save/load through the per-case dirs (recreated
+    # identically) must reproduce content + loaded maps. The missing and
+    # corrupt rows only exercise load (mirroring freeze).
+    real_file = oracle_mod.SIGNATURES_FILE
+    real_argv = sys.argv
+    try:
+        for fc in doc["extra"]["file_cases"]:
+            d = case_dir(fc["name"].replace(":", "-"))
+            path = os.path.join(d, "sigs.json")
+            if fc["name"] == "sigset:missing":
+                path = os.path.join(d, "absent.json")
+            elif fc["name"] == "sigset:corrupt":
+                path = os.path.join(d, "corrupt.json")
+            oracle_mod.SIGNATURES_FILE = path
+            if os.path.exists(path):
+                os.remove(path)
+            if fc["name"] == "sigset:corrupt":
+                with open(path, "w") as f:
+                    f.write(fc["content"])
+            elif fc["name"] != "sigset:missing":
+                oracle_mod.save_signatures(copy.deepcopy(fc["sigs"]))
+                with open(path) as f:
+                    content = f.read()
+                if content != fc["content"]:
+                    failures += 1
+                    print(f"FILE DIVERGE {fc['name']} content\n"
+                          f"  got:  {content[:200]}"
+                          f"\n  want: {fc['content'][:200]}")
+            loaded = oracle_mod.load_signatures()
+            if canon(loaded) != canon(fc["expected"]):
+                failures += 1
+                print(f"FILE DIVERGE {fc['name']} loaded")
+
+        # Facade replay D: argv dispatch (stdout + exit code). The -drift
+        # variants re-seed the drifted saved file first (mirroring freeze).
+        for mc in doc["extra"]["main_cases"]:
+            d = case_dir(mc["name"].replace(":", "-"))
+            oracle_mod.SIGNATURES_FILE = os.path.join(d, "sigs.json")
+            if os.path.exists(oracle_mod.SIGNATURES_FILE):
+                os.remove(oracle_mod.SIGNATURES_FILE)
+            if mc["name"].endswith("-drift"):
+                seed_drifted(d)
+            sys.argv = list(mc["argv"])
+            buf = io.StringIO()
+            code = 0
+            with redirect_stdout(buf):
+                try:
+                    oracle_mod.main()
+                except SystemExit as e:
+                    code = e.code if isinstance(e.code, int) else 1
+            if code != mc["exit_code"] or buf.getvalue() != mc["expected"]:
+                failures += 1
+                print(f"MAIN DIVERGE {mc['name']} exit {code}!= {mc['exit_code']}"
+                      f" or stdout drift\n  got:  {buf.getvalue()[:200]}"
+                      f"\n  want: {mc['expected'][:200]}")
+    finally:
+        oracle_mod.SIGNATURES_FILE = real_file
+        sys.argv = real_argv
+
+    # Kernel replay: ct_verdict over the frozen resolutions must
+    # reproduce the frozen delegated verdicts (the delegated path's
+    # contract — including the eager-projection crash row).
+    from rune import contracts_loader  # noqa: E402
+    if contracts_loader.core() is None:
+        print("REPLAY ABORT: kernel not built")
+        return 2
+    checked = 0
+    for dc in doc["extra"]["delegated_cases"]:
+        c = dc["contract"]
+        phase, err, call, proj = dc["resolution"]
+        out = contracts_loader.verdict({
+            "name": c["name"], "module": c["module"],
+            "function": c["function"], "expect_type": None,
+            "expect_keys": c["expect_keys"], "expect_len": c["expect_len"],
+            "expect_not_none": c["expect_not_none"],
+            "expect_callable": c["expect_callable"],
+            "skip_on_import_error": c["skip_on_import_error"],
+            "phase": phase, "phase_error": err, "callable": call,
+            "projection": proj})
+        checked += 1
+        if canon(out) != canon(dc["expected"]):
+            failures += 1
+            print(f"KERNEL DIVERGE {dc['name']}"
+                  f"\n  got:  {canon(out)[:300]}"
+                  f"\n  want: {canon(dc['expected'])[:300]}")
+    n_results = len(doc["extra"]["result_cases"])
+    n_delegated = len(doc["extra"]["delegated_cases"])
+    print(f"replay: failures={failures} "
+          f"(facade {n_results}, kernel {n_delegated} checked)")
+    return 1 if failures else 0
+
+
+def main():
+    if REPLAY:
+        return replay()
+    cases = freeze_harness_cases()
+    extra = freeze_extra()
+    doc = {"cases": cases, "extra": extra,
+           "note": "Frozen via scripts/gen_contracts_gap_truth.py against "
+                   "the live Python oracle (RUNE_CONTRACTS_CORE=python). "
+                   "cases: shared-harness schema (--verify clean, recv-form "
+                   "Contract receivers). extra: oracle-derived rows "
+                   "(dataclass/file/argv surface). Battery embeds this "
+                   "VERBATIM."}
+    with open(TRUTH_PATH, "w") as f:
+        json.dump(doc, f, indent=1)
+        f.write("\n")
+    total = (len(cases) + len(extra["result_cases"])
+             + len(extra["delegated_cases"]) + 1
+             + len(extra["run_all_inputs"]) + len(extra["probes"])
+             + len(extra["file_cases"]) + len(extra["main_cases"]))
+    print(f"frozen {total} rows -> {os.path.relpath(TRUTH_PATH, REPO)} "
+          f"({len(cases)} harness + {len(extra['result_cases'])} results + "
+          f"{len(extra['delegated_cases'])} delegated + "
+          f"run_all/{len(extra['run_all'])}+{len(extra['run_all_inputs'])} inputs"
+          f" + {len(extra['probes'])} probes + "
+          f"{len(extra['file_cases'])} file + {len(extra['main_cases'])} main)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/open_contract_contracts_gap.py

@@ -0,0 +1,40 @@
+#!/usr/bin/env python3
+"""Open the #2479 dichotomy-wave construction contract for contracts (Rust gap-close).
+Full lifecycle walk in ONE process (shared-store clobber hazard)."""
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from rune.construction_interface import ConstructionInterface
+
+iface = ConstructionInterface(builder="factory-gap-ct")
+
+r = iface.submit_need(
+    "Close the PARTIAL gaps in src/rust/contracts against the "
+    "src/rune/contracts.py oracle (dichotomy wave, issue #2479 "
+    "tranche): name-preserving 1:1 port of the 7 unmatched functions "
+    "(_ct_core, _resolve_and_project, _run_contract_python, "
+    "capture_signatures, save_signatures, load_signatures, main), "
+    "IO behind injectable seams, frozen-truth battery generated against "
+    "the live Python oracle via the shared harness "
+    "scripts/freeze_truth.py (RUNE core pinned to python, deep-copied "
+    "inputs), existing tests green + clippy clean, live parity 100% on "
+    "the new functions, Phase B language-advantage pass.",
+    origin="builder",
+)
+print("submit_need:", json.dumps(r)[:400])
+if not r.get("success"):
+    print("NEED FAILED — aborting")
+    sys.exit(1)
+cid = r["ids"]["contract_id"]
+print("contract_id:", cid)
+
+for phase in ("approved", "building", "verifying"):
+    t = iface.transition(cid, phase)
+    print(f"transition->{phase}:", json.dumps(t)[:300])
+    if not t.get("success"):
+        sys.exit(1)
+
+print("CONTRACT", cid, "IN VERIFYING (close happens at delivery with SHAs)")

src/rune/contracts_loader.py

@@ -7,11 +7,20 @@ FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
     phase, phase_error, callable, projection}) -> ContractResult fields
   * ct_drift({current, saved}) -> drift lines
   * ct_registry() -> the 42 frozen contract specs
+  * ct_gate({available|error}) -> bool (gap-close: _ct_core gate)
+  * ct_resolve({spec..., outcome}) -> [phase, error, callable, projection]
+  * ct_run_python(spec) -> ContractResult fields
+  * ct_run_all([specs]) -> [ContractResult fields]
+  * ct_capture({table, specs}) -> sigs map
+  * ct_save(sigs) -> the signatures file content
+  * ct_load(content|null) -> sigs map
+  * ct_main({argv, run_all, sigs, saved}) -> {exit_code, stdout}
 
 The IO (dynamic module import + call, signature capture via inspect, the
 signatures file) stays in contracts.py; this core owns the verdict branch
-tree and the drift compare. Override the .so with CONTRACTS_RUST_LIB (a
-FILE path).
+tree, the drift compare, and — gap-close tranche — the resolution mapping,
+the signature map/capture loop, the file content/parse kernels and the
+CLI dispatch. Override the .so with CONTRACTS_RUST_LIB (a FILE path).
 """
 from __future__ import annotations
 
@@ -30,6 +39,14 @@ _SYMBOLS = (
     "ct_verdict",
     "ct_drift",
     "ct_registry",
+    "ct_gate",
+    "ct_resolve",
+    "ct_run_python",
+    "ct_run_all",
+    "ct_capture",
+    "ct_save",
+    "ct_load",
+    "ct_main",
 )
 
 

src/rust/contracts/src/gap_battery.rs

@@ -0,0 +1,547 @@
+//! gap_battery — frozen-oracle battery for the contracts gap-close
+//! (dichotomy wave). The truth JSON is the LIVE oracle's output
+//! (scripts/gen_contracts_gap_truth.py; `cases` re-verified byte-identical
+//! by scripts/freeze_truth.py --verify) and is replayed VERBATIM.
+//!
+//! Coverage map (every new kernel, every branch):
+//!   * _ct_core            — probes (loader none / core none / raises / live)
+//!   * _resolve_and_project— every phase branch + expect_callable drop
+//!   * _run_contract_python— all 15 delegated rows + the oracle rows
+//!     (incl. the one documented eager/lazy divergence)
+//!   * run_all_contracts   — the full 42-row order-preserving replay
+//!   * capture_signatures  — key format + skip-on-None over the registry
+//!   * save/load_signatures— byte-parity content, missing/corrupt/unicode
+//!   * contracts_main      — argv dispatch, exact stdout bytes, exit codes
+//!
+//! Documented divergence (oracle-internal, frozen both ways): a result
+//! whose PROJECTION raises (rp-project-crash) is a crash on the delegated
+//! path (eager projection feeds the kernel) but a PASS on the lazy
+//! pure-Python oracle path — the battery pins both behaviors.
+
+use super::{
+    _ct_core, _resolve_and_project, _run_contract_python, capture_signatures,
+    check_signature_drift, contracts_main, load_signatures, registry, run_all_contracts,
+    save_signatures, CoreProbe, InvokeOutcome, Invoker, MainIO, Projection, RunSpec, Signaturer,
+    Verdict,
+};
+use serde_json::Value;
+use std::collections::BTreeMap;
+
+const TRUTH: &str = include_str!("ct_gap_truth.json");
+
+fn truth() -> Value {
+    serde_json::from_str(TRUTH).expect("ct_gap_truth.json parses")
+}
+
+fn spec_from_contract(v: &Value) -> RunSpec {
+    serde_json::from_value(v.clone()).expect("contract spec deserializes")
+}
+
+fn proj_from(v: &Value) -> Option<Projection> {
+    serde_json::from_value(v.clone()).ok()
+}
+
+fn overlay_resolution(spec: &mut RunSpec, resolution: &Value) {
+    let arr = resolution.as_array().expect("resolution is a 4-tuple");
+    spec.phase = arr[0].as_str().unwrap_or_default().to_string();
+    spec.phase_error = arr[1].as_str().unwrap_or_default().to_string();
+    spec.callable = arr[2].as_bool().unwrap_or(false);
+    spec.projection = proj_from(&arr[3]);
+}
+
+fn verdict_from(v: &Value) -> Verdict {
+    serde_json::from_value(v.clone()).expect("verdict deserializes")
+}
+
+fn sigs_from(v: &Value) -> BTreeMap<String, String> {
+    v.as_object()
+        .map(|m| {
+            m.iter()
+                .map(|(k, x)| (k.clone(), x.as_str().unwrap_or_default().to_string()))
+                .collect()
+        })
+        .unwrap_or_default()
+}
+
+// ── seam doubles ────────────────────────────────────────────────────────────
+
+struct Probe(Result<bool, String>);
+
+impl CoreProbe for Probe {
+    fn core_available(&self) -> Result<bool, String> {
+        self.0.clone()
+    }
+}
+
+struct Canned(InvokeOutcome);
+
+impl Invoker for Canned {
+    fn invoke(&self, _spec: &RunSpec) -> InvokeOutcome {
+        self.0.clone()
+    }
+}
+
+/// Rebuild the seam outcome the frozen resolution encodes.
+fn outcome_from_resolution(resolution: &Value) -> InvokeOutcome {
+    let arr = resolution.as_array().expect("resolution is a 4-tuple");
+    let phase = arr[0].as_str().unwrap_or_default();
+    let err = arr[1].as_str().unwrap_or_default().to_string();
+    let callable = arr[2].as_bool().unwrap_or(false);
+    let projection = proj_from(&arr[3]);
+    match phase {
+        "import_error" => InvokeOutcome::ImportError(err),
+        "crash" => InvokeOutcome::CallCrash {
+            error: err,
+            callable,
+        },
+        "not_found" => InvokeOutcome::NotFound,
+        _ => InvokeOutcome::Found {
+            callable,
+            projection,
+        },
+    }
+}
+
+struct CannedSigner(BTreeMap<String, String>);
+
+impl Signaturer for CannedSigner {
+    fn signature_of(&self, module: &str, function: &str) -> Option<String> {
+        self.0.get(&format!("{module}.{function}")).cloned()
+    }
+}
+
+struct CannedMain {
+    run_all: Vec<Verdict>,
+    sigs: BTreeMap<String, String>,
+    saved: BTreeMap<String, String>,
+}
+
+impl MainIO for CannedMain {
+    fn run_all(&self) -> Vec<Verdict> {
+        self.run_all.clone()
+    }
+    fn capture_signatures(&self) -> BTreeMap<String, String> {
+        self.sigs.clone()
+    }
+    fn load_signatures(&self) -> BTreeMap<String, String> {
+        self.saved.clone()
+    }
+    fn save_signatures(&self, _sigs: &BTreeMap<String, String>) {}
+}
+
+// ── batteries ───────────────────────────────────────────────────────────────
+
+/// _ct_core: every loader-state probe outcome maps to the oracle's
+/// None-propagation (only a live core yields true).
+#[test]
+fn test_ct_core_probes() {
+    let t = truth();
+    for p in t["extra"]["probes"].as_array().unwrap() {
+        let expected = p["expected"].as_bool().unwrap();
+        let probe = Probe(match p["name"].as_str().unwrap() {
+            "probe:core-live" => Ok(true),
+            "probe:core-raises" => Err("rig core failure".into()),
+            _ => Ok(false),
+        });
+        assert_eq!(
+            _ct_core(&probe),
+            expected,
+            "gate divergence on {}",
+            p["name"]
+        );
+    }
+}
+
+/// _resolve_and_project: every frozen harness case (recv-form Contract)
+/// replays its (phase, phase_error, callable, projection) tuple through
+/// the mapping kernel.
+#[test]
+fn test_resolve_cases_replay() {
+    let t = truth();
+    let mut n = 0;
+    for c in t["cases"].as_array().unwrap() {
+        if c["fn"].as_str().unwrap() != "_resolve_and_project" {
+            continue;
+        }
+        let spec = spec_from_contract(&c["recv"]);
+        let frozen = &c["expected"]["result"];
+        let got = _resolve_and_project(&spec, &Canned(outcome_from_resolution(frozen)));
+        let arr = frozen.as_array().unwrap();
+        assert_eq!(got.0, arr[0].as_str().unwrap(), "phase on {}", spec.name);
+        assert_eq!(got.1, arr[1].as_str().unwrap(), "error on {}", spec.name);
+        assert_eq!(
+            got.2,
+            arr[2].as_bool().unwrap(),
+            "callable on {}",
+            spec.name
+        );
+        assert_eq!(got.3, proj_from(&arr[3]), "projection on {}", spec.name);
+        n += 1;
+    }
+    assert_eq!(n, 14, "resolve case count");
+}
+
+/// The expect_callable branch DROPS the projection even when the seam
+/// produced one — the oracle never projects a callability-only check.
+#[test]
+fn test_resolve_drops_projection_for_expect_callable() {
+    let spec = RunSpec {
+        name: "drop-probe".into(),
+        module: "m".into(),
+        function: "f".into(),
+        expect_type: None,
+        expect_keys: None,
+        expect_len: None,
+        expect_not_none: false,
+        expect_callable: true,
+        skip_on_import_error: true,
+        phase: String::new(),
+        phase_error: String::new(),
+        callable: false,
+        projection: None,
+    };
+    let projection = Projection {
+        type_name: "dict".into(),
+        none: false,
+        type_ok: None,
+        is_dict: true,
+        keys: Some(vec!["k".into()]),
+        sized: true,
+        len: Some(1),
+    };
+    let got = _resolve_and_project(
+        &spec,
+        &Canned(InvokeOutcome::Found {
+            callable: true,
+            projection: Some(projection),
+        }),
+    );
+    assert_eq!(got.0, "found");
+    assert!(got.3.is_none(), "projection must be dropped");
+}
+
+/// _project: the projection computation stays Python-side (subclass
+/// semantics); the battery pins the SHAPE — every frozen projection
+/// deserializes into the kernel's Projection type with the oracle fields.
+#[test]
+fn test_project_shapes() {
+    let t = truth();
+    let mut n = 0;
+    for c in t["cases"].as_array().unwrap() {
+        if c["fn"].as_str().unwrap() != "_project" {
+            continue;
+        }
+        let p: Projection =
+            serde_json::from_value(c["expected"]["result"].clone()).expect("projection parses");
+        assert!(!p.type_name.is_empty());
+        n += 1;
+    }
+    assert_eq!(n, 7, "project case count");
+}
+
+/// _run_contract_python over the frozen delegated resolutions: all 15
+/// rows (every phase + expectation branch) replay exactly.
+#[test]
+fn test_run_python_delegated_rows() {
+    let t = truth();
+    let rows = t["extra"]["delegated_cases"].as_array().unwrap();
+    assert_eq!(rows.len(), 15);
+    for dc in rows {
+        let mut spec = spec_from_contract(&dc["contract"]);
+        overlay_resolution(&mut spec, &dc["resolution"]);
+        assert_eq!(
+            _run_contract_python(&spec),
+            verdict_from(&dc["expected"]),
+            "delegated divergence on {}",
+            dc["name"]
+        );
+    }
+}
+
+/// _run_contract_python vs the pure-Python oracle rows: identical on
+/// every row except the documented eager/lazy divergence, where the
+/// kernel matches the delegated verdict instead.
+#[test]
+fn test_run_python_oracle_rows() {
+    let t = truth();
+    let delegated: BTreeMap<String, &Value> = t["extra"]["delegated_cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|dc| (dc["contract"]["name"].as_str().unwrap().to_string(), dc))
+        .collect();
+    let mut matched = 0;
+    let mut diverged = 0;
+    for rc in t["extra"]["result_cases"].as_array().unwrap() {
+        if rc["fn"].as_str().unwrap() != "_run_contract_python" {
+            continue;
+        }
+        let cname = rc["contract"]["name"].as_str().unwrap();
+        let dc = delegated.get(cname).expect("resolution available");
+        let mut spec = spec_from_contract(&rc["contract"]);
+        overlay_resolution(&mut spec, &dc["resolution"]);
+        let got = _run_contract_python(&spec);
+        let want = verdict_from(&rc["expected"]);
+        if got == want {
+            matched += 1;
+        } else {
+            // The single documented divergence: eager projection crash.
+            assert_eq!(got, verdict_from(&dc["expected"]), "divergence row {cname}");
+            assert!(
+                cname.contains("rp proj"),
+                "unexpected divergence on {cname}"
+            );
+            diverged += 1;
+        }
+    }
+    assert_eq!(matched + diverged, 15, "oracle row count");
+    assert_eq!(diverged, 1, "exactly the documented divergence");
+}
+
+/// run_all_contracts: the 42-row order-preserving loop over the frozen
+/// resolutions reproduces the live run verdict-for-verdict.
+#[test]
+fn test_run_all_replay() {
+    let t = truth();
+    let inputs = t["extra"]["run_all_inputs"].as_array().unwrap();
+    let want = t["extra"]["run_all"].as_array().unwrap();
+    assert_eq!(inputs.len(), 42);
+    assert_eq!(want.len(), 42);
+    let specs: Vec<RunSpec> = inputs.iter().map(spec_from_contract).collect();
+    // order tie-in with the frozen registry
+    let reg = registry();
+    for (spec, r) in specs.iter().zip(reg.iter()) {
+        assert_eq!(spec.name, r.name, "run_all order vs registry");
+    }
+    let got = run_all_contracts(&specs);
+    for (g, w) in got.iter().zip(want.iter()) {
+        assert_eq!(g, &verdict_from(w), "run_all divergence on {}", w["name"]);
+    }
+}
+
+/// capture_signatures: the "module.function" key format and the
+/// skip-on-None discipline over the registry (the frozen capture has 37
+/// keys for 42 specs — attrs and duplicates yield no signature).
+#[test]
+fn test_capture_signatures_format_and_skip() {
+    let t = truth();
+    let frozen = t["cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .find(|c| c["fn"].as_str().unwrap() == "capture_signatures")
+        .expect("capture case present");
+    let full = sigs_from(&frozen["expected"]);
+    assert_eq!(full.len(), 37, "frozen capture size");
+    // signer drops three keys → capture drops exactly those three
+    let dropped = [
+        "abstraction.pipeline_stats",
+        "cache.get_cache",
+        "goal.check_goal",
+    ];
+    let mut table = full.clone();
+    for k in dropped {
+        table.remove(k);
+    }
+    let got = capture_signatures(&CannedSigner(table), &registry());
+    let mut want = full.clone();
+    for k in dropped {
+        want.remove(k);
+    }
+    assert_eq!(got, want);
+    assert!(got.contains_key("memory_store.connect"));
+    assert!(got.keys().all(|k| k.contains('.')));
+}
+
+/// save/load_signatures: byte-parity with json.dump(indent=2), the
+/// missing-file and corrupt-file degradation, and unicode escaping.
+#[test]
+fn test_file_cases_replay() {
+    let t = truth();
+    let cases = t["extra"]["file_cases"].as_array().unwrap();
+    assert_eq!(cases.len(), 5);
+    let mut saved_checks = 0;
+    for fc in cases {
+        let name = fc["name"].as_str().unwrap();
+        let corrupt = name.contains("corrupt");
+        let sigs = sigs_from(&fc["sigs"]);
+        let expected = sigs_from(&fc["expected"]);
+        match fc["content"].as_str() {
+            Some(content) => {
+                // the corrupt row's file content is rig-written (the
+                // load-degradation fixture), not save output
+                if !corrupt {
+                    assert_eq!(
+                        save_signatures(&sigs),
+                        content,
+                        "save content divergence on {name}"
+                    );
+                }
+                assert_eq!(
+                    load_signatures(Some(content)),
+                    expected,
+                    "load divergence on {name}"
+                );
+                if corrupt {
+                    assert_eq!(load_signatures(Some(content)), BTreeMap::new());
+                }
+                saved_checks += 1;
+            }
+            None => {
+                assert_eq!(load_signatures(None), BTreeMap::new(), "missing file");
+            }
+        }
+    }
+    assert_eq!(saved_checks, 4, "content-bearing rows");
+    // unicode escaping law: non-ASCII escapes as \uXXXX lowercase hex
+    let mut uni = BTreeMap::new();
+    uni.insert("m.f".to_string(), "(s='héllo → world')".to_string());
+    assert_eq!(
+        save_signatures(&uni),
+        "{\n  \"m.f\": \"(s='h\\u00e9llo \\u2192 world')\"\n}"
+    );
+    // empty map law
+    assert_eq!(save_signatures(&BTreeMap::new()), "{}");
+}
+
+/// contracts_main: argv dispatch, exact stdout bytes, exit codes. The
+/// `test` variants carry deterministic call-time prints from live
+/// contract modules (Python-side IO) between the header and the first
+/// result line — the kernel output equals the frozen stdout with that
+/// noise segment removed.
+#[test]
+fn test_main_cases_replay() {
+    let t = truth();
+    let run_all: Vec<Verdict> = t["extra"]["run_all"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(verdict_from)
+        .collect();
+    let capture_case = t["cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .find(|c| c["fn"].as_str().unwrap() == "capture_signatures")
+        .unwrap();
+    let sigs = sigs_from(&capture_case["expected"]);
+    let second_key = sigs
+        .keys()
+        .find(|k| k.as_str() != "abstraction.pipeline_stats")
+        .unwrap()
+        .clone();
+    for mc in t["extra"]["main_cases"].as_array().unwrap() {
+        let name = mc["name"].as_str().unwrap();
+        let argv: Vec<String> = mc["argv"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .map(|v| v.as_str().unwrap().to_string())
+            .collect();
+        let mut saved = BTreeMap::new();
+        if name.ends_with("-drift") {
+            saved.insert(
+                "abstraction.pipeline_stats".to_string(),
+                "(old_sig)".to_string(),
+            );
+            saved.insert(second_key.clone(), sigs[&second_key].clone());
+        }
+        let io = CannedMain {
+            run_all: run_all.clone(),
+            sigs: sigs.clone(),
+            saved,
+        };
+        let (code, stdout) = contracts_main(&argv, &io);
+        assert_eq!(
+            code,
+            mc["exit_code"].as_i64().unwrap() as i32,
+            "exit on {name}"
+        );
+        let frozen = mc["expected"].as_str().unwrap();
+        if name == "main:test" || name == "main:test-drift" {
+            // strip the deterministic call-time print noise: everything
+            // between the header rule and the first result line
+            let idx = frozen
+                .find("  \u{2705}")
+                .or_else(|| frozen.find("  \u{274C}"))
+                .expect("result line present");
+            let header = format!("\u{1F50D} Contract Tests\n{}\n", "=".repeat(50));
+            let want = format!("{header}{}", &frozen[idx..]);
+            assert_eq!(stdout, want, "test-cmd stdout (noise-stripped) on {name}");
+        } else {
+            assert_eq!(stdout, frozen, "stdout on {name}");
+        }
+    }
+}
+
+/// Branch inventory audit: every kernel branch is exercised by the
+/// battery inputs above.
+#[test]
+fn test_gap_battery_covers_every_branch() {
+    let t = truth();
+    let resolutions: Vec<&str> = t["extra"]["delegated_cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|dc| dc["resolution"][0].as_str().unwrap())
+        .collect();
+    for phase in ["import_error", "crash", "not_found", "found"] {
+        assert!(resolutions.contains(&phase), "{phase} branch");
+    }
+    // skip=False import-fail branch
+    assert!(t["extra"]["result_cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .any(|rc| rc["name"].as_str().unwrap().contains("noskip")
+            && rc["expected"]["error"]
+                .as_str()
+                .unwrap()
+                .starts_with("import failed")));
+    // expectation branches (from the delegated rows' verdicts)
+    let errors: Vec<&str> = t["extra"]["delegated_cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .map(|dc| dc["expected"]["error"].as_str().unwrap())
+        .collect();
+    assert!(errors
+        .iter()
+        .any(|e| e.starts_with("skipped (import error")));
+    assert!(errors.contains(&"returned None"));
+    assert!(errors.iter().any(|e| e.starts_with("missing keys")));
+    assert!(errors.iter().any(|e| e.starts_with("expected len")));
+    assert!(errors.iter().any(|e| e.starts_with("RuntimeError")));
+    // drift section frozen in the -drift main rows
+    assert!(t["extra"]["main_cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .any(|mc| mc["name"].as_str().unwrap() == "main:test-drift"
+            && mc["expected"]
+                .as_str()
+                .unwrap()
+                .contains("Signature drift detected (1)")));
+    // drift compare kernel re-check on the seeded drift line
+    let mc = t["extra"]["main_cases"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .find(|mc| mc["name"].as_str().unwrap() == "main:drift-drift")
+        .unwrap();
+    assert!(mc["expected"]
+        .as_str()
+        .unwrap()
+        .contains(mc["drift_line"].as_str().unwrap()));
+    let mut cur = BTreeMap::new();
+    cur.insert("abstraction.pipeline_stats".to_string(), "()".to_string());
+    let mut saved = BTreeMap::new();
+    saved.insert(
+        "abstraction.pipeline_stats".to_string(),
+        "(old_sig)".to_string(),
+    );
+    assert_eq!(
+        check_signature_drift(&cur, &saved),
+        vec!["abstraction.pipeline_stats: (old_sig) → ()".to_string()]
+    );
+}

src/rust/contracts/src/lib.rs

@@ -4,19 +4,33 @@
 //! Port of src/rune/contracts.py's decision surface: the 42-contract
 //! registry (data), the run_contract VERDICT kernel (every branch:
 //! import-error skip/fail, not-found, callable check, not-none, type,
-//! keys, len, crash), and check_signature_drift's compare. The IO stays
-//! Python: dynamic module import + call (the projection is computed
-//! Python-side), signature capture (inspect) and the signatures file.
+//! keys, len, crash), check_signature_drift's compare, and — gap-close
+//! tranche — the 7 audit-named functions (_ct_core,
+//! _resolve_and_project, _run_contract_python, capture_signatures,
+//! save_signatures, load_signatures, main as contracts_main) with their
+//! IO behind injectable seams (CoreProbe / Invoker / Signaturer /
+//! MainIO). The IO stays Python: dynamic module import + call (the
+//! projection is computed Python-side), signature capture (inspect) and
+//! the signatures file.
 //!
 //! FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
 //!   * ct_verdict({spec, phase, phase_error, callable, projection}) -> verdict
 //!   * ct_drift({current, saved}) -> drift lines
 //!   * ct_registry() -> the 42 frozen contract specs
+//!   * ct_gate({available|error}) -> bool
+//!   * ct_resolve({spec..., outcome}) -> [phase, error, callable, projection]
+//!   * ct_run_python(spec) -> verdict
+//!   * ct_run_all([specs]) -> [verdicts]
+//!   * ct_capture({table, specs}) -> sigs map
+//!   * ct_save(sigs) -> file content
+//!   * ct_load(content|null) -> sigs map
+//!   * ct_main({argv, run_all, sigs, saved}) -> {exit_code, stdout}
 //!
 //! NOTE: no free-symbol export (capability_scoring's cs_free_string serves
 //! the whole process — duplicate symbols break the cdylib link).
 
 use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct Contract {
@@ -71,8 +85,8 @@ impl Contract {
 /// the drift lines. Semantics (oracle-probed): changed values ARE drift;
 /// new keys are NOT drift; removed keys are NOT drift.
 pub fn check_signature_drift(
-    current: &std::collections::BTreeMap<String, String>,
-    saved: &std::collections::BTreeMap<String, String>,
+    current: &BTreeMap<String, String>,
+    saved: &BTreeMap<String, String>,
 ) -> Vec<String> {
     let mut drifts = Vec::new();
     for (key, sig) in current {
@@ -131,7 +145,7 @@ fn default_true() -> bool {
     true
 }
 
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 pub struct Projection {
     #[serde(rename = "type")]
     pub type_name: String,
@@ -173,11 +187,119 @@ fn py_str_list(items: &[String]) -> String {
     )
 }
 
-/// run_contract's verdict kernel — the FULL branch tree of the oracle:
-/// import-error skip/fail, crash passthrough, not-found, expect_callable,
-/// then (in oracle order) not-none → type → keys → len, with the exact
-/// error strings.
+/// run_contract's verdict kernel — the FULL branch tree of the oracle
+/// (delegates to `_run_contract_python`, the name-preserving port of the
+/// oracle's own branch tree; one logic, two entries).
 pub fn verdict(spec: &RunSpec) -> Verdict {
+    _run_contract_python(spec)
+}
+
+/// The registry as typed specs (order = oracle CONTRACTS order).
+pub fn registry() -> Vec<RunSpec> {
+    serde_json::from_str::<serde_json::Value>(REGISTRY_JSON)
+        .ok()
+        .and_then(|v| {
+            serde_json::from_value::<Vec<RunSpec>>(v.get("contracts").cloned().unwrap_or_default())
+                .ok()
+        })
+        .unwrap_or_default()
+}
+
+// ═══════════════════════════════════════════════════════════════════════════════
+// GAP-CLOSE (dichotomy wave): the 7 audit-named missing functions, ported
+// name-preserving with IO behind injectable seams. The dynamic import /
+// getattr / call, the inspect-based signature capture and the signatures
+// FILE are the Python side (seams below); everything decidable is here.
+// Frozen truth: ct_gap_truth.json (battery: gap_battery.rs).
+// ═══════════════════════════════════════════════════════════════════════════════
+
+/// The loader-state probe seam for `_ct_core`. The oracle returns the
+/// loader MODULE when its kernel probe succeeds and None on every other
+/// outcome (loader absent, core() None, core() raising) — the portable
+/// logic is that None-propagation, modeled here as a bool gate.
+pub trait CoreProbe {
+    /// Ok(true): the core is live. Ok(false): absent/disabled.
+    /// Err: the probe raised (kernel failure → fallback).
+    fn core_available(&self) -> Result<bool, String>;
+}
+
+/// contracts.py `_ct_core` — the strangler-law probe decision. True only
+/// when the loader module is importable AND its kernel probe returns a
+/// live core; every failure mode (including a raising probe) maps to
+/// False, exactly like the oracle's cascade of `None` returns.
+pub fn _ct_core(probe: &dyn CoreProbe) -> bool {
+    probe.core_available().unwrap_or(false)
+}
+
+/// One outcome of the resolution seam (the IO half of run_contract that
+/// stays Python): the dynamic import, getattr, the call, and the
+/// projection of the result. Which kind of failure happened decides the
+/// phase — that distinction is the portable logic of
+/// `_resolve_and_project`.
+#[derive(Debug, Clone)]
+pub enum InvokeOutcome {
+    /// `__import__` raised ImportError (module genuinely missing).
+    ImportError(String),
+    /// `__import__` raised a non-ImportError (oracle: outer catch → crash).
+    ImportCrash(String),
+    /// getattr returned None (function/attribute not in the module).
+    NotFound,
+    /// The call raised (oracle: outer catch → crash). `callable` is the
+    /// pre-call callable flag the oracle carries through.
+    CallCrash { error: String, callable: bool },
+    /// The projection raised on a hostile result (oracle: crash).
+    ProjectCrash { error: String, callable: bool },
+    /// Import + getattr (+ call for non-expect_callable specs) succeeded.
+    /// `projection` is None when the spec short-circuited (expect_callable)
+    /// — the projection itself is Python-computed (subclass semantics).
+    Found {
+        callable: bool,
+        projection: Option<Projection>,
+    },
+}
+
+/// The resolution seam: performs the Python-side IO for one contract.
+pub trait Invoker {
+    fn invoke(&self, spec: &RunSpec) -> InvokeOutcome;
+}
+
+/// contracts.py `_resolve_and_project` — the resolution half's decision
+/// tree: import-error vs import-crash vs not-found vs call/project crash
+/// vs found, with the oracle's exact (phase, phase_error, callable,
+/// projection|None) tuple shape. The expect_callable branch DROPS the
+/// projection (the oracle never projects when only callability is
+/// checked).
+pub fn _resolve_and_project(
+    spec: &RunSpec,
+    invoker: &dyn Invoker,
+) -> (String, String, bool, Option<Projection>) {
+    match invoker.invoke(spec) {
+        InvokeOutcome::ImportError(e) => ("import_error".into(), e, false, None),
+        InvokeOutcome::ImportCrash(e) => ("crash".into(), e, false, None),
+        InvokeOutcome::NotFound => ("not_found".into(), String::new(), false, None),
+        InvokeOutcome::CallCrash { error, callable }
+        | InvokeOutcome::ProjectCrash { error, callable } => {
+            ("crash".into(), error, callable, None)
+        }
+        InvokeOutcome::Found {
+            callable,
+            projection,
+        } => {
+            if spec.expect_callable {
+                ("found".into(), String::new(), callable, None)
+            } else {
+                ("found".into(), String::new(), callable, projection)
+            }
+        }
+    }
+}
+
+/// contracts.py `_run_contract_python` — the full oracle branch tree over
+/// a resolved spec: import-error skip/fail, crash passthrough, not-found,
+/// expect_callable, then (in oracle order) not-none → type → keys → len,
+/// with the exact error strings. `verdict` (the FFI kernel) delegates
+/// here — one tree, two name-preserving entries.
+pub fn _run_contract_python(spec: &RunSpec) -> Verdict {
     let base = Verdict {
         name: spec.name.clone(),
         passed: false,
@@ -290,15 +412,174 @@ pub fn verdict(spec: &RunSpec) -> Verdict {
     }
 }
 
-/// The registry as typed specs (order = oracle CONTRACTS order).
-pub fn registry() -> Vec<RunSpec> {
-    serde_json::from_str::<serde_json::Value>(REGISTRY_JSON)
-        .ok()
-        .and_then(|v| {
-            serde_json::from_value::<Vec<RunSpec>>(v.get("contracts").cloned().unwrap_or_default())
-                .ok()
-        })
-        .unwrap_or_default()
+/// contracts.py `run_all_contracts` — the order-preserving loop over the
+/// registry (resolution is pre-computed per spec by the caller's seam).
+pub fn run_all_contracts(specs: &[RunSpec]) -> Vec<Verdict> {
+    specs.iter().map(_run_contract_python).collect()
+}
+
+/// The signature-capture seam: `inspect.signature` on a contracted
+/// function is Python-only IO. Exceptions and non-callables surface as
+/// None (the oracle's suppressed-exception skip).
+pub trait Signaturer {
+    fn signature_of(&self, module: &str, function: &str) -> Option<String>;
+}
+
+/// contracts.py `capture_signatures` — the loop over the contract
+/// registry with the oracle's "module.function" key format and the
+/// skip-on-None discipline (import failures and non-callables drop out).
+pub fn capture_signatures(signer: &dyn Signaturer, specs: &[RunSpec]) -> BTreeMap<String, String> {
+    let mut sigs = BTreeMap::new();
+    for spec in specs {
+        if let Some(sig) = signer.signature_of(&spec.module, &spec.function) {
+            sigs.insert(format!("{}.{}", spec.module, spec.function), sig);
+        }
+    }
+    sigs
+}
+
+/// Python `json.dumps` string escaping (ensure_ascii=True): quotes and
+/// backslashes; \n \t \r \b \f shorthands; every other codepoint outside
+/// 0x20..=0x7E as \uXXXX lowercase hex (surrogate pairs above BMP).
+fn py_json_escape(s: &str) -> String {
+    let mut out = String::with_capacity(s.len() + 2);
+    for ch in s.chars() {
+        match ch {
+            '"' => out.push_str("\\\""),
+            '\\' => out.push_str("\\\\"),
+            '\n' => out.push_str("\\n"),
+            '\t' => out.push_str("\\t"),
+            '\r' => out.push_str("\\r"),
+            '\u{8}' => out.push_str("\\b"),
+            '\u{c}' => out.push_str("\\f"),
+            c if (c as u32) >= 0x20 && (c as u32) <= 0x7E => out.push(c),
+            c => {
+                let cp = c as u32;
+                if cp <= 0xFFFF {
+                    out.push_str(&format!("\\u{:04x}", cp));
+                } else {
+                    let v = cp - 0x1_0000;
+                    let hi = 0xD800 + (v >> 10);
+                    let lo = 0xDC00 + (v & 0x3FF);
+                    out.push_str(&format!("\\u{:04x}\\u{:04x}", hi, lo));
+                }
+            }
+        }
+    }
+    out
+}
+
+/// contracts.py `save_signatures` — the file CONTENT half (the makedirs +
+/// write stay at the seam). Byte-parity with `json.dump(sigs, f,
+/// indent=2)`: two-space indent, ": " separators, ensure_ascii escaping.
+/// Key order: the oracle preserves dict insertion order; this port emits
+/// sorted order (BTreeMap) — identical bytes for every sorted-inserted
+/// map, including all frozen rows.
+pub fn save_signatures(sigs: &BTreeMap<String, String>) -> String {
+    if sigs.is_empty() {
+        return "{}".to_string();
+    }
+    let body: Vec<String> = sigs
+        .iter()
+        .map(|(k, v)| format!("  \"{}\": \"{}\"", py_json_escape(k), py_json_escape(v)))
+        .collect();
+    format!("{{\n{}\n}}", body.join(",\n"))
+}
+
+/// contracts.py `load_signatures` — the parse half: missing file (None),
+/// corrupt JSON, and non-string values all degrade to the empty map,
+/// exactly like the oracle's suppressed-exception `return {}`.
+pub fn load_signatures(content: Option<&str>) -> BTreeMap<String, String> {
+    let Some(text) = content else {
+        return BTreeMap::new();
+    };
+    serde_json::from_str::<BTreeMap<String, String>>(text).unwrap_or_default()
+}
+
+/// The argv/stdout/exit-code IO seam for `contracts_main`.
+pub trait MainIO {
+    /// The full registry run (resolution seam applied per spec).
+    fn run_all(&self) -> Vec<Verdict>;
+    /// Live signature capture (Signaturer seam applied).
+    fn capture_signatures(&self) -> BTreeMap<String, String>;
+    /// The saved signatures file (load seam).
+    fn load_signatures(&self) -> BTreeMap<String, String>;
+    /// Persist captured signatures (save seam).
+    fn save_signatures(&self, sigs: &BTreeMap<String, String>);
+}
+
+/// contracts.py `main` — the CLI dispatch kernel: argv branch selection,
+/// the exact output bytes (emoji, = rules, count line, drift section),
+/// and the exit-code rules (usage/bogus/failures exit 1). Returns
+/// (exit_code, stdout) instead of writing/process-exiting — the IO the
+/// oracle does to sys.stdout/sys.exit stays at the seam.
+pub fn contracts_main(argv: &[String], io: &dyn MainIO) -> (i32, String) {
+    let mut out = String::new();
+    if argv.len() < 2 {
+        out.push_str("Usage: contracts.py [test|signatures|drift]\n");
+        return (1, out);
+    }
+    let cmd = argv[1].as_str();
+    match cmd {
+        "test" => {
+            out.push_str("\u{1F50D} Contract Tests\n");
+            out.push_str(&"=".repeat(50));
+            out.push('\n');
+            let results = io.run_all();
+            let passed = results.iter().filter(|r| r.passed).count();
+            let failed = results.len() - passed;
+            for r in &results {
+                if r.passed {
+                    out.push_str(&format!("  \u{2705} {}\n", r.name));
+                } else {
+                    out.push_str(&format!("  \u{274C} {}: {}\n", r.name, r.error));
+                }
+            }
+            out.push_str(&"=".repeat(50));
+            out.push('\n');
+            out.push_str(&format!("Results: {passed} passed, {failed} failed\n"));
+            let current = io.capture_signatures();
+            let saved = io.load_signatures();
+            let drifts = check_signature_drift(&current, &saved);
+            if !drifts.is_empty() {
+                out.push_str(&format!(
+                    "\n\u{26A0}\u{FE0F}  Signature drift detected ({}):\n",
+                    drifts.len()
+                ));
+                for d in &drifts {
+                    out.push_str(&format!("  {d}\n"));
+                }
+            }
+            io.save_signatures(&current);
+            (i32::from(failed > 0), out)
+        }
+        "signatures" => {
+            let sigs = io.capture_signatures();
+            for (key, sig) in &sigs {
+                out.push_str(&format!("{key}{sig}\n"));
+            }
+            io.save_signatures(&sigs);
+            (0, out)
+        }
+        "drift" => {
+            let current = io.capture_signatures();
+            let saved = io.load_signatures();
+            let drifts = check_signature_drift(&current, &saved);
+            if !drifts.is_empty() {
+                out.push_str(&format!("Signature drift detected ({}):\n", drifts.len()));
+                for d in &drifts {
+                    out.push_str(&format!("  {d}\n"));
+                }
+            } else {
+                out.push_str("No signature drift detected.\n");
+            }
+            (0, out)
+        }
+        other => {
+            out.push_str(&format!("Unknown command: {other}\n"));
+            (1, out)
+        }
+    }
 }
 
 // ── FFI ─────────────────────────────────────────────────────────────────────
@@ -338,7 +619,7 @@ pub extern "C" fn ct_verdict(arg: *const std::os::raw::c_char) -> *mut std::os::
 #[no_mangle]
 pub extern "C" fn ct_drift(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
     envelope(read_arg(arg).and_then(|a| {
-        let parse = |k: &str| -> Result<std::collections::BTreeMap<String, String>, String> {
+        let parse = |k: &str| -> Result<BTreeMap<String, String>, String> {
             match a.get(k) {
                 Some(serde_json::Value::Object(m)) => m
                     .iter()
@@ -351,7 +632,7 @@ pub extern "C" fn ct_drift(arg: *const std::os::raw::c_char) -> *mut std::os::ra
                         ))
                     })
                     .collect(),
-                _ => Ok(std::collections::BTreeMap::new()),
+                _ => Ok(BTreeMap::new()),
             }
         };
         let current = parse("current")?;
@@ -367,6 +648,221 @@ pub extern "C" fn ct_registry() -> *mut std::os::raw::c_char {
     envelope(Ok(serde_json::json!(registry())))
 }
 
+// ── FFI (gap-close kernels) ─────────────────────────────────────────────────
+
+struct FfiProbe(Result<bool, String>);
+
+impl CoreProbe for FfiProbe {
+    fn core_available(&self) -> Result<bool, String> {
+        self.0.clone()
+    }
+}
+
+/// ct_gate({"available": bool} | {"error": str}) -> bool — the _ct_core
+/// gate decision over a loader-state probe outcome.
+#[no_mangle]
+pub extern "C" fn ct_gate(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).map(|a| {
+        let probe = if let Some(e) = a.get("error").and_then(|v| v.as_str()) {
+            FfiProbe(Err(e.to_string()))
+        } else {
+            FfiProbe(Ok(a
+                .get("available")
+                .and_then(|v| v.as_bool())
+                .unwrap_or(false)))
+        };
+        serde_json::json!(_ct_core(&probe))
+    }))
+}
+
+/// ct_resolve({spec..., "outcome": {kind, error?, callable?, projection?}})
+/// -> [phase, phase_error, callable, projection|null] — the
+/// _resolve_and_project mapping over one resolution-seam outcome.
+#[no_mangle]
+pub extern "C" fn ct_resolve(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let spec: RunSpec = serde_json::from_value(a.clone()).map_err(|e| e.to_string())?;
+        let out = a.get("outcome").cloned().unwrap_or(serde_json::Value::Null);
+        let kind = out.get("kind").and_then(|v| v.as_str()).unwrap_or("found");
+        let err = |k: &str| {
+            out.get(k)
+                .and_then(|v| v.as_str())
+                .unwrap_or_default()
+                .to_string()
+        };
+        let callable = out
+            .get("callable")
+            .and_then(|v| v.as_bool())
+            .unwrap_or(false);
+        let projection: Option<Projection> = out
+            .get("projection")
+            .cloned()
+            .and_then(|p| serde_json::from_value(p).ok());
+        let outcome = match kind {
+            "import_error" => InvokeOutcome::ImportError(err("error")),
+            "import_crash" => InvokeOutcome::ImportCrash(err("error")),
+            "not_found" => InvokeOutcome::NotFound,
+            "call_crash" => InvokeOutcome::CallCrash {
+                error: err("error"),
+                callable,
+            },
+            "project_crash" => InvokeOutcome::ProjectCrash {
+                error: err("error"),
+                callable,
+            },
+            _ => InvokeOutcome::Found {
+                callable,
+                projection,
+            },
+        };
+        struct FfiInvoker(InvokeOutcome);
+        impl Invoker for FfiInvoker {
+            fn invoke(&self, _spec: &RunSpec) -> InvokeOutcome {
+                self.0.clone()
+            }
+        }
+        let (phase, phase_error, callable, projection) =
+            _resolve_and_project(&spec, &FfiInvoker(outcome));
+        Ok(serde_json::json!([
+            phase,
+            phase_error,
+            callable,
+            projection,
+        ]))
+    }))
+}
+
+/// ct_run_python(spec) -> ContractResult fields — the
+/// `_run_contract_python` tree over a resolved spec.
+#[no_mangle]
+pub extern "C" fn ct_run_python(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let spec: RunSpec = serde_json::from_value(a).map_err(|e| e.to_string())?;
+        serde_json::to_value(_run_contract_python(&spec)).map_err(|e| e.to_string())
+    }))
+}
+
+/// ct_run_all([specs]) -> [ContractResult fields] — the
+/// `run_all_contracts` order-preserving loop.
+#[no_mangle]
+pub extern "C" fn ct_run_all(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let specs: Vec<RunSpec> = serde_json::from_value(a).map_err(|e| e.to_string())?;
+        serde_json::to_value(run_all_contracts(&specs)).map_err(|e| e.to_string())
+    }))
+}
+
+/// ct_capture({"table": {"module.function": sig}, "specs": [specs]}) ->
+/// the capture_signatures map (missing table key = None → skipped).
+#[no_mangle]
+pub extern "C" fn ct_capture(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let table: BTreeMap<String, String> = a
+            .get("table")
+            .and_then(|v| v.as_object())
+            .map(|m| {
+                m.iter()
+                    .map(|(k, v)| Ok((k.clone(), v.as_str().ok_or("non-string sig")?.to_string())))
+                    .collect::<Result<_, String>>()
+            })
+            .unwrap_or_else(|| Ok(BTreeMap::new()))?;
+        let specs: Vec<RunSpec> = serde_json::from_value(
+            a.get("specs")
+                .cloned()
+                .unwrap_or_else(|| serde_json::json!([])),
+        )
+        .map_err(|e| e.to_string())?;
+        struct FfiSigner(BTreeMap<String, String>);
+        impl Signaturer for FfiSigner {
+            fn signature_of(&self, module: &str, function: &str) -> Option<String> {
+                self.0.get(&format!("{module}.{function}")).cloned()
+            }
+        }
+        serde_json::to_value(capture_signatures(&FfiSigner(table), &specs))
+            .map_err(|e| e.to_string())
+    }))
+}
+
+/// ct_save(sigs) -> the save_signatures file content (Python
+/// json.dump(indent=2) bytes).
+#[no_mangle]
+pub extern "C" fn ct_save(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let sigs: BTreeMap<String, String> =
+            serde_json::from_value(a).map_err(|e| e.to_string())?;
+        Ok(serde_json::json!(save_signatures(&sigs)))
+    }))
+}
+
+/// ct_load(content | null) -> the load_signatures map (missing/corrupt →
+/// {}).
+#[no_mangle]
+pub extern "C" fn ct_load(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let content = a.as_str().map(|s| s.to_string());
+        serde_json::to_value(load_signatures(content.as_deref())).map_err(|e| e.to_string())
+    }))
+}
+
+/// ct_main({"argv": [...], "run_all": [verdicts], "sigs": {...},
+/// "saved": {...}}) -> {exit_code, stdout} — the contracts_main dispatch
+/// kernel over canned runner state.
+#[no_mangle]
+pub extern "C" fn ct_main(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let argv: Vec<String> = serde_json::from_value(
+            a.get("argv")
+                .cloned()
+                .unwrap_or_else(|| serde_json::json!([])),
+        )
+        .map_err(|e| e.to_string())?;
+        let run_all: Vec<Verdict> = serde_json::from_value(
+            a.get("run_all")
+                .cloned()
+                .unwrap_or_else(|| serde_json::json!([])),
+        )
+        .map_err(|e| e.to_string())?;
+        let sigs: BTreeMap<String, String> = serde_json::from_value(
+            a.get("sigs")
+                .cloned()
+                .unwrap_or_else(|| serde_json::json!({})),
+        )
+        .map_err(|e| e.to_string())?;
+        let saved: BTreeMap<String, String> = serde_json::from_value(
+            a.get("saved")
+                .cloned()
+                .unwrap_or_else(|| serde_json::json!({})),
+        )
+        .map_err(|e| e.to_string())?;
+        struct FfiMain {
+            run_all: Vec<Verdict>,
+            sigs: BTreeMap<String, String>,
+            saved: BTreeMap<String, String>,
+        }
+        impl MainIO for FfiMain {
+            fn run_all(&self) -> Vec<Verdict> {
+                self.run_all.clone()
+            }
+            fn capture_signatures(&self) -> BTreeMap<String, String> {
+                self.sigs.clone()
+            }
+            fn load_signatures(&self) -> BTreeMap<String, String> {
+                self.saved.clone()
+            }
+            fn save_signatures(&self, _sigs: &BTreeMap<String, String>) {}
+        }
+        let (code, stdout) = contracts_main(
+            &argv,
+            &FfiMain {
+                run_all,
+                sigs,
+                saved,
+            },
+        );
+        Ok(serde_json::json!({"exit_code": code, "stdout": stdout}))
+    }))
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -423,3 +919,7 @@ mod tests {
 #[cfg(test)]
 #[path = "ct_battery.rs"]
 mod ct_battery;
+
+#[cfg(test)]
+#[path = "gap_battery.rs"]
+mod gap_battery;