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

scripts/gen_abstraction_ladder_truth.py

@@ -0,0 +1,293 @@
+#!/usr/bin/env python3
+"""gen_abstraction_ladder_truth.py — freeze/replay parity truth for
+abstraction_ladder (dichotomy wave, #2479 tranche).
+
+Freeze mode (default): exercises EVERY public function of the LIVE Python
+oracle (src/rune/abstraction_ladder.py) across every branch — next_stage
+(all 7 stages / top / unknown / non-string), can_promote (empty item,
+per-stage threshold miss at exactly threshold-1, the unknown-stage→"top"
+quirk, falsy provenance variants, contradictions, missing/empty
+consequence, belief/rule verified_outcome gate, full promote path per
+stage) and promote (failure verdict + verbatim item, success item
+mutation stage/sightings reset) — and records (kernel_input, expected)
+into src/rust/abstraction_ladder/frozen_al_truth.json.
+
+Replay mode (--replay): re-runs the SAME scenarios with the Rust core
+delegation ACTIVE (RUNE_AL_CORE unset) and asserts both kernel-level
+(loader direct) and facade-level (oracle functions delegate) parity
+against the frozen file.
+
+Poisoned-truth law: freeze pins RUNE_AL_CORE=python BEFORE any import.
+Mutation law: case inputs are deep-copied BEFORE they reach the oracle.
+SIGHTINGS-INT LAW: sightings are ints only in frozen cases — JSON cannot
+distinguish int 2 from float 2.0, and the oracle's f-string renders them
+differently ("2" vs "2.0"), so float sightings are unportable truth.
+"""
+import copy
+import json
+import os
+import sys
+from datetime import datetime, timezone
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TRUTH_PATH = os.path.join(REPO, "src", "rust", "abstraction_ladder",
+                          "frozen_al_truth.json")
+REPLAY = "--replay" in sys.argv
+
+if not REPLAY:
+    os.environ["RUNE_AL_CORE"] = "python"  # poisoned-truth law (freeze side)
+else:
+    os.environ.pop("RUNE_AL_CORE", None)  # live-parity law (facade delegates)
+
+sys.path.insert(0, os.path.join(REPO, "src"))
+sys.path.insert(0, os.path.join(REPO, "src", "rune"))
+
+import abstraction_ladder as oracle_mod  # noqa: E402
+
+if REPLAY:
+    from rune import al_loader  # noqa: E402
+    if not al_loader.available():
+        print("REPLAY ABORT: kernel not built —", al_loader.LIB_PATH)
+        sys.exit(2)
+
+cases = []
+
+
+def case(name, fn, kernel_input, expected):
+    cases.append({
+        "name": name,
+        "fn": fn,
+        "input": kernel_input,
+        "ops": [],
+        "expected": expected,
+    })
+
+
+def complete(stage, sightings=None):
+    it = {"stage": stage, "sightings": oracle_mod.THRESHOLDS[stage],
+          "provenance": "witnessed", "contradictions": [],
+          "consequence": "changes behaviour", "verified_outcome": True}
+    if sightings is not None:
+        it["sightings"] = sightings
+    return it
+
+
+# ── next_stage: every stage, top, unknown, non-string ──────────────────────
+for i, stage in enumerate(oracle_mod.STAGES):
+    want = oracle_mod.STAGES[i + 1] if i + 1 < len(oracle_mod.STAGES) else None
+    case(f"next_stage:{stage}", "next_stage", {"stage": stage}, want)
+case("next_stage:unknown", "next_stage", {"stage": "gossip"}, None)
+case("next_stage:empty", "next_stage", {"stage": ""}, None)
+case("next_stage:missing", "next_stage", {}, None)
+case("next_stage:non_string", "next_stage", {"stage": 7}, None)
+case("next_stage:prefix_like", "next_stage", {"stage": "episodex"}, None)
+
+# ── can_promote: threshold branch per stage (one below) ────────────────────
+for stage in oracle_mod.STAGES:
+    thr = oracle_mod.THRESHOLDS[stage]
+    item = {"stage": stage, "sightings": thr - 1, "provenance": "witnessed",
+            "contradictions": [], "consequence": "changes behaviour",
+            "verified_outcome": True}
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:below_threshold:{stage}", "can_promote",
+         {"item": item}, want)
+
+# exact-threshold boundary for every stage (full promote path)
+for stage in oracle_mod.STAGES:
+    item = complete(stage)
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:at_threshold:{stage}", "can_promote",
+         {"item": item}, want)
+
+# empty item — defaults: stage episode, sightings 0
+want = oracle_mod.can_promote({})
+case("can_promote:empty_item", "can_promote", {"item": {}}, want)
+
+# missing sightings only
+want = oracle_mod.can_promote({"stage": "fact", "provenance": "witnessed",
+                               "consequence": "c"})
+case("can_promote:default_sightings", "can_promote",
+     {"item": {"stage": "fact", "provenance": "witnessed",
+               "consequence": "c"}},
+     want)
+
+# non-string / null stage keys: item.get hands them to STAGES.index,
+# which raises ValueError -> next_stage None -> "top" quirk
+for key, bad_stage in (("non_string", 7), ("null", None)):
+    item = {"stage": bad_stage, "sightings": 9, "provenance": "witnessed",
+            "consequence": "c"}
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:stage_{key}_quirk", "can_promote", {"item": item},
+         want)
+
+# the unknown-stage quirk: next_stage(None result) collapses to "top"
+want = oracle_mod.can_promote({"stage": "gossip", "sightings": 99,
+                               "provenance": "witnessed",
+                               "consequence": "c"})
+case("can_promote:unknown_stage_quirk", "can_promote",
+     {"item": {"stage": "gossip", "sightings": 99,
+               "provenance": "witnessed", "consequence": "c"}},
+     want)
+
+# falsy provenance variants: missing, empty string, empty list, 0, False
+for key, prov in (("missing", None), ("empty_str", ""),
+                  ("empty_list", []), ("zero", 0), ("false", False)):
+    item = {"stage": "observation", "sightings": 5,
+            "contradictions": [], "consequence": "c"}
+    if prov is not None:
+        item["provenance"] = prov
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:provenance_{key}", "can_promote", {"item": item}, want)
+
+# truthy provenance variants survive the gate (promote path)
+for key, prov in (("str", "log"), ("list", ["log"]), ("one", 1),
+                  ("true", True)):
+    item = {"stage": "episode", "sightings": 1, "provenance": prov,
+            "contradictions": [], "consequence": "c"}
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:provenance_{key}_passes", "can_promote",
+         {"item": item}, want)
+
+# contradictions: truthy list, truthy string, dict; falsy empty list /
+# None / False pass
+for key, contra, want_promote in (
+        ("list", ["c1"], False), ("str", "disputed", False),
+        ("dict", {"a": 1}, False), ("empty_list", [], True),
+        ("none", None, True), ("false", False, True)):
+    item = {"stage": "episode", "sightings": 1, "provenance": "log",
+            "consequence": "c"}
+    if contra is not None:
+        item["contradictions"] = contra
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:contradictions_{key}", "can_promote",
+         {"item": item}, want)
+
+# consequence: missing / empty string / empty list / 0 fail; non-empty pass
+for key, cons in (("missing", None), ("empty_str", ""),
+                  ("empty_list", []), ("zero", 0), ("str", "moves"),
+                  ("list", ["moves"])):
+    item = {"stage": "episode", "sightings": 1, "provenance": "log",
+            "contradictions": []}
+    if cons is not None:
+        item["consequence"] = cons
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:consequence_{key}", "can_promote", {"item": item},
+         want)
+
+# belief/rule verified_outcome gate: missing, False, 0 fail; True passes;
+# concept needs NO verified outcome
+for stage in ("belief", "rule"):
+    for key, vo in (("missing", None), ("false", False), ("zero", 0)):
+        item = {"stage": stage, "sightings": 9, "provenance": "log",
+                "contradictions": [], "consequence": "c"}
+        if vo is not None:
+            item["verified_outcome"] = vo
+        want = oracle_mod.can_promote(copy.deepcopy(item))
+        case(f"can_promote:{stage}_vo_{key}", "can_promote",
+             {"item": item}, want)
+    item = {"stage": stage, "sightings": 9, "provenance": "log",
+            "contradictions": [], "consequence": "c",
+            "verified_outcome": True}
+    want = oracle_mod.can_promote(copy.deepcopy(item))
+    case(f"can_promote:{stage}_vo_true", "can_promote", {"item": item}, want)
+
+item = {"stage": "concept", "sightings": 3, "provenance": "log",
+        "contradictions": [], "consequence": "c",
+        "verified_outcome": False}
+want = oracle_mod.can_promote(copy.deepcopy(item))
+case("can_promote:concept_ignores_vo", "can_promote", {"item": item}, want)
+
+# gate order: threshold beats provenance beats contradictions beats
+# consequence beats verified_outcome
+item = {"stage": "belief", "sightings": 1}
+want = oracle_mod.can_promote(copy.deepcopy(item))
+case("can_promote:gate_order_threshold_first", "can_promote",
+     {"item": item}, want)
+item = {"stage": "belief", "sightings": 9, "contradictions": ["x"],
+        "consequence": "c", "verified_outcome": True}
+want = oracle_mod.can_promote(copy.deepcopy(item))
+case("can_promote:gate_order_provenance_second", "can_promote",
+     {"item": item}, want)
+
+# ── promote: failure keeps the item verbatim ───────────────────────────────
+for name, item in (
+        ("empty", {}),
+        ("below", {"stage": "observation", "sightings": 2,
+                   "provenance": "log", "consequence": "c"}),
+        ("top", {"stage": "skill", "sightings": 9, "provenance": "log",
+                 "consequence": "c"}),
+        ("unknown", {"stage": "gossip", "sightings": 9}),
+        ("no_consequence", {"stage": "fact", "sightings": 4,
+                            "provenance": "log"}),
+        ("belief_no_vo", {"stage": "belief", "sightings": 9,
+                          "provenance": "log", "consequence": "c"})):
+    want = oracle_mod.promote(copy.deepcopy(item))
+    case(f"promote:refusal_{name}", "promote", {"item": item}, want)
+
+# ── promote: success mutates stage + resets sightings ──────────────────────
+for stage in ("episode", "fact", "observation", "concept", "belief", "rule"):
+    item = complete(stage)
+    want = oracle_mod.promote(copy.deepcopy(item))
+    case(f"promote:success_{stage}", "promote", {"item": item}, want)
+
+# extra fields preserved through the copy; only stage/sightings touched
+item = {"stage": "rule", "sightings": 5, "provenance": "log",
+        "contradictions": [], "consequence": "c", "verified_outcome": True,
+        "id": "x9", "tags": ["a", "b"], "nested": {"k": [1, 2]}}
+want = oracle_mod.promote(copy.deepcopy(item))
+case("promote:extra_fields_preserved", "promote", {"item": item}, want)
+
+# ── freeze / replay ─────────────────────────────────────────────────────────
+
+def kernel_call(fn, kernel_input):
+    if fn == "next_stage":
+        return al_loader.next_stage(kernel_input.get("stage"))
+    if fn == "can_promote":
+        return al_loader.can_promote(kernel_input["item"])
+    return al_loader.promote(kernel_input["item"])
+
+
+def facade_call(fn, kernel_input):
+    if fn == "next_stage":
+        return oracle_mod.next_stage(kernel_input.get("stage"))
+    if fn == "can_promote":
+        return oracle_mod.can_promote(kernel_input["item"])
+    return oracle_mod.promote(kernel_input["item"])
+
+
+if not REPLAY:
+    for c in cases:
+        got = facade_call(c["fn"], copy.deepcopy(c["input"]))
+        assert got == c["expected"], (
+            f"self-check failed on {c['name']}: {got!r} != {c['expected']!r}")
+    doc = {
+        "frozen": datetime.now(timezone.utc).isoformat(),
+        "oracle": "src/rune/abstraction_ladder.py",
+        "pin": "RUNE_AL_CORE=python",
+        "note": "Frozen via scripts/gen_abstraction_ladder_truth.py "
+                "against the live Python oracle. Battery embeds this "
+                "VERBATIM. SIGHTINGS-INT LAW: sightings are ints only "
+                "(JSON cannot distinguish int 2 from float 2.0 and the "
+                "oracle's f-string renders them differently).",
+        "cases": cases,
+    }
+    with open(TRUTH_PATH, "w") as f:
+        json.dump(doc, f, indent=1)
+        f.write("\n")
+    print(f"FROZEN {len(cases)} cases -> {TRUTH_PATH}")
+else:
+    fails = []
+    for c in cases:
+        want = c["expected"]
+        kg = kernel_call(c["fn"], copy.deepcopy(c["input"]))
+        if kg != want:
+            fails.append(f"KERNEL {c['name']}: {kg!r} != {want!r}")
+        fg = facade_call(c["fn"], copy.deepcopy(c["input"]))
+        if fg != want:
+            fails.append(f"FACADE {c['name']}: {fg!r} != {want!r}")
+    if fails:
+        print(f"REPLAY FAILED {len(fails)}/{len(cases)}:")
+        for f_ in fails:
+            print(" ", f_)
+        sys.exit(1)
+    print(f"REPLAY OK {len(cases)}/{len(cases)} (kernel + facade)")

src/rune/abstraction_ladder.py

@@ -5,10 +5,35 @@ episode → fact → observation → concept → belief → rule → skill.
 Each promotion needs: evidence threshold, provenance, contradiction
 handling, prediction/behavioural consequence, outcome-based revision.
 Schema alone does not count — the ladder moves items or it doesn't.
+
+Dichotomy wave #2479 tranche: the ladder kernel is ported to FULL Rust
+(src/rust/abstraction_ladder, parity oracle frozen_al_truth.json). This
+facade delegates to the Rust core via rune/al_loader when built; the
+Python implementation below is the strangler fallback, pinned with
+RUNE_AL_CORE=python.
 """
 
 from __future__ import annotations
 
+import os
+
+
+def _core():
+    """The Rust abstraction_ladder core when available (strangler: the
+    Python fallback is load-bearing). RUNE_AL_CORE=python pins the
+    fallback."""
+    if os.environ.get("RUNE_AL_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import al_loader
+    except Exception:  # noqa: BLE001 — any loader failure means fallback
+        return None
+    try:
+        return al_loader if al_loader.available() else None
+    except Exception:  # noqa: BLE001 — availability probe must never crash
+        return None
+
+
 STAGES = ("episode", "fact", "observation", "concept", "belief", "rule", "skill")
 
 # Evidence threshold per promotion (sightings required)
@@ -25,6 +50,9 @@ THRESHOLDS = {
 
 def next_stage(stage: str) -> str | None:
     """The stage after this one (None at the top)."""
+    core = _core()
+    if core is not None:
+        return core.next_stage(stage)
     try:
         i = STAGES.index(stage)
     except ValueError:
@@ -36,6 +64,9 @@ def can_promote(item: dict) -> dict:
     """Can this item promote? Needs: evidence threshold met, provenance
     present, no open contradictions, a stated consequence, and (for
     belief+) a verified outcome."""
+    core = _core()
+    if core is not None:
+        return core.can_promote(item)
     stage = item.get("stage", "episode")
     nxt = next_stage(stage)
     if nxt is None:
@@ -59,6 +90,9 @@ def can_promote(item: dict) -> dict:
 
 def promote(item: dict) -> dict:
     """Promote one stage (or return the refusal)."""
+    core = _core()
+    if core is not None:
+        return core.promote(item)
     verdict = can_promote(item)
     if not verdict["promote"]:
         return {"item": item, **verdict}

src/rune/al_loader.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+"""al_loader.py — ctypes loader for the Rust abstraction_ladder core
+(dichotomy wave #2479 tranche). Mirrors lc_loader.py / broker_loader.py:
+
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * al_next_stage({"stage"})        -> stage string or null
+  * al_can_promote({"item": {...}}) -> verdict doc
+  * al_promote({"item": {...}})     -> {"item": ..., ...verdict}
+
+Pure kernel: no ops, no store seams, no op callback.
+
+Free: capability_scoring's cs_free_string (shared process). NEVER re-export
+a free symbol here. Override the .so with AL_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" / "abstraction_ladder"
+LIB_PATH = Path(os.environ.get("AL_RUST_LIB",
+                               _CRATE_DIR / "target" / "release" /
+                               "libabstraction_ladder.so"))
+
+_lib = None
+
+SYMBOLS = ("al_next_stage", "al_can_promote", "al_promote")
+
+
+class AbstractionLadderCoreUnavailable(FileNotFoundError):
+    """The Rust abstraction_ladder core is not built. Callers fall back
+    to Python (strangler law; RUNE_AL_CORE=python pins the fallback)."""
+
+
+def available() -> bool:
+    return LIB_PATH.is_file()
+
+
+def _load() -> ctypes.CDLL:
+    global _lib
+    if _lib is not None:
+        return _lib
+    if not LIB_PATH.is_file():
+        raise AbstractionLadderCoreUnavailable(str(LIB_PATH))
+    lib = ctypes.CDLL(str(LIB_PATH))
+    for sym in SYMBOLS:
+        fn = getattr(lib, sym)
+        fn.argtypes = [ctypes.c_void_p]
+        fn.restype = ctypes.c_void_p  # char* — never default c_int
+    _lib = lib
+    return lib
+
+
+def _free(ptr) -> None:
+    """Free via capability_scoring's cs_free_string (same process)."""
+    try:
+        scoring = ctypes.CDLL(str(
+            _REPO / "src" / "rust" / "capability_scoring" / "target" /
+            "release" / "libcapability_scoring.so"))
+        scoring.cs_free_string.argtypes = [ctypes.c_void_p]
+        scoring.cs_free_string(ptr)
+    except OSError:
+        pass  # leak-on-fallback only if capability_scoring absent; bounded
+
+
+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(ptr)
+    doc = json.loads(out)
+    if not doc.get("ok"):
+        raise RuntimeError(doc.get("err", f"{sym} failed"))
+    return doc["value"]
+
+
+def next_stage(stage):
+    return _call("al_next_stage", {"stage": stage})
+
+
+def can_promote(item: dict) -> dict:
+    return _call("al_can_promote", {"item": item})
+
+
+def promote(item: dict) -> dict:
+    return _call("al_promote", {"item": item})

src/rust/abstraction_ladder/Cargo.lock

@@ -0,0 +1,14 @@
+# This file is automatically @generated by Cargo.
+# It is not intended for manual editing.
+version = 4
+
+[[package]]
+name = "abstraction_ladder"
+version = "0.1.0"
+dependencies = [
+ "capability_scoring",
+]
+
+[[package]]
+name = "capability_scoring"
+version = "0.1.0"

src/rust/abstraction_ladder/Cargo.toml

@@ -0,0 +1,12 @@
+[package]
+name = "abstraction_ladder"
+version = "0.1.0"
+edition = "2021"
+description = "Promotion ladder kernel (episode→skill) — dichotomy wave #2479 tranche; zero-dep (pyjson via capability_scoring)"
+
+[lib]
+name = "abstraction_ladder"
+crate-type = ["rlib", "cdylib", "staticlib"]
+
+[dependencies]
+capability_scoring = { path = "../capability_scoring" }

src/rust/abstraction_ladder/src/branch_battery.rs

@@ -0,0 +1,97 @@
+//! branch_battery — replays frozen_al_truth.json VERBATIM (76 cases
+//! frozen from the live Python oracle, RUNE_AL_CORE=python pinned,
+//! inputs deep-copied before capture). Order-insensitive object
+//! comparison; numbers compared exactly (both sides IEEE-754 f64).
+
+use capability_scoring::pyjson::{parse, dumps, Json};
+
+use crate::ladder;
+
+const TRUTH: &str = include_str!("../frozen_al_truth.json");
+
+fn jeq(a: &Json, b: &Json) -> bool {
+    match (a, b) {
+        (Json::Num(x), Json::Num(y)) => x == y,
+        (Json::Str(x), Json::Str(y)) => x == y,
+        (Json::Bool(x), Json::Bool(y)) => x == y,
+        (Json::Null, Json::Null) => true,
+        (Json::Arr(x), Json::Arr(y)) => {
+            x.len() == y.len() && x.iter().zip(y).all(|(i, j)| jeq(i, j))
+        }
+        (Json::Obj(x), Json::Obj(y)) => {
+            x.len() == y.len()
+                && x.iter()
+                    .all(|(k, v)| y.iter().any(|(k2, v2)| k == k2 && jeq(v, v2)))
+        }
+        _ => false,
+    }
+}
+
+fn as_arr(v: &Json) -> Vec<Json> {
+    match v {
+        Json::Arr(a) => a.clone(),
+        _ => vec![],
+    }
+}
+
+fn as_str(v: Option<&Json>) -> Option<&str> {
+    match v {
+        Some(Json::Str(s)) => Some(s.as_str()),
+        _ => None,
+    }
+}
+
+fn run_case(fn_name: &str, args: &Json) -> Result<Json, String> {
+    match fn_name {
+        "next_stage" => {
+            let stage = as_str(args.get("stage"));
+            Ok(match ladder::next_stage(stage) {
+                Some(nxt) => Json::Str(nxt.into()),
+                None => Json::Null,
+            })
+        }
+        "can_promote" => {
+            let item = args.get("item").ok_or("missing item")?;
+            Ok(ladder::can_promote(item))
+        }
+        "promote" => {
+            let item = args.get("item").ok_or("missing item")?;
+            Ok(ladder::promote(item))
+        }
+        other => Err(format!("unknown fn {}", other)),
+    }
+}
+
+#[test]
+fn frozen_truth_replays_verbatim() {
+    let truth = parse(TRUTH).expect("frozen_al_truth.json parses");
+    let cases = as_arr(truth.get("cases").expect("cases key"));
+    assert!(cases.len() >= 76, "expected the full frozen battery, got {}", cases.len());
+    let mut failures: Vec<String> = vec![];
+    for (i, case) in cases.iter().enumerate() {
+        let fn_name = match case.get("fn") {
+            Some(Json::Str(s)) => s.as_str(),
+            _ => panic!("case {i} missing fn"),
+        };
+        let args = case.get("input").cloned().unwrap_or(Json::Obj(vec![]));
+        let expected = case.get("expected").expect("expected key");
+        match run_case(fn_name, &args) {
+            Ok(got) if jeq(&got, expected) => {}
+            Ok(got) => failures.push(format!(
+                "case {i} ({} {}): kernel {:?} != expected {}",
+                fn_name,
+                case.get("name").and_then(|v| as_str(Some(v))).unwrap_or("?"),
+                dumps(&got),
+                dumps(expected),
+            )),
+            Err(e) => failures.push(format!("case {i} ({}): error {}", fn_name, e)),
+        }
+    }
+    assert!(
+        failures.is_empty(),
+        "frozen-truth battery failed {}/{}:\n{}",
+        failures.len(),
+        cases.len(),
+        failures.join("\n"),
+    );
+}

src/rust/abstraction_ladder/src/ladder.rs

@@ -0,0 +1,193 @@
+//! ladder.rs — the promotion ladder kernel, a verbatim port of
+//! src/rune/abstraction_ladder.py (dichotomy wave #2479 tranche).
+//!
+//! episode → fact → observation → concept → belief → rule → skill.
+//! Each promotion needs: evidence threshold, provenance, contradiction
+//! handling, prediction/behavioural consequence, outcome-based revision.
+//! Schema alone does not count — the ladder moves items or it doesn't.
+//!
+//! PYTHON-MIRRORING LAWS:
+//!   * unknown / non-string stage → next_stage None → can_promote says
+//!     "already at the top (skill)" (the oracle's ValueError collapse).
+//!   * missing stage key defaults to "episode"; missing sightings to 0.
+//!   * provenance / contradictions / consequence / verified_outcome use
+//!     Python truthiness (empty string/list/object, 0, null, false all
+//!     falsy).
+//!   * SIGHTINGS-INT LAW: the frozen truth exercises int sightings only
+//!     — JSON cannot distinguish int 2 from float 2.0 and the oracle's
+//!     f-string renders them differently.
+
+use capability_scoring::pyjson::Json;
+
+pub const STAGES: [&str; 7] = [
+    "episode", "fact", "observation", "concept", "belief", "rule", "skill",
+];
+
+/// Evidence threshold per promotion (sightings required). The 1.0
+/// default mirrors THRESHOLDS.get(stage, 1).
+fn threshold(stage: &str) -> f64 {
+    match stage {
+        "episode" => 1.0,
+        "fact" => 2.0,
+        "observation" => 3.0,
+        "concept" => 3.0,
+        "belief" => 4.0,
+        "rule" => 4.0,
+        "skill" => 5.0,
+        _ => 1.0,
+    }
+}
+
+/// The stage after this one (None at the top). A non-string stage
+/// mirrors the oracle's STAGES.index ValueError → None.
+pub fn next_stage(stage: Option<&str>) -> Option<&'static str> {
+    let i = STAGES.iter().position(|s| Some(*s) == stage)?;
+    STAGES.get(i + 1).copied()
+}
+
+/// Python truthiness for JSON values (bools are ints; 0/0.0, "", [],
+/// {}, null, false are all falsy).
+pub fn truthy(v: &Json) -> bool {
+    match v {
+        Json::Null => false,
+        Json::Bool(b) => *b,
+        Json::Num(x) => *x != 0.0,
+        Json::Str(s) => !s.is_empty(),
+        Json::Arr(a) => !a.is_empty(),
+        Json::Obj(o) => !o.is_empty(),
+    }
+}
+
+/// Python f-string number form: integral values render without a
+/// fractional part; everything else is shortest round-trip repr
+/// (identical to Python for all values the truth exercises).
+fn fmt_num(x: f64) -> String {
+    if x.fract() == 0.0 && x.is_finite() && x.abs() < 9.0e15 {
+        format!("{}", x as i64)
+    } else {
+        format!("{}", x)
+    }
+}
+
+fn stage_of(item: &Json) -> Result<&str, ()> {
+    match item.get("stage") {
+        // missing key → the oracle's item.get("stage", "episode") default
+        None => Ok("episode"),
+        Some(Json::Str(s)) => Ok(s.as_str()),
+        // present but non-string → STAGES.index raises → "top" quirk
+        Some(_) => Err(()),
+    }
+}
+
+fn verdict_top() -> Json {
+    Json::Obj(vec![(
+        "promote".into(),
+        Json::Bool(false),
+    ), (
+        "reason".into(),
+        Json::Str("already at the top (skill)".into()),
+    )])
+}
+
+fn verdict_refuse(reason: String) -> Json {
+    Json::Obj(vec![
+        ("promote".into(), Json::Bool(false)),
+        ("reason".into(), Json::Str(reason)),
+    ])
+}
+
+/// Can this item promote? Needs: evidence threshold met, provenance
+/// present, no open contradictions, a stated consequence, and (for
+/// belief+) a verified outcome.
+pub fn can_promote(item: &Json) -> Json {
+    let stage = match stage_of(item) {
+        Ok(s) => s,
+        Err(()) => return verdict_top(),
+    };
+    let nxt = match next_stage(Some(stage)) {
+        Some(n) => n,
+        None => return verdict_top(),
+    };
+    let sightings = match item.get("sightings") {
+        Some(Json::Num(x)) => *x,
+        _ => 0.0,
+    };
+    let thr = threshold(stage);
+    if sightings < thr {
+        return verdict_refuse(format!(
+            "evidence {} < {}",
+            fmt_num(sightings),
+            fmt_num(thr)
+        ));
+    }
+    if !item.get("provenance").map(truthy).unwrap_or(false) {
+        return verdict_refuse("no provenance".into());
+    }
+    if item.get("contradictions").map(truthy).unwrap_or(false) {
+        return verdict_refuse("open contradictions".into());
+    }
+    if !item.get("consequence").map(truthy).unwrap_or(false) {
+        return verdict_refuse("no stated consequence".into());
+    }
+    if (stage == "belief" || stage == "rule")
+        && !item.get("verified_outcome").map(truthy).unwrap_or(false)
+    {
+        return verdict_refuse("belief+ requires a verified outcome".into());
+    }
+    Json::Obj(vec![
+        ("promote".into(), Json::Bool(true)),
+        ("to".into(), Json::Str(nxt.into())),
+        (
+            "reason".into(),
+            Json::Str(format!("threshold met, promoting to {}", nxt)),
+        ),
+    ])
+}
+
+/// dict(item); item[k] = v — an existing key updates IN PLACE (keeping
+/// its position), a new key appends (Python 3.7+ dict law).
+fn set_key(item: &Json, key: &str, val: Json) -> Json {
+    match item {
+        Json::Obj(pairs) => {
+            let mut out: Vec<(String, Json)> = Vec::with_capacity(pairs.len() + 1);
+            let mut placed = false;
+            for (k, v) in pairs {
+                if k == key {
+                    out.push((k.clone(), val.clone()));
+                    placed = true;
+                } else {
+                    out.push((k.clone(), v.clone()));
+                }
+            }
+            if !placed {
+                out.push((key.to_string(), val));
+            }
+            Json::Obj(out)
+        }
+        other => other.clone(),
+    }
+}
+
+/// Promote one stage (or return the refusal).
+pub fn promote(item: &Json) -> Json {
+    let verdict = can_promote(item);
+    let promoted = verdict.get("promote").map(|v| matches!(v, Json::Bool(true)));
+    match promoted {
+        Some(true) => {
+            let to = verdict.get("to").cloned().unwrap_or(Json::Null);
+            let new_item = set_key(&set_key(item, "stage", to), "sightings", Json::Num(0.0));
+            let mut out: Vec<(String, Json)> = vec![("item".into(), new_item)];
+            if let Json::Obj(pairs) = verdict {
+                out.extend(pairs);
+            }
+            Json::Obj(out)
+        }
+        _ => {
+            let mut out: Vec<(String, Json)> = vec![("item".into(), item.clone())];
+            if let Json::Obj(pairs) = verdict {
+                out.extend(pairs);
+            }
+            Json::Obj(out)
+        }
+    }
+}

src/rust/abstraction_ladder/src/lib.rs

@@ -0,0 +1,87 @@
+//! abstraction_ladder — FFI surface for the promotion ladder kernel
+//! (dichotomy wave #2479 tranche). Python facade keeps a
+//! RUNE_AL_CORE=python pin.
+//!
+//! FFI contract (doc-envelope law — every call returns {"ok","value"|"err"}):
+//!   * al_next_stage({"stage"})        -> stage string or null
+//!   * al_can_promote({"item": {...}}) -> verdict doc
+//!   * al_promote({"item": {...}})     -> {"item": ..., ...verdict}
+//!
+//! Pure kernel: no ops, no store seams, no op callback.
+//!
+//! Parity oracle: frozen_al_truth.json (76 cases, frozen from the live
+//! Python oracle).
+
+pub mod ladder;
+
+pub use capability_scoring::pyjson;
+
+#[path = "branch_battery.rs"]
+#[cfg(test)]
+mod branch_battery;
+
+use std::ffi::{CStr, CString};
+use std::os::raw::c_char;
+
+use pyjson::Json;
+
+fn read_arg(ptr: *const c_char) -> Result<Json, String> {
+    if ptr.is_null() {
+        return Err("null argument pointer".into());
+    }
+    let x = unsafe { CStr::from_ptr(ptr) }.to_str().map_err(|e| e.to_string())?;
+    pyjson::parse(x)
+}
+
+fn envelope(r: Result<Json, String>) -> *mut c_char {
+    let v = match r {
+        Ok(x) => Json::Obj(vec![
+            ("ok".into(), Json::Bool(true)),
+            ("value".into(), x),
+        ]),
+        Err(e) => Json::Obj(vec![
+            ("ok".into(), Json::Bool(false)),
+            ("err".into(), Json::Str(e)),
+        ]),
+    };
+    CString::new(pyjson::dumps(&v))
+        .map(|c| c.into_raw())
+        .unwrap_or(std::ptr::null_mut())
+}
+
+/// al_next_stage({"stage"}) — the stage after this one, null at the top.
+#[no_mangle]
+pub extern "C" fn al_next_stage(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| match a.get("stage") {
+        Some(Json::Str(s)) => match ladder::next_stage(Some(s.as_str())) {
+            Some(nxt) => Json::Str(nxt.into()),
+            None => Json::Null,
+        },
+        // non-string / missing stage: the oracle's STAGES.index
+        // ValueError collapses to None
+        _ => Json::Null,
+    }))
+}
+
+/// al_can_promote({"item"}) — the verdict doc.
+#[no_mangle]
+pub extern "C" fn al_can_promote(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).and_then(|a| match a.get("item") {
+        Some(item) => Ok(ladder::can_promote(item)),
+        None => Err("missing item".to_string()),
+    }))
+}
+
+/// al_promote({"item"}) — {"item": mutated copy, ...verdict}.
+#[no_mangle]
+pub extern "C" fn al_promote(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).and_then(|a| match a.get("item") {
+        Some(item) => Ok(ladder::promote(item)),
+        None => Err("missing item".to_string()),
+    }))
+}
+
+// NOTE: al_free_string is deliberately NOT exported — capability_scoring
+// (linked into the same Python process) already exports cs_free_string, and
+// a duplicate free symbol breaks the cdylib link. Callers free every
+// returned pointer through that path.