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

scripts/check_iteration_parity.py

@@ -0,0 +1,301 @@
+#!/usr/bin/env python3
+"""Live parity: iteration Rust kernel vs Python oracle (dichotomy wave).
+
+Three layers, all must be green for parity 100%:
+  1. FROZEN   — the 29 frozen-oracle cases replayed VERBATIM through both
+                the Python floor (oracle sanity) and the Rust kernel.
+  2. LIVE     — oracle (RUNE_ITR_CORE=python) vs facade (kernel auto) on the
+                same inputs; `iteration._now` is pinned so generated_at /
+                recorded_at / invalidation clocks are byte-comparable.
+  3. FUZZ     — 400 seeded random structural models / ledgers / behaviour
+                docs (seed 2479), every public function, both cores.
+"""
+import copy
+import json
+import os
+import random
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "src" / "rust" / "evidence_provenance"))
+
+os.environ.setdefault("RUNE_LEDGER_CORE", "python")
+
+import rune.iteration as it  # noqa: E402
+import rune.behaviour_ledger as bl_mod  # noqa: E402
+from rune.structural_model import StructuralModel  # noqa: E402
+from rune.capability_ledger import CapabilityLedger  # noqa: E402
+from rune.behaviour_ledger import BehaviourLedger  # noqa: E402
+from rune import itr_loader  # noqa: E402
+
+PIN_NOW = "2026-10-12T00:00:00Z"
+it._now = lambda: PIN_NOW
+bl_mod._now = lambda: PIN_NOW
+
+TRUTH = json.loads(
+    (REPO / "src" / "rust" / "iteration" / "src" / "frozen_it_truth.json")
+    .read_text())
+
+KINDS = ["project", "domain", "subsystem", "component", "interface",
+         "state_store", "test_suite", "deployment_surface"]
+REL_TYPES = ["belongs_to", "owns", "depends_on", "calls", "provides",
+             "consumes", "verified_by", "deployed_by"]
+MATURITIES = ["DECLARED", "implemented", "retired", "superseded", "candidate"]
+
+fails = []
+counts = {"frozen": 0, "live": 0, "fuzz": 0}
+
+
+def fail(layer, name, detail):
+    fails.append((layer, name, detail))
+    print(f"FAIL[{layer}] {name}: {detail}")
+
+
+def wrap(doc, cls):
+    o = cls.__new__(cls)
+    o.doc = doc
+    return o
+
+
+# ── fixture factory (oracle-shaped) ─────────────────────────────────────
+
+def make_fixture(rnd, tmp, i):
+    model = StructuralModel(Path(tmp) / f"m{i}", f"p{i}")
+    n_ent = rnd.randint(2, 9)
+    ids = []
+    for j in range(n_ent):
+        kind = rnd.choice(KINDS)
+        eid = model.add(kind, f"e{j}")
+        ids.append(eid)
+        if rnd.random() < 0.4:
+            for e in model.doc["entities"]:
+                if e["id"] == eid:
+                    e.setdefault("identities", {})["source"] = f"src/{j}.py"
+    for _ in range(rnd.randint(0, 12)):
+        a, b = rnd.choice(ids), rnd.choice(ids)
+        if a == b:
+            continue
+        try:
+            model.relate(a, rnd.choice(REL_TYPES), b,
+                         status=rnd.choice(["active", "stale"]))
+        except ValueError:
+            pass
+    model.doc["schema_version"] = 1
+    ledger = CapabilityLedger(Path(tmp) / f"l{i}")
+    for j in range(rnd.randint(1, 4)):
+        ledger.define(
+            capability_id=f"cap{j}", name=f"c{j}", purpose="p",
+            owning_project=f"p{i}",
+            component=rnd.choice(ids),
+            source_locations=[f"src/{rnd.randint(0, n_ent)}.py"])
+    bh = BehaviourLedger(Path(tmp) / f"b{i}")
+    for j in range(rnd.randint(1, 3)):
+        bh.define(behaviour_id=f"b{j}", owning_project=f"p{i}",
+                  intended_behaviour="x",
+                  owner_component=rnd.choice(ids),
+                  dependencies=[rnd.choice(ids)])
+        if rnd.random() < 0.6:
+            bh.record_real_world(f"b{j}", passed=True, detail="ok")
+    return model, ledger, bh
+
+
+def run_all(model, ledger, bh, target):
+    """Call every public fn through whatever core is currently pinned."""
+    out = {}
+    out["context_packet"] = it.context_packet(
+        model, ledger, target, intent="i", basis_summary={"d": 1})
+    out["predictive_impact"] = it.predictive_impact(model, target)
+    out["descriptive_impact"] = it.descriptive_impact(
+        model, target, observed_changed=[target, "surprise:x"],
+        generation="gf")
+    out["post_change_obligations"] = it.post_change_obligations(
+        model, target, behaviours=bh)
+    out["post_change_obligations_nobh"] = it.post_change_obligations(
+        model, target)
+    out["invalidate"] = None
+    return out
+
+
+# ── 1. frozen replay ────────────────────────────────────────────────────
+
+def frozen():
+    for group, cases in TRUTH.items():
+        if group in ("module", "now"):
+            continue
+        for c in cases:
+            counts["frozen"] += 1
+            args = copy.deepcopy(c["args"])
+            fn = c["fn"]
+            now = args.get("now", PIN_NOW)
+            if fn == "context_packet":
+                m = wrap(args["model"], StructuralModel)
+                l = wrap(args.get("ledger") or {"schema_version": 1, "capabilities": {}}, CapabilityLedger)
+                want = c["expect"]
+                got_o = it.context_packet(
+                    m, l, args["target"], intent=args.get("intent", ""),
+                    basis_summary=args.get("basis_summary"),
+                    max_local_files=args.get("max_local_files", 12))
+                got_k = itr_loader.context_packet(
+                    args["model"], args["ledger"], args["target"],
+                    intent=args.get("intent", ""),
+                    basis_summary=args.get("basis_summary"),
+                    max_local_files=args.get("max_local_files", 12),
+                    now=now)
+            elif fn == "predictive_impact":
+                m = wrap(args["model"], StructuralModel)
+                want = c["expect"]
+                got_o = it.predictive_impact(m, args["target"])
+                got_k = itr_loader.predictive_impact(args["model"],
+                                                    args["target"])
+            elif fn == "descriptive_impact":
+                m = wrap(args["model"], StructuralModel)
+                want = c["expect"]
+                got_o = it.descriptive_impact(
+                    m, args["target"],
+                    observed_changed=args["observed_changed"],
+                    generation=args.get("generation", ""))
+                got_k = itr_loader.descriptive_impact(
+                    args["model"], args["target"],
+                    observed_changed=args["observed_changed"],
+                    generation=args.get("generation", ""), now=now)
+            elif fn == "record_impact_delta":
+                want = c["expect"]
+                got_o = json.dumps(args["delta"], sort_keys=True)
+                got_k = itr_loader.record_delta_line(args["delta"])
+            elif fn == "invalidate_behaviours":
+                want = c["expect"]
+                b = wrap(copy.deepcopy(args["behaviours"]), BehaviourLedger)
+                recs_o = it.invalidate_behaviours(
+                    wrap({"relationships": []}, StructuralModel), b,
+                    changed_entities=set(args["changed"]),
+                    generation=args["generation"])
+                got_o = {"doc": b.doc, "recs": recs_o}
+                doc_k, recs_k = itr_loader.invalidate(
+                    args["behaviours"], args["changed"],
+                    args["generation"], now)
+                got_k = {"doc": doc_k, "recs": recs_k}
+            elif fn == "post_change_obligations":
+                m = wrap(args["model"], StructuralModel)
+                want = c["expect"]
+                bh = (wrap(args["behaviours"], BehaviourLedger)
+                      if "behaviours" in args else None)
+                got_o = it.post_change_obligations(
+                    m, args["target"],
+                    changed_entities=set(args["changed"])
+                    if "changed" in args else None,
+                    behaviours=bh)
+                got_k = itr_loader.post_change_obligations(
+                    args["model"], args["target"],
+                    changed=args.get("changed"),
+                    behaviours_doc=args.get("behaviours"))
+            else:  # pragma: no cover
+                continue
+            for tag, got in (("oracle", got_o), ("kernel", got_k)):
+                if json.dumps(got, sort_keys=True) != json.dumps(
+                        want, sort_keys=True):
+                    fail("frozen", f"{fn}/{c['name']}/{tag}",
+                         f"want={json.dumps(want, sort_keys=True)[:300]} "
+                         f"got={json.dumps(got, sort_keys=True)[:300]}")
+
+
+# ── 2+3. live + fuzz ────────────────────────────────────────────────────
+
+def compare_pair(name, o, k):
+    counts["live"] += 1
+    jo, jk = json.dumps(o, sort_keys=True), json.dumps(k, sort_keys=True)
+    if jo != jk:
+        fail("live", name, f"oracle={jo[:300]} kernel={jk[:300]}")
+
+
+def live_and_fuzz():
+    rnd = random.Random(2479)
+    with tempfile.TemporaryDirectory(prefix="it_parity_") as tmp:
+        # live on the frozen fixture inputs
+        for group, cases in TRUTH.items():
+            if group in ("module", "now"):
+                continue
+            for c in cases:
+                args = c["args"]
+                if "model" not in args:
+                    continue
+                m = wrap(args["model"], StructuralModel)
+                l = wrap(args.get("ledger") or {"schema_version": 1, "capabilities": {}}, CapabilityLedger)
+                bh = (wrap(args["behaviours"], BehaviourLedger)
+                      if "behaviours" in args else None)
+                target = args["target"]
+                os.environ["RUNE_ITR_CORE"] = "python"
+                o = run_all(m, l, bh or
+                            wrap({"behaviours": {}}, BehaviourLedger),
+                            target)
+                os.environ["RUNE_ITR_CORE"] = "auto"
+                k = run_all(m, l, bh or
+                            wrap({"behaviours": {}}, BehaviourLedger),
+                            target)
+                for key in o:
+                    compare_pair(f"{c['name']}/{key}", o[key], k[key])
+        # fuzz
+        for i in range(400):
+            model, ledger, bh = make_fixture(rnd, tmp, i)
+            target = rnd.choice(
+                [e["id"] for e in model.doc["entities"]])
+            os.environ["RUNE_ITR_CORE"] = "python"
+            o = run_all(model, ledger, bh, target)
+            os.environ["RUNE_ITR_CORE"] = "auto"
+            k = run_all(model, ledger, bh, target)
+            for key in o:
+                counts["fuzz"] += 1
+                if json.dumps(o[key], sort_keys=True) != json.dumps(
+                        k[key], sort_keys=True):
+                    fail("fuzz", f"fuzz{i}/{key}",
+                         f"oracle={json.dumps(o[key], sort_keys=True)[:300]} "
+                         f"kernel={json.dumps(k[key], sort_keys=True)[:300]}")
+            # record_impact_delta line parity
+            delta = o["descriptive_impact"]
+            counts["fuzz"] += 1
+            if json.dumps(delta, sort_keys=True) != \
+                    itr_loader.record_delta_line(delta):
+                fail("fuzz", f"fuzz{i}/delta_line", "line mismatch")
+            # invalidate parity (deep-copied docs both sides)
+            changed = set(rnd.sample(
+                [e["id"] for e in model.doc["entities"]],
+                k=min(2, len(model.doc["entities"]))))
+            bo = BehaviourLedger.__new__(BehaviourLedger)
+            bo.doc = copy.deepcopy(bh.doc)
+            os.environ["RUNE_ITR_CORE"] = "python"
+            recs_o = it.invalidate_behaviours(model, bo,
+                                              changed_entities=changed,
+                                              generation="gz")
+            os.environ["RUNE_ITR_CORE"] = "auto"
+            bk = BehaviourLedger.__new__(BehaviourLedger)
+            bk.doc = copy.deepcopy(bh.doc)
+            recs_k = it.invalidate_behaviours(model, bk,
+                                              changed_entities=changed,
+                                              generation="gz")
+            counts["fuzz"] += 1
+            if json.dumps(bo.doc, sort_keys=True) != json.dumps(
+                    bk.doc, sort_keys=True) or \
+                    json.dumps(recs_o, sort_keys=True) != \
+                    json.dumps(recs_k, sort_keys=True):
+                fail("fuzz", f"fuzz{i}/invalidate", "doc/recs mismatch")
+
+
+def main():
+    if not itr_loader.available():
+        print("kernel not built — build src/rust/iteration first")
+        return 1
+    os.environ["RUNE_ITR_CORE"] = "python"
+    frozen()
+    live_and_fuzz()
+    os.environ["RUNE_ITR_CORE"] = "auto"
+    total = counts["frozen"] + counts["live"] + counts["fuzz"]
+    print(f"parity {total - len(fails)}/{total} "
+          f"(frozen {counts['frozen']}, live {counts['live']}, "
+          f"fuzz {counts['fuzz']})")
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/gen_iteration_truth.py

@@ -0,0 +1,263 @@
+#!/usr/bin/env python3
+"""Freeze the dichotomy-wave truth corpus for src/rune/iteration.py.
+
+LAW (poisoned-truth): RUNE_IT_CORE=python is pinned at the top — truth is
+generated by the PYTHON ORACLE ONLY, never by the kernel under test.
+RUNE_LEDGER_CORE=python pins the capability ledger to its Python floor.
+The evidence_provenance cdylib stays LIVE: it is part of the oracle stack
+(behaviour_ledger has no Python floor) and is already parity-verified.
+
+Every public function is exercised across ALL branches. Timestamps:
+iteration._now / behaviour_ledger._now are monkeypatched to a FIXED value
+so the frozen corpus is byte-stable; the Rust kernel receives `now` as an
+argument (the same determinism seam the facade uses live).
+"""
+import copy
+import json
+import os
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+
+os.environ["RUNE_IT_CORE"] = "python"          # poisoned-truth law
+os.environ["RUNE_LEDGER_CORE"] = "python"      # capability_ledger floor
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "src" / "rust" / "evidence_provenance"))
+
+import rune.iteration as it  # noqa: E402
+import rune.behaviour_ledger as bl_mod  # noqa: E402
+from rune.structural_model import StructuralModel  # noqa: E402
+from rune.capability_ledger import CapabilityLedger  # noqa: E402
+from rune.behaviour_ledger import BehaviourLedger  # noqa: E402
+
+FIXED_NOW = "2026-10-12T00:00:00Z"
+it._now = lambda: FIXED_NOW
+bl_mod._now = lambda: FIXED_NOW
+
+OUT = REPO / "src" / "rust" / "iteration" / "src" / "frozen_it_truth.json"
+
+cases = []
+
+
+def case(fn, name, args, expect):
+    cases.append({"fn": fn, "name": name,
+                  "args": args, "expect": expect})
+
+
+def build_fixture(tmp):
+    model = StructuralModel(Path(tmp) / "model", "demo")
+    model.add("subsystem", "core")
+    model.add("component", "alpha", parent="subsystem:core",
+              ext={"source": "src/alpha.py"})
+    model.add("component", "beta", parent="subsystem:core")
+    for e in model.doc["entities"]:
+        if e["id"] == "component:beta":
+            e["identities"]["source"] = "src/beta.py"
+    model.add("interface", "api")
+    model.add("interface", "legacy")
+    model.add("state_store", "db")
+    model.add("test_suite", "t1")
+    model.add("deployment_surface", "prod")
+    model.add("component", "gamma", parent="subsystem:core")
+    model.add("component", "delta", parent="subsystem:core")
+    model.relate("component:alpha", "provides", "interface:api")
+    model.relate("component:gamma", "consumes", "interface:api")
+    model.relate("component:delta", "depends_on", "component:alpha",
+                 status="stale")
+    model.relate("component:alpha", "consumes", "interface:legacy")
+    model.relate("component:alpha", "owns", "state_store:db")
+    model.relate("component:alpha", "verified_by", "test_suite:t1")
+    model.relate("component:alpha", "deployed_by", "deployment_surface:prod")
+    model.relate("component:gamma", "provides", "interface:legacy")
+    model.relate("component:gamma", "owns", "state_store:db")
+    model.save()
+
+    ledger = CapabilityLedger(Path(tmp) / "ledger")
+    ledger.define(
+        capability_id="alpha-cap", name="alpha", purpose="alpha purpose",
+        owning_project="demo", component="component:alpha",
+        source_locations=["src/alpha.py", "src/alpha_util.py"],
+        contract="alpha contract", state_touched="state_store:db",
+        dependencies=["interface:legacy"], consumers=["component:gamma"],
+        invariants=["inv-1"], behaviour_specs=["b-alpha"],
+        constraints="no network", generation="g1")
+    ledger.define(
+        capability_id="beta-cap", name="beta", purpose="beta purpose",
+        owning_project="demo", component="component:beta",
+        source_locations=["src/beta.py"])
+    ledger.doc["capabilities"]["capability:beta-cap"]["maturity"] = "retired"
+    ledger.define(
+        capability_id="gamma-cap", name="gamma", purpose="gamma purpose",
+        owning_project="demo", component="component:gamma",
+        source_locations=["src/gamma.py", "src/gamma_more.py"])
+
+    behaviours = BehaviourLedger(Path(tmp) / "behaviours")
+    behaviours.define(
+        behaviour_id="b-alpha", owning_project="demo",
+        intended_behaviour="alpha behaves", owner_component="component:alpha",
+        dependencies=["component:beta"])
+    behaviours.define(
+        behaviour_id="b-gamma", owning_project="demo",
+        intended_behaviour="gamma behaves", owner_component="component:gamma")
+    behaviours.record_real_world("b-alpha", passed=True, detail="ok")
+    return model, ledger, behaviours
+
+
+def main():
+    tmp = tempfile.mkdtemp(prefix="it_truth_")
+    try:
+        model, ledger, behaviours = build_fixture(tmp)
+        model_doc = model.doc
+        ledger_doc = ledger.doc
+        bh_doc = behaviours.doc
+
+        # ── context_packet ────────────────────────────────────────────
+        r = it.context_packet(model, ledger, "component:alpha")
+        case("context_packet", "alpha_full",
+             {"model": model_doc, "ledger": ledger_doc,
+              "target": "component:alpha", "now": FIXED_NOW}, r)
+        r = it.context_packet(model, ledger, "nope:missing")
+        case("context_packet", "unknown_target",
+             {"model": model_doc, "ledger": ledger_doc,
+              "target": "nope:missing", "now": FIXED_NOW}, r)
+        r = it.context_packet(model, ledger, "component:gamma",
+                              intent="ship gamma", basis_summary={"depth": 2})
+        case("context_packet", "gamma_intent_basis",
+             {"model": model_doc, "ledger": ledger_doc,
+              "target": "component:gamma", "intent": "ship gamma",
+              "basis_summary": {"depth": 2}, "now": FIXED_NOW}, r)
+        r = it.context_packet(model, ledger, "component:alpha",
+                              max_local_files=1)
+        case("context_packet", "alpha_truncated",
+             {"model": model_doc, "ledger": ledger_doc,
+              "target": "component:alpha", "max_local_files": 1,
+              "now": FIXED_NOW}, r)
+        r = it.context_packet(model, ledger, "component:beta")
+        case("context_packet", "beta_identities_source",
+             {"model": model_doc, "ledger": ledger_doc,
+              "target": "component:beta", "now": FIXED_NOW}, r)
+        r = it.context_packet(model, ledger, "interface:api")
+        case("context_packet", "interface_target",
+             {"model": model_doc, "ledger": ledger_doc,
+              "target": "interface:api", "now": FIXED_NOW}, r)
+        r = it.context_packet(model, ledger, "project:demo")
+        case("context_packet", "root_project",
+             {"model": model_doc, "ledger": ledger_doc,
+              "target": "project:demo", "now": FIXED_NOW}, r)
+        empty = CapabilityLedger.__new__(CapabilityLedger)
+        empty.doc = {"schema_version": 1, "capabilities": {}}
+        r = it.context_packet(model, empty, "component:alpha")
+        case("context_packet", "empty_ledger",
+             {"model": model_doc, "ledger": empty.doc,
+              "target": "component:alpha", "now": FIXED_NOW}, r)
+
+        # ── predictive_impact ─────────────────────────────────────────
+        for name, target in [("alpha", "component:alpha"),
+                             ("unknown", "nope:missing"),
+                             ("leaf", "component:beta")]:
+            r = it.predictive_impact(model, target)
+            case("predictive_impact", name,
+                 {"model": model_doc, "target": target}, r)
+
+        # ── descriptive_impact ────────────────────────────────────────
+        for name, kwargs in [
+            ("full_overlap", {"observed_changed": ["component:alpha",
+                                                   "component:gamma"]}),
+            ("unexpected", {"observed_changed": ["component:alpha",
+                                                 "surprise:one"]}),
+            ("none_observed", {"observed_changed": []}),
+        ]:
+            r = it.descriptive_impact(model, "component:alpha", **kwargs)
+            case("descriptive_impact", name,
+                 {"model": model_doc, "target": "component:alpha",
+                  "observed_changed": kwargs["observed_changed"],
+                  "now": FIXED_NOW}, r)
+        r = it.descriptive_impact(model, "component:alpha",
+                                  observed_changed=["component:gamma"],
+                                  generation="g7")
+        case("descriptive_impact", "generation_tag",
+             {"model": model_doc, "target": "component:alpha",
+              "observed_changed": ["component:gamma"],
+              "generation": "g7", "now": FIXED_NOW}, r)
+        r = it.descriptive_impact(model, "nope:missing",
+                                  observed_changed=["x"])
+        case("descriptive_impact", "unknown_target",
+             {"model": model_doc, "target": "nope:missing",
+              "observed_changed": ["x"], "now": FIXED_NOW}, r)
+
+        # ── record_impact_delta (exact jsonl line — file IO facade) ───
+        deltas = [
+            {"target": "component:alpha", "generation": "g1",
+             "predicted": ["component:alpha"], "observed": [],
+             "unexpected": [], "unaffected_as_predicted":
+                 ["component:alpha"], "recorded_at": FIXED_NOW},
+            {"b": 1, "a": [1.5, 2], "c": None, "s": "é\nx"},
+            {"nested": {"z": True, "y": 0.1}},
+        ]
+        for i, delta in enumerate(deltas):
+            d = Path(tmp) / f"deltas{i}"
+            d.mkdir()
+            it.record_impact_delta(d, delta)
+            line = (d / "impact_deltas.jsonl").read_text()
+            assert line.endswith("\n") and line.count("\n") == 1
+            case("record_impact_delta", f"line{i}", {"delta": delta},
+                 line[:-1])
+
+        # ── invalidate_behaviours ─────────────────────────────────────
+        for name, changed in [
+            ("hit_deps_and_owner", {"component:alpha", "component:beta"}),
+            ("owner_only", {"component:gamma"}),
+            ("no_intersection", {"component:delta"}),
+        ]:
+            b = BehaviourLedger.__new__(BehaviourLedger)
+            b.doc = copy.deepcopy(bh_doc)
+            recs = it.invalidate_behaviours(
+                model, b, changed_entities=changed, generation="g9")
+            case("invalidate_behaviours", name,
+                 {"behaviours": bh_doc, "changed": sorted(changed),
+                  "generation": "g9", "now": FIXED_NOW},
+                 {"doc": b.doc, "recs": recs})
+
+        # ── post_change_obligations ───────────────────────────────────
+        for name, target, changed in [
+            ("alpha_default", "component:alpha", None),
+            ("alpha_two_changed", "component:alpha",
+             ["component:alpha", "component:gamma"]),
+            ("beta_leaf", "component:beta", None),
+            ("gamma_consumes_iface", "component:gamma", None),
+            ("delta_stale_dep", "component:delta", None),
+            ("unknown_in_changed", "component:beta",
+             ["component:beta", "ghost:x"]),
+        ]:
+            if changed is None:
+                r = it.post_change_obligations(model, target)
+                args = {"model": model_doc, "target": target}
+            else:
+                r = it.post_change_obligations(
+                    model, target, changed_entities=set(changed))
+                args = {"model": model_doc, "target": target,
+                        "changed": changed}
+            case("post_change_obligations", name, args, r)
+        r = it.post_change_obligations(model, "component:alpha",
+                                       behaviours=behaviours)
+        case("post_change_obligations", "alpha_with_behaviours",
+             {"model": model_doc, "target": "component:alpha",
+              "behaviours": bh_doc}, r)
+
+        OUT.parent.mkdir(parents=True, exist_ok=True)
+        groups = {}
+        for c in cases:
+            groups.setdefault(f"{c['fn']}_cases", []).append(c)
+        OUT.write_text(json.dumps(
+            {"module": "iteration", "now": FIXED_NOW, **groups},
+            indent=1, sort_keys=True) + "\n")
+        print(f"frozen {len(cases)} cases -> {OUT}")
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()

scripts/open_contract_batch_a.py

@@ -0,0 +1,43 @@
+#!/usr/bin/env python3
+"""Open the #2479 dichotomy-wave construction contract for the BATCH-A
+modules (iteration, subconscious_pass, rune_profiler, subsystem_health).
+Walks draft→approved→building in ONE process (shared-store clobber hazard).
+
+Usage: open_contract_batch_a.py <module> [contract_id]
+"""
+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
+
+module = sys.argv[1]
+sym = {"iteration": "IT", "subconscious_pass": "SP",
+       "rune_profiler": "RP", "subsystem_health": "SH"}[module]
+
+iface = ConstructionInterface(builder="factory-batch-a")
+
+cid = sys.argv[2] if len(sys.argv) > 2 else None
+r = iface.submit_need(
+    f"Port src/rune/{module}.py to FULL Rust (dichotomy factory, issue "
+    f"#2479 tranche): doc-envelope FFI, ctypes loader, facade delegation "
+    f"with RUNE_{sym}_CORE=python pin, frozen-truth battery, live parity "
+    f"100%, Phase B language-advantage pass.",
+    origin="builder",
+    contract_id=cid,
+)
+print("submit_need:", json.dumps(r)[:400])
+if not r.get("success"):
+    print("NEED FAILED — aborting")
+    sys.exit(1)
+c = r["ids"]["contract_id"]
+print("contract_id:", c)
+
+for phase in ("approved", "building"):
+    t = iface.transition(c, phase)
+    print(f"transition->{phase}:", json.dumps(t)[:300])
+    if not t.get("success"):
+        sys.exit(1)
+print("OPENED", c)

src/rune/iteration.py

@@ -23,6 +23,24 @@ from rune.capability_ledger import CapabilityLedger
 from rune.behaviour_ledger import BehaviourLedger
 
 
+def _kernel():
+    """The Rust iteration core (itr_loader) when RUNE_ITR_CORE allows it,
+    else None — every public function delegates there first and falls
+    back to the Python body below (same process, same semantics)."""
+    import os
+
+    if os.environ.get("RUNE_ITR_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import itr_loader
+
+        if itr_loader.available():
+            return itr_loader
+    except Exception:  # noqa: BLE001 — kernel failure → Python floor
+        return None
+    return None
+
+
 def _now() -> str:
     return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
 
@@ -36,6 +54,15 @@ def context_packet(model, ledger: CapabilityLedger, target: str, *,
     Bounded and proportional: the whole graph is NEVER dumped. Local =
     the target's own ledger entry; connected = 1-hop structural
     neighbours with their ledgers on demand."""
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.context_packet(
+                model.doc, ledger.doc, target, intent=intent,
+                basis_summary=basis_summary,
+                max_local_files=max_local_files, now=_now())
+        except Exception:  # noqa: BLE001 — kernel failure → Python floor
+            pass
     entity = model.get(target)
     if entity is None:
         return {"error": f"unknown structural target {target!r}"}
@@ -88,6 +115,12 @@ def context_packet(model, ledger: CapabilityLedger, target: str, *,
 
 def predictive_impact(model, target: str) -> dict:
     """What MAY be affected BEFORE the change (structural closure)."""
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.predictive_impact(model.doc, target)
+        except Exception:  # noqa: BLE001
+            pass
     return model.impact_closure(target)
 
 
@@ -96,6 +129,14 @@ def descriptive_impact(model, target: str, *,
                        generation: str = "") -> dict:
     """What was MATERIALLY affected AFTER the change — from actual
     evidence (diff/observations), never assumed from the graph alone."""
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.descriptive_impact(
+                model.doc, target, observed_changed=observed_changed,
+                generation=generation, now=_now())
+        except Exception:  # noqa: BLE001
+            pass
     predicted = model.impact_closure(target)
     predicted_scope = set(predicted.get("mutation_scope", []))
     observed = set(observed_changed)
@@ -113,10 +154,20 @@ def descriptive_impact(model, target: str, *,
 def record_impact_delta(project_store: Path, delta: dict) -> Path:
     """Persist predicted-vs-observed deltas — they compound into better
     future impact prediction (rule 19H)."""
+    k = _kernel()
+    if k is not None:
+        try:
+            line = k.record_delta_line(delta) + "\n"
+        except Exception:  # noqa: BLE001
+            line = None
+    else:
+        line = None
+    if line is None:
+        import json
+        line = json.dumps(delta, sort_keys=True) + "\n"
     p = Path(project_store) / "impact_deltas.jsonl"
-    import json
     with p.open("a") as f:
-        f.write(json.dumps(delta, sort_keys=True) + "\n")
+        f.write(line)
     return p
 
 
@@ -128,6 +179,15 @@ def invalidate_behaviours(model, behaviours: BehaviourLedger, *,
     """Selective invalidation driven by STRUCTURAL truth: behaviours
     whose dependencies/owner intersect the changed closure reopen;
     everything else keeps its confidence (rule 19I)."""
+    k = _kernel()
+    if k is not None:
+        try:
+            behaviours.doc, recs = k.invalidate(
+                behaviours.doc, sorted(changed_entities),
+                generation, _now())
+            return recs
+        except Exception:  # noqa: BLE001
+            pass
     return behaviours.invalidate_for_change(
         changed_entities=changed_entities, generation=generation)
 
@@ -140,6 +200,17 @@ def post_change_obligations(model, target: str, *,
                             ) -> list[dict]:
     """Machine-derived follow-up obligations from STRUCTURAL truth
     (rule 19K) — never an improvised generic checklist."""
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.post_change_obligations(
+                model.doc, target,
+                changed=sorted(changed_entities)
+                if changed_entities else None,
+                behaviours_doc=behaviours.doc if behaviours is not None
+                else None)
+        except Exception:  # noqa: BLE001
+            pass
     changed = changed_entities or {target}
     obligations: list[dict] = []
     seen: set[tuple] = set()

src/rune/itr_loader.py

@@ -0,0 +1,131 @@
+#!/usr/bin/env python3
+"""ctypes loader for the Rust iteration core (prefix ITR — the it_* wire prefix belongs to inference_trace) (dichotomy wave, #2479
+tranche). Mirrors broker_loader.py's contract:
+
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * itr_context_packet({model, ledger, target, intent, basis_summary,
+                       max_local_files, now})              -> packet dict
+  * itr_predictive_impact({model, target})                 -> closure dict
+  * itr_descriptive_impact({model, target, observed_changed,
+                           generation, now})              -> delta dict
+  * itr_record_delta_line({delta})                         -> line body
+  * itr_invalidate({behaviours, changed, generation, now}) -> {doc, recs}
+  * itr_post_change_obligations({model, target, changed, behaviours})
+                                                          -> [obligations]
+
+Model/ledger/behaviours ride as their JSON docs; timestamps are facade
+ARGUMENTS (now). Free: capability_scoring's cs_free_string (shared
+process). Override the .so with ITR_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" / "iteration"
+_LIB_PATH = Path(
+    os.environ.get("ITR_RUST_LIB", _CRATE_DIR / "target" / "release" / "libiteration.so")
+)
+
+_ERR_TYPES = {
+    "ValueError": ValueError,
+    "KeyError": KeyError,
+    "AttributeError": AttributeError,
+    "TypeError": TypeError,
+}
+
+_lib = None
+
+
+class IterationCoreUnavailable(FileNotFoundError):
+    """The Rust iteration core (prefix ITR — the it_* wire prefix belongs to inference_trace) is not built. Callers fall back to Python."""
+
+
+def available() -> bool:
+    return _LIB_PATH.is_file()
+
+
+def _load() -> ctypes.CDLL:
+    global _lib
+    if _lib is not None:
+        return _lib
+    if not _LIB_PATH.is_file():
+        raise IterationCoreUnavailable(str(_LIB_PATH))
+    lib = ctypes.CDLL(str(_LIB_PATH))
+    for sym in ("itr_context_packet", "itr_predictive_impact",
+                "itr_descriptive_impact", "itr_record_delta_line",
+                "itr_invalidate", "itr_post_change_obligations"):
+        getattr(lib, sym).argtypes = [ctypes.c_void_p]
+        getattr(lib, sym).restype = ctypes.c_void_p
+    try:
+        lib.cs_free_string
+    except AttributeError:
+        scoring = ctypes.CDLL(str(
+            _REPO / "src" / "rust" / "capability_scoring" / "target" / "release" / "libcapability_scoring.so"
+        ))
+        scoring.cs_free_string.argtypes = [ctypes.c_void_p]
+        lib._free = scoring.cs_free_string
+    _lib = lib
+    return lib
+
+
+def _call(sym: str, payload: dict):
+    lib = _load()
+    data = json.dumps(payload).encode()
+    buf = ctypes.create_string_buffer(data)
+    ptr = getattr(lib, sym)(ctypes.cast(buf, ctypes.c_void_p))
+    if not ptr:
+        raise RuntimeError(f"{sym} returned null")
+    out = ctypes.string_at(ptr).decode()
+    free = getattr(lib, "_free", None)
+    if free is not None:
+        free(ptr)
+    doc = json.loads(out)
+    if not doc.get("ok"):
+        etype = doc.get("type", "RuntimeError")
+        raise _ERR_TYPES.get(etype, RuntimeError)(doc.get("err", f"{sym} failed"))
+    return doc["value"]
+
+
+def context_packet(model_doc: dict, ledger_doc: dict, target: str, *,
+                   intent: str = "", basis_summary=None,
+                   max_local_files: int = 12, now: str) -> dict:
+    return _call("itr_context_packet", {
+        "model": model_doc, "ledger": ledger_doc, "target": target,
+        "intent": intent, "basis_summary": basis_summary,
+        "max_local_files": max_local_files, "now": now})
+
+
+def predictive_impact(model_doc: dict, target: str) -> dict:
+    return _call("itr_predictive_impact", {"model": model_doc, "target": target})
+
+
+def descriptive_impact(model_doc: dict, target: str, *,
+                       observed_changed: list, generation: str = "",
+                       now: str) -> dict:
+    return _call("itr_descriptive_impact", {
+        "model": model_doc, "target": target,
+        "observed_changed": observed_changed, "generation": generation,
+        "now": now})
+
+
+def record_delta_line(delta: dict) -> str:
+    return _call("itr_record_delta_line", {"delta": delta})
+
+
+def invalidate(behaviours_doc: dict, changed: list, generation: str,
+               now: str) -> tuple[dict, list]:
+    out = _call("itr_invalidate", {"behaviours": behaviours_doc,
+                                  "changed": changed,
+                                  "generation": generation, "now": now})
+    return out["doc"], out["recs"]
+
+
+def post_change_obligations(model_doc: dict, target: str, *,
+                            changed=None, behaviours_doc=None) -> list:
+    return _call("itr_post_change_obligations", {
+        "model": model_doc, "target": target, "changed": changed,
+        "behaviours": behaviours_doc})

src/rust/iteration/Cargo.lock

@@ -3,9 +3,36 @@
 version = 4
 
 [[package]]
-name = "iteration"
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
+[[package]]
+name = "evidence_provenance"
 version = "0.1.0"
+
+[[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 = "iteration"
+version = "0.2.0"
+dependencies = [
+ "evidence_provenance",
  "serde_json",
 ]
 
@@ -74,6 +101,7 @@ version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
+ "indexmap",
  "itoa",
  "memchr",
  "serde",

src/rust/iteration/Cargo.toml

@@ -1,10 +1,12 @@
 [package]
 name = "iteration"
-version = "0.1.0"
+version = "0.2.0"
 edition = "2021"
+description = "FULL parity port of src/rune/iteration.py (dichotomy wave, #2479 tranche)"
 
 [lib]
 crate-type = ["rlib", "cdylib", "staticlib"]
 
 [dependencies]
-serde_json = "1"
+serde_json = { version = "1", features = ["preserve_order", "float_roundtrip"] }
+evidence_provenance = { path = "../evidence_provenance" }

src/rust/iteration/src/branch_battery.rs

@@ -0,0 +1,219 @@
+//! branch_battery — replay the FROZEN oracle truth VERBATIM
+//! (frozen_it_truth.json, generated by scripts/gen_iteration_truth.py
+//! against the LIVE Python oracle with RUNE_IT_CORE=python and
+//! RUNE_LEDGER_CORE=python pinned).
+//!
+//! Every case: feed the recorded input doc through the kernel's public
+//! functions and compare the result against the oracle's recorded
+//! expectation — parsed as Python values (dict/list/str/num/bool/None),
+//! so key order is insensitive but every value must match exactly.
+
+use crate::{
+    context_packet, descriptive_impact, invalidate_behaviours,
+    post_change_obligations, predictive_impact, record_delta_line,
+};
+use evidence_provenance::pyjson::{self, Json};
+
+const TRUTH: &str = include_str!("frozen_it_truth.json");
+
+fn truth() -> Json {
+    pyjson::parse(TRUTH).expect("frozen truth parses")
+}
+
+fn cases(t: &Json, group: &str) -> Vec<Json> {
+    match t.get(group) {
+        Some(Json::Arr(items)) => items.clone(),
+        _ => panic!("truth group {group} missing"),
+    }
+}
+
+fn at<'a>(v: &'a Json, key: &str) -> Option<&'a Json> {
+    match v {
+        Json::Obj(pairs) => pairs.iter().find(|(k, _)| k == key).map(|(_, x)| x),
+        _ => None,
+    }
+}
+
+fn atv<'a>(v: &'a Json, path: &[&str]) -> Option<&'a Json> {
+    let mut cur = v;
+    for k in path {
+        cur = at(cur, k)?;
+    }
+    Some(cur)
+}
+
+fn jstr(v: Option<&Json>) -> String {
+    match v {
+        Some(Json::Str(s)) => s.clone(),
+        _ => String::new(),
+    }
+}
+
+fn jlist(v: Option<&Json>) -> Vec<String> {
+    match v {
+        Some(Json::Arr(items)) => items
+            .iter()
+            .filter_map(|x| match x {
+                Json::Str(s) => Some(s.clone()),
+                _ => None,
+            })
+            .collect(),
+        _ => Vec::new(),
+    }
+}
+
+fn jnum(v: Option<&Json>, default: f64) -> f64 {
+    match v {
+        Some(Json::Num(n)) => *n,
+        _ => default,
+    }
+}
+
+/// Python-value equality on Json (num compares by f64).
+fn peq(a: &Json, b: &Json) -> bool {
+    match (a, b) {
+        (Json::Null, Json::Null) => true,
+        (Json::Bool(x), Json::Bool(y)) => x == y,
+        (Json::Num(x), Json::Num(y)) => x == y,
+        (Json::Str(x), Json::Str(y)) => x == y,
+        (Json::Arr(x), Json::Arr(y)) => {
+            x.len() == y.len() && x.iter().zip(y).all(|(a, b)| peq(a, b))
+        }
+        (Json::Obj(x), Json::Obj(y)) => {
+            x.len() == y.len()
+                && x.iter().all(|(k, v)| {
+                    y.iter()
+                        .find(|(k2, _)| k2 == k)
+                        .map(|(_, v2)| peq(v, v2))
+                        .unwrap_or(false)
+                })
+        }
+        _ => false,
+    }
+}
+
+fn assert_eq(case_name: &str, want: &Json, got: &Json) {
+    assert!(
+        peq(want, got),
+        "case {case_name} diverged:\n want={}\n  got={}",
+        pyjson::dumps(want),
+        pyjson::dumps(got),
+    );
+}
+
+#[test]
+fn replay_context_packet() {
+    let t = truth();
+    for c in cases(&t, "context_packet_cases") {
+        let got = context_packet(
+            atv(&c, &["args", "model"]).expect("model"),
+            atv(&c, &["args", "ledger"]).expect("ledger"),
+            &jstr(atv(&c, &["args", "target"])),
+            &jstr(atv(&c, &["args", "intent"])),
+            atv(&c, &["args", "basis_summary"]).cloned().unwrap_or(Json::Null),
+            jnum(atv(&c, &["args", "max_local_files"]), 12.0) as usize,
+            &jstr(atv(&c, &["args", "now"])),
+        );
+        assert_eq(
+            &format!("context_packet/{}", jstr(at(&c, "name"))),
+            at(&c, "expect").expect("expect"),
+            &got,
+        );
+    }
+}
+
+#[test]
+fn replay_predictive_impact() {
+    let t = truth();
+    for c in cases(&t, "predictive_impact_cases") {
+        let got = predictive_impact(
+            atv(&c, &["args", "model"]).expect("model"),
+            &jstr(atv(&c, &["args", "target"])),
+        );
+        assert_eq(
+            &format!("predictive_impact/{}", jstr(at(&c, "name"))),
+            at(&c, "expect").expect("expect"),
+            &got,
+        );
+    }
+}
+
+#[test]
+fn replay_descriptive_impact() {
+    let t = truth();
+    for c in cases(&t, "descriptive_impact_cases") {
+        let got = descriptive_impact(
+            atv(&c, &["args", "model"]).expect("model"),
+            &jstr(atv(&c, &["args", "target"])),
+            &jlist(atv(&c, &["args", "observed_changed"])),
+            &jstr(atv(&c, &["args", "generation"])),
+            &jstr(atv(&c, &["args", "now"])),
+        );
+        assert_eq(
+            &format!("descriptive_impact/{}", jstr(at(&c, "name"))),
+            at(&c, "expect").expect("expect"),
+            &got,
+        );
+    }
+}
+
+#[test]
+fn replay_record_delta_line() {
+    let t = truth();
+    for c in cases(&t, "record_impact_delta_cases") {
+        let got = record_delta_line(atv(&c, &["args", "delta"]).expect("delta"));
+        assert_eq(
+            &format!("record_impact_delta/{}", jstr(at(&c, "name"))),
+            at(&c, "expect").expect("expect"),
+            &Json::Str(got),
+        );
+    }
+}
+
+#[test]
+fn replay_invalidate_behaviours() {
+    let t = truth();
+    for c in cases(&t, "invalidate_behaviours_cases") {
+        let mut bh = atv(&c, &["args", "behaviours"])
+            .cloned()
+            .expect("behaviours");
+        let recs = invalidate_behaviours(
+            &mut bh,
+            &jlist(atv(&c, &["args", "changed"])),
+            &jstr(atv(&c, &["args", "generation"])),
+            &jstr(atv(&c, &["args", "now"])),
+        )
+        .expect("invalidate ok");
+        let name = jstr(at(&c, "name"));
+        assert_eq(
+            &format!("invalidate_behaviours/{name}/doc"),
+            atv(&c, &["expect", "doc"]).expect("doc"),
+            &bh,
+        );
+        assert_eq(
+            &format!("invalidate_behaviours/{name}/recs"),
+            atv(&c, &["expect", "recs"]).expect("recs"),
+            &recs,
+        );
+    }
+}
+
+#[test]
+fn replay_post_change_obligations() {
+    let t = truth();
+    for c in cases(&t, "post_change_obligations_cases") {
+        let changed = atv(&c, &["args", "changed"]).map(|v| jlist(Some(v)));
+        let behaviours = atv(&c, &["args", "behaviours"]).cloned();
+        let got = post_change_obligations(
+            atv(&c, &["args", "model"]).expect("model"),
+            &jstr(atv(&c, &["args", "target"])),
+            changed.as_deref(),
+            behaviours.as_ref(),
+        );
+        assert_eq(
+            &format!("post_change_obligations/{}", jstr(at(&c, "name"))),
+            at(&c, "expect").expect("expect"),
+            &got,
+        );
+    }
+}

src/rust/iteration/src/lib.rs

@@ -1,14 +1,848 @@
-//! iteration — post-change iteration tracking (W6).
-pub fn record_impact_delta(
-    _delta_path: &str,
-    contract_id: &str,
-    scope_files: &[String],
-) -> serde_json::Value {
-    serde_json::json!({"contract_id": contract_id, "scope_files": scope_files, "recorded": true})
+//! iteration — T19 §19E/19H/19I/19K: the canonical dependency-aware
+//! iteration engine.
+//!
+//! FULL parity port of src/rune/iteration.py (dichotomy wave, #2479
+//! tranche). The oracle consumes live StructuralModel / CapabilityLedger /
+//! BehaviourLedger objects; over FFI they ride as their JSON docs and the
+//! model queries (children/beneath/consumers/impact_closure/...) are
+//! reimplemented from the Python floor semantics (the sm_loader kernel is
+//! already parity-verified against the same floor).
+//!
+//! FFI (doc-envelope law): every call takes one JSON doc and returns
+//! {"ok": true, "value": ...} | {"ok": false, "err": ..., "type": ...}.
+//!   * itr_context_packet({model, ledger, target, intent?, basis_summary?,
+//!     max_local_files?, now}) -> packet dict
+//!   * itr_predictive_impact({model, target}) -> closure dict
+//!   * itr_descriptive_impact({model, target, observed_changed,
+//!     generation?, now}) -> delta dict
+//!   * itr_record_delta_line({delta}) -> jsonl line
+//!   * itr_invalidate({behaviours, changed, generation?, now}) -> {doc, recs}
+//!   * itr_post_change_obligations({model, target, changed?, behaviours?})
+//!     -> [obligations]
+//!
+//! Determinism seams: `generated_at` / `recorded_at` / the invalidation
+//! clock are facade ARGUMENTS (`now`) — the frozen-truth battery replays
+//! them verbatim. The impact_deltas.jsonl FILE append stays in the facade;
+//! the kernel owns the exact line bytes (json.dumps sort_keys parity via
+//! pyjson::dumps_sorted).
+//!
+//! invalidate_behaviours delegates to the evidence_provenance kernel's
+//! behaviour::invalidate_for_change — the SAME Rust function the oracle's
+//! ep_loader routes to (rule C: no second invalidation authority).
+//!
+//! NOTE: no itr_free_string export — callers free through capability_
+//! scoring's cs_free_string (shared process allocator).
+
+use evidence_provenance::behaviour;
+use evidence_provenance::pyjson::{self, Json};
+use std::collections::{BTreeSet, HashSet};
+use std::ffi::{CStr, CString};
+use std::os::raw::c_char;
+
+// ── helpers ──────────────────────────────────────────────────────────────
+
+fn obj(pairs: Vec<(&str, Json)>) -> Json {
+    Json::Obj(pairs.into_iter().map(|(k, v)| (k.to_string(), v)).collect())
 }
-pub fn post_change_obligations(scope_files: &[String]) -> Vec<String> {
-    scope_files
+
+trait IntoJson {
+    fn into_json(self) -> Json;
+}
+impl IntoJson for &str {
+    fn into_json(self) -> Json {
+        Json::Str(self.to_string())
+    }
+}
+impl IntoJson for String {
+    fn into_json(self) -> Json {
+        Json::Str(self)
+    }
+}
+impl IntoJson for Json {
+    fn into_json(self) -> Json {
+        self
+    }
+}
+
+fn arr<S: IntoJson>(v: Vec<S>) -> Json {
+    Json::Arr(v.into_iter().map(IntoJson::into_json).collect())
+}
+
+fn s(v: &Json, key: &str) -> String {
+    match v.get(key) {
+        Some(Json::Str(x)) => x.clone(),
+        _ => String::new(),
+    }
+}
+
+fn entities(doc: &Json) -> Vec<&Json> {
+    match doc.get("entities") {
+        Some(Json::Arr(items)) => items.iter().collect(),
+        _ => Vec::new(),
+    }
+}
+
+fn relationships(doc: &Json) -> Vec<Json> {
+    match doc.get("relationships") {
+        Some(Json::Arr(items)) => items.clone(),
+        _ => Vec::new(),
+    }
+}
+
+fn get_entity<'a>(doc: &'a Json, eid: &str) -> Option<&'a Json> {
+    entities(doc).into_iter().find(|e| s(e, "id") == eid)
+}
+
+/// model.children: belongs_to edges reversed, relationship order kept.
+fn children(doc: &Json, eid: &str) -> Vec<String> {
+    relationships(doc)
         .iter()
-        .map(|f| format!("invalidate_behaviours: {}", f))
+        .filter(|r| {
+            s(r, "type") == "belongs_to"
+                && s(r, "to") == eid
+                && s(r, "from") != eid
+        })
+        .map(|r| s(r, "from"))
         .collect()
 }
+
+/// model.beneath: transitive descendants via LIFO frontier, then sorted.
+fn beneath(doc: &Json, eid: &str) -> Vec<String> {
+    let mut seen: HashSet<String> = HashSet::new();
+    seen.insert(eid.to_string());
+    let mut out: Vec<String> = Vec::new();
+    let mut frontier: Vec<String> = vec![eid.to_string()];
+    while let Some(cur) = frontier.pop() {
+        for child in children(doc, &cur) {
+            if seen.insert(child.clone()) {
+                out.push(child.clone());
+                frontier.push(child);
+            }
+        }
+    }
+    out.sort();
+    out
+}
+
+fn owning_domain(doc: &Json, eid: &str) -> Json {
+    let mut cur = eid.to_string();
+    let mut hops = 0;
+    while hops < 16 {
+        if cur.starts_with("domain:") {
+            return Json::Str(cur);
+        }
+        let parents: Vec<String> = relationships(doc)
+            .iter()
+            .filter(|r| s(r, "type") == "belongs_to" && s(r, "from") == cur)
+            .map(|r| s(r, "to"))
+            .collect();
+        match parents.into_iter().next() {
+            Some(p) => cur = p,
+            None => return Json::Null,
+        }
+        hops += 1;
+    }
+    Json::Null
+}
+
+fn dependencies(doc: &Json, eid: &str) -> Vec<String> {
+    let mut v: Vec<String> = relationships(doc)
+        .iter()
+        .filter(|r| {
+            s(r, "type") == "depends_on"
+                && s(r, "from") == eid
+                && s(r, "status") != "stale"
+        })
+        .map(|r| s(r, "to"))
+        .collect();
+    v.sort();
+    v.dedup();
+    v
+}
+
+fn consumers(doc: &Json, eid: &str) -> Vec<String> {
+    let mut v: Vec<String> = relationships(doc)
+        .iter()
+        .filter(|r| {
+            let t = s(r, "type");
+            (t == "depends_on" || t == "consumes")
+                && s(r, "to") == eid
+                && s(r, "status") != "stale"
+        })
+        .map(|r| s(r, "from"))
+        .collect();
+    v.sort();
+    v.dedup();
+    v
+}
+
+fn interfaces_for(doc: &Json, eid: &str) -> Vec<String> {
+    let mut v: Vec<String> = relationships(doc)
+        .iter()
+        .filter(|r| s(r, "type") == "provides" && s(r, "from") == eid)
+        .map(|r| s(r, "to"))
+        .collect();
+    v.sort();
+    v.dedup();
+    v
+}
+
+fn consumes_of(doc: &Json, eid: &str) -> Vec<String> {
+    let mut v: Vec<String> = relationships(doc)
+        .iter()
+        .filter(|r| s(r, "type") == "consumes" && s(r, "from") == eid)
+        .map(|r| s(r, "to"))
+        .collect();
+    v.sort();
+    v.dedup();
+    v
+}
+
+fn edges_from(doc: &Json, eid: &str, ty: &str) -> Vec<String> {
+    let mut v: Vec<String> = relationships(doc)
+        .iter()
+        .filter(|r| s(r, "type") == ty && s(r, "from") == eid)
+        .map(|r| s(r, "to"))
+        .collect();
+    v.sort();
+    v.dedup();
+    v
+}
+
+/// impact_closure — the structurally bounded scope (rule J), oracle-identical
+/// (Nx affected semantics, bounded reverse-dependency BFS).
+fn impact_closure(doc: &Json, eid: &str) -> Json {
+    if get_entity(doc, eid).is_none() {
+        return obj(vec![(
+            "error",
+            Json::Str(format!("unknown entity {}", py_repr_str(eid))),
+        )]);
+    }
+    let mut seen: HashSet<String> = HashSet::new();
+    seen.insert(eid.to_string());
+    let mut consumers_set: HashSet<String> = HashSet::new();
+    let mut frontier: Vec<String> = vec![eid.to_string()];
+    let mut depth = 0usize;
+    while !frontier.is_empty() && depth < 8 {
+        let mut nxt = Vec::new();
+        for cur in &frontier {
+            for c in consumers(doc, cur) {
+                if seen.insert(c.clone()) {
+                    consumers_set.insert(c.clone());
+                    nxt.push(c);
+                }
+            }
+        }
+        frontier = nxt;
+        depth += 1;
+    }
+    let direct_deps = dependencies(doc, eid);
+    for iface in interfaces_for(doc, eid) {
+        for c in consumers(doc, &iface) {
+            if seen.insert(c.clone()) {
+                consumers_set.insert(c);
+            }
+        }
+    }
+    let mut scope: BTreeSet<String> = BTreeSet::new();
+    scope.insert(eid.to_string());
+    scope.extend(consumers_set.iter().cloned());
+    let mut state_surfaces: BTreeSet<String> = BTreeSet::new();
+    let mut verification_scope: BTreeSet<String> = BTreeSet::new();
+    let mut deployment_scope: BTreeSet<String> = BTreeSet::new();
+    for e in std::iter::once(eid.to_string()).chain(consumers_set.iter().cloned()) {
+        for x in edges_from(doc, &e, "owns") {
+            state_surfaces.insert(x);
+        }
+        for x in edges_from(doc, &e, "verified_by") {
+            verification_scope.insert(x);
+        }
+        for x in edges_from(doc, &e, "deployed_by") {
+            deployment_scope.insert(x);
+        }
+    }
+    let uncertainty = arr(vec![obj(vec![
+        ("kind", Json::Str("dynamic-consumer".into())),
+        (
+            "note",
+            Json::Str(
+                "static model cannot see dynamic dispatch/string-keyed \
+                 lookup/late-binding consumers \u{2014} reconcilers must \
+                 verify at runtime"
+                    .into(),
+            ),
+        ),
+    ])]);
+    obj(vec![
+        ("target", Json::Str(eid.to_string())),
+        ("owning_domain", owning_domain(doc, eid)),
+        ("mutation_scope", arr(scope.into_iter().collect::<Vec<_>>())),
+        (
+            "observation_scope",
+            obj(vec![
+                ("direct_dependencies", arr(direct_deps)),
+                (
+                    "interfaces_provided",
+                    arr(interfaces_for(doc, eid)),
+                ),
+                ("interfaces_consumed", arr(consumes_of(doc, eid))),
+            ]),
+        ),
+        (
+            "state_surfaces",
+            arr(state_surfaces.into_iter().collect::<Vec<_>>()),
+        ),
+        (
+            "verification_scope",
+            arr(verification_scope.into_iter().collect::<Vec<_>>()),
+        ),
+        (
+            "deployment_scope",
+            arr(deployment_scope.into_iter().collect::<Vec<_>>()),
+        ),
+        (
+            "invariants",
+            arr(vec![
+                Json::Str(format!(
+                    "belongs_to hierarchy of {eid} remains stable"
+                )),
+                Json::Str("canonical-state ownership unchanged".into()),
+            ]),
+        ),
+        ("non_goals", Json::Arr(vec![])),
+        ("uncertainty", uncertainty),
+        ("closure_depth", Json::Num(depth as f64)),
+    ])
+}
+
+// ── capability ledger floor (by_component / context_packet) ─────────────
+
+const LEDGER_SCHEMA_VERSION: f64 = 1.0;
+
+fn ledger_by_component<'a>(ledger: &'a Json, component: &str) -> Vec<&'a Json> {
+    match ledger.get("capabilities") {
+        Some(Json::Obj(pairs)) => pairs
+            .iter()
+            .map(|(_, r)| r)
+            .filter(|r| {
+                let m = s(r, "maturity");
+                m != "retired" && m != "superseded" && s(r, "component") == component
+            })
+            .collect(),
+        _ => Vec::new(),
+    }
+}
+
+fn ledger_context_packet(ledger: &Json, capability_id: &str) -> Json {
+    let key = if capability_id.contains(':') {
+        capability_id.to_string()
+    } else {
+        format!("capability:{capability_id}")
+    };
+    let rec = match ledger.get("capabilities") {
+        Some(Json::Obj(pairs)) => pairs
+            .iter()
+            .find(|(k, _)| k.as_str() == key)
+            .map(|(_, r)| r),
+        _ => None,
+    };
+    let rec = match rec {
+        Some(r) => r,
+        None => {
+            return obj(vec![(
+                "error",
+                Json::Str(format!("unknown capability {}", py_repr_str(capability_id))),
+            )])
+        }
+    };
+    let fields = [
+        "capability_id", "name", "purpose", "component",
+        "source_locations", "contract", "state_touched", "dependencies",
+        "consumers", "invariants", "behaviour_specs", "maturity",
+        "constraints", "generation", "limitations",
+    ];
+    let mut pairs: Vec<(String, Json)> =
+        vec![("schema_version".into(), Json::Num(LEDGER_SCHEMA_VERSION))];
+    for f in fields {
+        pairs.push((f.to_string(), rec.get(f).cloned().unwrap_or(Json::Null)));
+    }
+    Json::Obj(pairs)
+}
+
+// ── 19E: context_packet ──────────────────────────────────────────────────
+
+pub fn context_packet(
+    model: &Json,
+    ledger: &Json,
+    target: &str,
+    intent: &str,
+    basis_summary: Json,
+    max_local_files: usize,
+    now: &str,
+) -> Json {
+    let entity = match get_entity(model, target) {
+        Some(e) => e,
+        None => {
+            return obj(vec![(
+                "error",
+                Json::Str(format!("unknown structural target {}", py_repr_str(target))),
+            )])
+        }
+    };
+    let caps = ledger_by_component(ledger, target);
+    let local = match caps.first() {
+        Some(cap) => ledger_context_packet(ledger, &s(cap, "capability_id")),
+        None => Json::Null,
+    };
+    let has_source = |e: &Json| -> bool {
+        let src = match e.get("identities") {
+            Some(i) => match i.get("source") {
+                Some(Json::Str(x)) => !x.is_empty(),
+                _ => false,
+            },
+            None => false,
+        };
+        src
+    };
+    let beneath_count = beneath(model, target)
+        .into_iter()
+        .filter(|b| match get_entity(model, b) {
+            Some(e) => !has_source(e),
+            None => false,
+        })
+        .count();
+    let mut files: Vec<String> = Vec::new();
+    for eid in std::iter::once(target.to_string())
+        .chain(children(model, target))
+    {
+        let e = get_entity(model, &eid);
+        let mut src = String::new();
+        if let Some(e) = e {
+            if let Some(Json::Str(x)) = e
+                .get("ext")
+                .and_then(|x| x.get("source"))
+            {
+                src = x.clone();
+            }
+            if src.is_empty() {
+                if let Some(Json::Str(x)) = e
+                    .get("identities")
+                    .and_then(|x| x.get("source"))
+                {
+                    src = x.clone();
+                }
+            }
+        }
+        if !src.is_empty() {
+            files.push(src);
+        }
+    }
+    for cap in &caps {
+        if let Some(Json::Arr(locs)) = cap.get("source_locations") {
+            for l in locs {
+                if let Json::Str(x) = l {
+                    files.push(x.clone());
+                }
+            }
+        }
+    }
+    let mut file_set: BTreeSet<String> = BTreeSet::new();
+    file_set.extend(files);
+    let local_files: Vec<String> =
+        file_set.into_iter().take(max_local_files).collect();
+    let closure = impact_closure(model, target);
+    let mut iface_consumers: BTreeSet<String> = BTreeSet::new();
+    for iface in interfaces_for(model, target) {
+        for c in consumers(model, &iface) {
+            if c != target {
+                iface_consumers.insert(c);
+            }
+        }
+    }
+    let mut connected_consumers: BTreeSet<String> = BTreeSet::new();
+    connected_consumers.extend(consumers(model, target));
+    connected_consumers.extend(iface_consumers.iter().cloned());
+    let uncertainty = closure.get("uncertainty").cloned().unwrap_or(Json::Arr(vec![]));
+    let connected = obj(vec![
+        ("dependencies", arr(dependencies(model, target))),
+        (
+            "consumers",
+            arr(connected_consumers.into_iter().collect::<Vec<_>>()),
+        ),
+        (
+            "interface_mediated_consumers",
+            arr(iface_consumers.into_iter().collect::<Vec<_>>()),
+        ),
+        ("interfaces_provided", arr(interfaces_for(model, target))),
+        ("state_owned", arr(edges_from(model, target, "owns"))),
+        (
+            "verification",
+            arr(edges_from(model, target, "verified_by")),
+        ),
+        ("deployment", arr(edges_from(model, target, "deployed_by"))),
+        ("uncertainty", uncertainty),
+    ]);
+    let target_kind = match entity.get("kind") {
+        Some(Json::Str(k)) => Json::Str(k.clone()),
+        _ => Json::Null,
+    };
+    obj(vec![
+        ("schema_version", Json::Num(1.0)),
+        ("intent", Json::Str(intent.to_string())),
+        ("target", Json::Str(target.to_string())),
+        ("target_kind", target_kind),
+        ("owning_domain", owning_domain(model, target)),
+        ("local", local),
+        ("local_files", arr(local_files)),
+        ("beneath_count", Json::Num(beneath_count as f64)),
+        ("connected", connected),
+        ("basis_summary", basis_summary),
+        ("generated_at", Json::Str(now.to_string())),
+    ])
+}
+
+// ── 19H: predictive vs descriptive impact ────────────────────────────────
+
+pub fn predictive_impact(model: &Json, target: &str) -> Json {
+    impact_closure(model, target)
+}
+
+pub fn descriptive_impact(
+    model: &Json,
+    target: &str,
+    observed_changed: &[String],
+    generation: &str,
+    now: &str,
+) -> Json {
+    let predicted = impact_closure(model, target);
+    let predicted_scope: HashSet<String> = match predicted.get("mutation_scope") {
+        Some(Json::Arr(items)) => items
+            .iter()
+            .filter_map(|v| match v {
+                Json::Str(x) => Some(x.clone()),
+                _ => None,
+            })
+            .collect(),
+        _ => HashSet::new(),
+    };
+    let observed: HashSet<String> = observed_changed.iter().cloned().collect();
+    let mut predicted_sorted: Vec<String> =
+        predicted_scope.iter().cloned().collect();
+    predicted_sorted.sort();
+    let mut observed_sorted: Vec<String> = observed.iter().cloned().collect();
+    observed_sorted.sort();
+    let mut unexpected: Vec<String> =
+        observed.difference(&predicted_scope).cloned().collect();
+    unexpected.sort();
+    let mut unaffected: Vec<String> =
+        predicted_scope.difference(&observed).cloned().collect();
+    unaffected.sort();
+    obj(vec![
+        ("target", Json::Str(target.to_string())),
+        ("generation", Json::Str(generation.to_string())),
+        ("predicted", arr(predicted_sorted)),
+        ("observed", arr(observed_sorted)),
+        ("unexpected", arr(unexpected)),
+        ("unaffected_as_predicted", arr(unaffected)),
+        ("recorded_at", Json::Str(now.to_string())),
+    ])
+}
+
+/// record_impact_delta's exact line body (json.dumps sort_keys parity);
+/// the trailing newline is the facade's file-append concern.
+pub fn record_delta_line(delta: &Json) -> String {
+    pyjson::dumps_sorted(delta)
+}
+
+// ── 19I: selective invalidation bridge ───────────────────────────────────
+
+pub fn invalidate_behaviours(
+    behaviours_doc: &mut Json,
+    changed: &[String],
+    generation: &str,
+    now: &str,
+) -> Result<Json, String> {
+    behaviour::invalidate_for_change(behaviours_doc, changed, generation, &Json::Null, now)
+}
+
+// ── 19K: post-change obligations ─────────────────────────────────────────
+
+fn py_repr_str(x: &str) -> String {
+    let mut out = String::from("'");
+    for c in x.chars() {
+        match c {
+            '\\' => out.push_str("\\\\"),
+            '\'' => out.push_str("\\'"),
+            '\n' => out.push_str("\\n"),
+            '\r' => out.push_str("\\r"),
+            '\t' => out.push_str("\\t"),
+            _ => out.push(c),
+        }
+    }
+    out.push('\'');
+    out
+}
+
+fn py_list_repr(items: &[String]) -> String {
+    let inner: Vec<String> = items.iter().map(|i| py_repr_str(i)).collect();
+    format!("[{}]", inner.join(", "))
+}
+
+pub fn post_change_obligations(
+    model: &Json,
+    target: &str,
+    changed: Option<&[String]>,
+    behaviours: Option<&Json>,
+) -> Json {
+    let changed_set: BTreeSet<String> = match changed {
+        Some(c) if !c.is_empty() => c.iter().cloned().collect(),
+        _ => {
+            let mut s = BTreeSet::new();
+            s.insert(target.to_string());
+            s
+        }
+    };
+    let mut obligations: Vec<Json> = Vec::new();
+    let mut seen: HashSet<(String, String)> = HashSet::new();
+    let add = |kind: &str, target_e: &str, why: String,
+                   obligations: &mut Vec<Json>, seen: &mut HashSet<(String, String)>| {
+        if seen.insert((kind.to_string(), target_e.to_string())) {
+            obligations.push(obj(vec![
+                ("kind", Json::Str(kind.to_string())),
+                ("target", Json::Str(target_e.to_string())),
+                ("why", Json::Str(why)),
+            ]));
+        }
+    };
+    for eid in changed_set.iter().map(|x| x.as_str()).collect::<Vec<_>>() {
+        for consumer in consumers(model, eid) {
+            add(
+                "inspect_consumer",
+                &consumer,
+                format!("consumes/depends on changed {eid}"),
+                &mut obligations,
+                &mut seen,
+            );
+        }
+        for iface in interfaces_for(model, eid) {
+            add(
+                "verify_interface",
+                &iface,
+                format!("provided by changed {eid}"),
+                &mut obligations,
+                &mut seen,
+            );
+            for consumer in consumers(model, &iface) {
+                add(
+                    "inspect_consumer",
+                    &consumer,
+                    format!("consumes {iface} provided by changed {eid}"),
+                    &mut obligations,
+                    &mut seen,
+                );
+            }
+        }
+        for store in edges_from(model, eid, "owns") {
+            add(
+                "check_state_compatibility",
+                &store,
+                format!("state owned by changed {eid}"),
+                &mut obligations,
+                &mut seen,
+            );
+        }
+        for iface in consumes_of(model, eid) {
+            add(
+                "check_interface_contract",
+                &iface,
+                format!("changed {eid} consumes {iface} \u{2014} contract must still hold"),
+                &mut obligations,
+                &mut seen,
+            );
+        }
+        for dep in edges_from(model, eid, "deployed_by") {
+            add(
+                "verify_deployment",
+                &dep,
+                format!("deployed surface of changed {eid}"),
+                &mut obligations,
+                &mut seen,
+            );
+        }
+    }
+    if let Some(bh) = behaviours {
+        if let Some(Json::Obj(pairs)) = bh.get("behaviours") {
+            for (_, rec) in pairs {
+                let mut deps: HashSet<String> = HashSet::new();
+                if let Some(Json::Arr(items)) = rec.get("dependencies") {
+                    for d in items {
+                        if let Json::Str(x) = d {
+                            deps.insert(x.clone());
+                        }
+                    }
+                }
+                if let Some(Json::Str(o)) = rec.get("owner_component") {
+                    // Python unions {rec.get("owner_component")} — a None
+                    // owner lands in deps but can never intersect the
+                    // all-string changed set, so it is simply skipped.
+                    deps.insert(o.clone());
+                }
+                let inter: Vec<String> = deps
+                    .iter()
+                    .filter(|d| changed_set.contains(*d))
+                    .cloned()
+                    .collect();
+                if !inter.is_empty() {
+                    let mut sorted_inter = inter.clone();
+                    sorted_inter.sort();
+                    let bid = s(rec, "behaviour_id");
+                    add(
+                        "retest",
+                        &bid,
+                        format!("depends on changed {}", py_list_repr(&sorted_inter)),
+                        &mut obligations,
+                        &mut seen,
+                    );
+                }
+            }
+        }
+    }
+    Json::Arr(obligations)
+}
+
+// ── FFI (doc-envelope) ───────────────────────────────────────────────────
+
+fn envelope(result: Result<Json, (String, String)>) -> *mut c_char {
+    let body = match result {
+        Ok(value) => obj(vec![
+            ("ok", Json::Bool(true)),
+            ("value", value),
+        ]),
+        Err((err, ty)) => obj(vec![
+            ("ok", Json::Bool(false)),
+            ("err", Json::Str(err)),
+            ("type", Json::Str(ty)),
+        ]),
+    };
+    let out = pyjson::dumps(&body);
+    match CString::new(out) {
+        Ok(c) => c.into_raw(),
+        Err(_) => CString::new(r#"{"ok":false,"err":"nul in output","type":"RuntimeError"}"#)
+            .unwrap()
+            .into_raw(),
+    }
+}
+
+fn parse_doc(raw: &str) -> Result<Json, (String, String)> {
+    pyjson::parse(raw).map_err(|e| (e, "ValueError".to_string()))
+}
+
+fn read_arg(ptr: *const c_char) -> Result<String, (String, String)> {
+    if ptr.is_null() {
+        return Ok(String::new());
+    }
+    unsafe { CStr::from_ptr(ptr) }
+        .to_str()
+        .map(|s| s.to_string())
+        .map_err(|e| (e.to_string(), "ValueError".to_string()))
+}
+
+macro_rules! ffi {
+    ($name:ident, $body:expr) => {
+        #[no_mangle]
+        pub extern "C" fn $name(arg: *const c_char) -> *mut c_char {
+            let out: Result<Json, (String, String)> =
+                read_arg(arg).and_then(|raw| $body(&raw));
+            envelope(out)
+        }
+    };
+}
+
+fn str_list(v: Option<&Json>) -> Vec<String> {
+    match v {
+        Some(Json::Arr(items)) => items
+            .iter()
+            .filter_map(|x| match x {
+                Json::Str(s) => Some(s.clone()),
+                _ => None,
+            })
+            .collect(),
+        _ => Vec::new(),
+    }
+}
+
+ffi!(itr_context_packet, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let model = doc.get("model").cloned().unwrap_or(Json::Null);
+    let ledger = doc.get("ledger").cloned().unwrap_or(Json::Null);
+    let target = s(&doc, "target");
+    let intent = match doc.get("intent") {
+        Some(Json::Str(x)) => x.clone(),
+        _ => String::new(),
+    };
+    let basis = doc.get("basis_summary").cloned().unwrap_or(Json::Null);
+    let max_local = match doc.get("max_local_files") {
+        Some(Json::Num(n)) => *n as usize,
+        _ => 12,
+    };
+    let now = s(&doc, "now");
+    Ok(context_packet(&model, &ledger, &target, &intent, basis, max_local, &now))
+});
+
+ffi!(itr_predictive_impact, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let model = doc.get("model").cloned().unwrap_or(Json::Null);
+    let target = s(&doc, "target");
+    Ok(predictive_impact(&model, &target))
+});
+
+ffi!(itr_descriptive_impact, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let model = doc.get("model").cloned().unwrap_or(Json::Null);
+    let target = s(&doc, "target");
+    let observed = str_list(doc.get("observed_changed"));
+    let generation = match doc.get("generation") {
+        Some(Json::Str(x)) => x.clone(),
+        _ => String::new(),
+    };
+    let now = s(&doc, "now");
+    Ok(descriptive_impact(&model, &target, &observed, &generation, &now))
+});
+
+ffi!(itr_record_delta_line, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let delta = doc.get("delta").cloned().unwrap_or(Json::Null);
+    Ok(Json::Str(record_delta_line(&delta)))
+});
+
+ffi!(itr_invalidate, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let mut bh = doc.get("behaviours").cloned().unwrap_or(Json::Null);
+    let changed = str_list(doc.get("changed"));
+    let generation = match doc.get("generation") {
+        Some(Json::Str(x)) => x.clone(),
+        _ => String::new(),
+    };
+    let now = s(&doc, "now");
+    match invalidate_behaviours(&mut bh, &changed, &generation, &now) {
+        Ok(recs) => Ok(obj(vec![("doc", bh), ("recs", recs)])),
+        Err(e) => Err((e, "RuntimeError".to_string())),
+    }
+});
+
+ffi!(itr_post_change_obligations, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let model = doc.get("model").cloned().unwrap_or(Json::Null);
+    let target = s(&doc, "target");
+    let changed = doc.get("changed").map(|c| str_list(Some(c)));
+    let behaviours = doc.get("behaviours").cloned();
+    Ok(post_change_obligations(
+        &model,
+        &target,
+        changed.as_deref(),
+        behaviours.as_ref(),
+    ))
+});
+
+#[cfg(test)]
+mod branch_battery;