chore(release): auto-bump version to 0.264.0 — earned by # (minor: converged release (push-to-master bump))

Promotion record: (unknown)

README.md

@@ -1,6 +1,6 @@
 # Rune
 
-**Current release: v0.262.0**
+**Current release: v0.264.0**
 
 **Rune is a persistent personal agent designed to accumulate experience, develop reusable competence, and remain recognisably the same agent even when its underlying foundation model changes.**
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "rune"
-version = "0.262.0"
+version = "0.264.0"
 description = "Rune — a stateful coding assistant with persistent memory, orchestration, and TUI"
 requires-python = ">=3.10"
 

scripts/check_meaningful_events_parity.py

@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+"""Live parity: meaningful_events Rust kernel vs Python oracle (dichotomy
+wave). Layers:
+  1. FROZEN — the 96 frozen-oracle cases replayed through the facade in
+     BOTH core modes (RUNE_ME_CORE=python floor vs auto kernel); exact
+     compare, inputs deep-copied.
+  2. FUZZ   — 400 seeded random trigger/state/now triples (seed 2479)
+     driven through both modes; exact compare (should_fire, fire_due,
+     mark_fired round-trips).
+"""
+import copy
+import json
+import os
+import random
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "src" / "rune"))
+
+import meaningful_events as me  # noqa: E402
+from rune import me_loader  # noqa: E402
+
+if not me_loader.available():
+    print("FATAL: Rust meaningful_events core not built")
+    sys.exit(2)
+
+TRUTH = json.loads(
+    (REPO / "src" / "rust" / "meaningful_events" / "frozen_truth.json")
+    .read_text())
+
+fails = []
+counts = {"frozen": 0, "fuzz": 0}
+
+
+def fail(layer, name, detail):
+    fails.append((layer, name, detail))
+    print(f"FAIL[{layer}] {name}: {detail}")
+
+
+def drive(mode, fn, *args):
+    os.environ["RUNE_ME_CORE"] = mode
+    try:
+        return fn(*copy.deepcopy(args))
+    finally:
+        os.environ["RUNE_ME_CORE"] = "auto"
+
+
+def frozen():
+    for c in TRUTH["should_fire_cases"]:
+        counts["frozen"] += 1
+        args = (c["args"]["trigger"], c["args"]["state"], c["args"]["now"])
+        py = drive("python", me.should_fire, *args)
+        rs = drive("auto", me.should_fire, *args)
+        if py != rs or py != c["expect"]:
+            fail("frozen", f"should_fire/{c['name']}", f"{py!r} != {rs!r}")
+    for c in TRUTH["fire_due_cases"]:
+        counts["frozen"] += 1
+        args = (c["args"]["state"], c["args"]["now"])
+        py = drive("python", me.fire_due, *args)
+        rs = drive("auto", me.fire_due, *args)
+        if py != rs or py != c["expect"]:
+            fail("frozen", f"fire_due/{c['name']}", f"{py!r} != {rs!r}")
+    for c in TRUTH["mark_fired_cases"]:
+        counts["frozen"] += 1
+        args = (c["args"]["state"], c["args"]["trigger"], c["args"]["now"])
+        py = drive("python", me.mark_fired, *args)
+        rs = drive("auto", me.mark_fired, *args)
+        if py != rs or py != c["expect"]:
+            fail("frozen", f"mark_fired/{c['name']}", f"{py!r} != {rs!r}")
+
+
+TRIGGER_NAMES = list(me.TRIGGERS) + ["mystery", ""]
+COND_KEYS = sorted({t[0] for t in me.TRIGGERS.values()})
+STOP_KEYS = sorted({t[2] for t in me.TRIGGERS.values()})
+VALUES = [True, False, 0, 1, "", "yes", None, 0.5]
+
+
+def rand_state(rng):
+    state = {}
+    for k in COND_KEYS + STOP_KEYS:
+        if rng.random() < 0.3:
+            state[k] = rng.choice(VALUES)
+    for name in list(me.TRIGGERS):
+        if rng.random() < 0.3:
+            state[f"last_fired_{name}"] = rng.choice(
+                [0, 0.0, rng.uniform(-500, 1500), 999.0])
+    return state
+
+
+def fuzz():
+    rng = random.Random(2479)
+    for i in range(400):
+        trigger = rng.choice(TRIGGER_NAMES)
+        state = rand_state(rng)
+        now = round(rng.uniform(-100, 5000), 4)
+        py = drive("python", me.should_fire, trigger, state, now)
+        rs = drive("auto", me.should_fire, trigger, state, now)
+        counts["fuzz"] += 1
+        if py != rs:
+            fail("fuzz", f"fire-{i}",
+                 f"{trigger!r}@{now}: {py!r} != {rs!r}")
+            continue
+        py = drive("python", me.fire_due, state, now)
+        rs = drive("auto", me.fire_due, state, now)
+        counts["fuzz"] += 1
+        if py != rs:
+            fail("fuzz", f"due-{i}", f"{py!r} != {rs!r}")
+            continue
+        mark_trigger = rng.choice(TRIGGER_NAMES)
+        py = drive("python", me.mark_fired, state, mark_trigger, now)
+        rs = drive("auto", me.mark_fired, state, mark_trigger, now)
+        counts["fuzz"] += 1
+        if py != rs:
+            fail("fuzz", f"mark-{i}", f"{py!r} != {rs!r}")
+
+
+frozen()
+fuzz()
+total = counts["frozen"] + counts["fuzz"]
+print(f"parity: {total - len(fails)}/{total} "
+      f"(frozen {counts['frozen']}, fuzz {counts['fuzz']})")
+sys.exit(1 if fails else 0)

scripts/freeze_truth.py

@@ -17,6 +17,23 @@ Usage (as a library, from a generator script):
         ("other_fn", (deep_copied_obj,), {}),
     ], oracle_module="rune.module_name")
 
+Class-based modules: a case may be a dict with a dotted fn name and a
+"recv" receiver spec — the receiver is rebuilt from JSON via shape-matched
+module dataclasses, called as an unbound method, and the expected carries
+BOTH the result and the post-call receiver attributes:
+
+    cases = freeze("rune.build_mode", [
+        {"fn": "BuildManager.start_plan", "recv": {"cls": "BuildManager", "state": {...}},
+         "args": [], "kwargs": {}},
+    ], oracle_module="rune.build_mode")
+
+Pinned clock: the harness docstring always promised one — now it exists.
+Set RUNE_FROZEN_NOW (ISO-8601) to patch the oracle module's module-global
+`datetime` with a frozen clock for BOTH freeze and verify (timestamp-bearing
+fns stay byte-identical across processes). Envelope files written via
+write_envelope({"frozen_now", "cases"}) carry their own clock, so `--verify`
+needs no env at all.
+
 Usage (as a script, to re-verify a frozen file against the live oracle):
     python3 scripts/freeze_truth.py --verify src/rust/<crate>/frozen_truth.json \
         --module rune.<module>
@@ -29,10 +46,12 @@ from __future__ import annotations
 
 import argparse
 import copy
+import dataclasses
 import importlib
 import json
 import os
 import sys
+from datetime import datetime
 from pathlib import Path
 
 REPO = Path(__file__).resolve().parents[1]
@@ -44,42 +63,136 @@ def _pin_oracle_cores(module_name: str) -> None:
     os.environ[f"RUNE_{stem.upper()}_CORE"] = "python"
 
 
+def _frozen_clock(iso: str):
+    """A datetime subclass pinned to `iso` (naive); .now(tz) honors tz."""
+    base = datetime.fromisoformat(iso)
+
+    class FrozenDatetime(datetime):
+        @classmethod
+        def now(cls, tz=None):
+            if tz is not None:
+                return base.replace(tzinfo=tz)
+            return base
+
+    return FrozenDatetime
+
+
+def _pin_clock(mod) -> None:
+    """Patch the oracle module's module-global `datetime` when RUNE_FROZEN_NOW
+    is set (modules that read a global datetime — e.g. build_mode._log_phase —
+    then produce identical timestamps in every process)."""
+    frozen_now = os.environ.get("RUNE_FROZEN_NOW", "")
+    if frozen_now and hasattr(mod, "datetime"):
+        mod.datetime = _frozen_clock(frozen_now)
+
+
+def _resolve_fn(mod, fn_name: str):
+    """Module-level name OR dotted path to a class method ("Cls.method")."""
+    obj = mod
+    for part in fn_name.split("."):
+        obj = getattr(obj, part)
+    return obj
+
+
+def _shape_rebuild(mod, value):
+    """Rebuild a JSON value into module types: a dict whose key set EXACTLY
+    matches a module-level dataclass's field names is reconstructed as that
+    dataclass (the convention that lets a frozen `state` dict become a live
+    BuildState again in a fresh process). Everything else deep-copies."""
+    if isinstance(value, dict):
+        for name in dir(mod):
+            obj = getattr(mod, name)
+            if isinstance(obj, type) and dataclasses.is_dataclass(obj):
+                field_names = {f.name for f in dataclasses.fields(obj)}
+                if field_names and field_names == set(value.keys()):
+                    return obj(**{k: _shape_rebuild(mod, v) for k, v in value.items()})
+        return {k: _shape_rebuild(mod, v) for k, v in value.items()}
+    if isinstance(value, list):
+        return [_shape_rebuild(mod, v) for v in value]
+    return copy.deepcopy(value)
+
+
+def _rebuild_receiver(mod, recv: dict):
+    """recv = {"cls": "BuildManager", <attr>: <json value>, ...} → a live
+    instance built via __new__ (no __init__ — that would touch the real
+    filesystem) with the named attrs rebuilt through _shape_rebuild."""
+    cls = getattr(mod, recv["cls"])
+    inst = cls.__new__(cls)
+    for key, value in recv.items():
+        if key == "cls":
+            continue
+        setattr(inst, key, _shape_rebuild(mod, value))
+    return inst
+
+
+def _out(value):
+    """JSON-safe outward snapshot: dataclass instances → asdict."""
+    if dataclasses.is_dataclass(value) and not isinstance(value, type):
+        return dataclasses.asdict(value)
+    return copy.deepcopy(value)
+
+
 def _snap(value):
     """Deep-snapshot a value into JSON-safe form (live-reference aliasing law)."""
     return copy.deepcopy(value)
 
 
+def _run_case(mod, entry) -> dict:
+    """Execute one case (tuple or dict form) against the live module."""
+    recv = None
+    if isinstance(entry, dict):
+        fn_name = entry["fn"]
+        args = entry.get("args", [])
+        kwargs = entry.get("kwargs", {})
+        recv = entry.get("recv")
+    elif len(entry) == 3:
+        fn_name, args, kwargs = entry
+    elif len(entry) == 2:
+        fn_name, args = entry
+        kwargs = {}
+    else:
+        raise ValueError(f"bad case entry: {entry!r}")
+
+    fn = _resolve_fn(mod, fn_name)
+    a_snap = _snap(list(args))
+    k_snap = _snap(dict(kwargs))
+
+    if recv is None:
+        result = fn(*copy.deepcopy(a_snap), **copy.deepcopy(k_snap))
+        expected = _snap(result)
+        return {"fn": fn_name, "args": a_snap, "kwargs": k_snap, "expected": expected}
+
+    inst = _rebuild_receiver(mod, dict(recv))
+    result = fn(inst, *copy.deepcopy(a_snap), **copy.deepcopy(k_snap))
+    expected = {"result": _snap(result)}
+    for key in recv:
+        if key == "cls":
+            continue
+        try:
+            expected[key] = _out(getattr(inst, key))
+        except AttributeError:
+            pass  # attr was never set — the case reconstructed a bare receiver
+    return {
+        "fn": fn_name, "recv": dict(recv),
+        "args": a_snap, "kwargs": k_snap, "expected": expected,
+    }
+
+
 def freeze(module_name: str, cases, oracle_module: str | None = None) -> list[dict]:
-    """Run each (fn_name, args, kwargs) against the live oracle; return truth.
+    """Run each case against the live oracle; return truth.
 
     Args and kwargs are deep-copied before EVERY call — the oracle may
     mutate its inputs (live-reference aliasing trap). The recorded "args"
     are the post-copy snapshots, so the battery replays byte-identical
-    inputs.
+    inputs. Case entries are (fn, args[, kwargs]) tuples or dicts with a
+    dotted "fn" and optional "recv" receiver spec (class-based modules).
     """
     _pin_oracle_cores(oracle_module or module_name)
     mod = importlib.import_module(oracle_module or module_name)
+    _pin_clock(mod)
     truth = []
     for entry in cases:
-        if len(entry) == 3:
-            fn_name, args, kwargs = entry
-        elif len(entry) == 2:
-            fn_name, args = entry
-            kwargs = {}
-        else:
-            raise ValueError(f"bad case entry: {entry!r}")
-        fn = getattr(mod, fn_name, None)
-        if fn is None:
-            raise AttributeError(f"{oracle_module or module_name}.{fn_name} missing — case list must name real functions")
-        a_snap = _snap(list(args))
-        k_snap = _snap(dict(kwargs))
-        result = fn(*copy.deepcopy(a_snap), **copy.deepcopy(k_snap))
-        truth.append({
-            "fn": fn_name,
-            "args": a_snap,
-            "kwargs": k_snap,
-            "expected": _snap(result),
-        })
+        truth.append(_run_case(mod, entry))
     return truth
 
 
@@ -88,14 +201,38 @@ def write(truth: list[dict], path: Path) -> None:
     path.write_text(json.dumps(truth, indent=2, ensure_ascii=False, sort_keys=False) + "\n")
 
 
+def write_envelope(frozen_now: str, truth: list[dict], path: Path) -> None:
+    """Self-contained envelope: the file carries its own pinned clock, so
+    `--verify` needs no RUNE_FROZEN_NOW env. List-format files (write)
+    stay legacy-valid."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(
+        {"frozen_now": frozen_now, "cases": truth},
+        indent=2, ensure_ascii=False, sort_keys=False) + "\n")
+
+
 def verify(path: Path, oracle_module: str) -> int:
     """Re-freeze from the live oracle and byte-compare against the frozen file.
 
     Catches self-consistent misreads: the battery and the generator can
-    agree while the oracle drifted. Returns 0 on byte-exact match.
+    agree while the oracle drifted. Returns 0 on byte-exact match. Accepts
+    a bare case list (legacy) or a {"frozen_now", "cases"} envelope (clock
+    pinned from the file itself).
     """
-    frozen = json.loads(path.read_text())
-    cases = [(c["fn"], c["args"], c["kwargs"]) for c in frozen]
+    raw = json.loads(path.read_text())
+    if isinstance(raw, dict):
+        if raw.get("frozen_now"):
+            os.environ["RUNE_FROZEN_NOW"] = raw["frozen_now"]
+        frozen = raw["cases"]
+    else:
+        frozen = raw
+    cases = []
+    for c in frozen:
+        if isinstance(c, dict):
+            cases.append({"fn": c["fn"], "args": c.get("args", []),
+                          "kwargs": c.get("kwargs", {}), "recv": c.get("recv")})
+        else:
+            cases.append((c["fn"], c["args"], c["kwargs"]))
     fresh = freeze(oracle_module, cases, oracle_module)
     if json.dumps(fresh, sort_keys=True) != json.dumps(frozen, sort_keys=True):
         print(f"MISMATCH: regenerated truth differs from {path}")

scripts/gen_build_mode_gap_truth.py

@@ -0,0 +1,245 @@
+#!/usr/bin/env python3
+"""gen_build_mode_gap_truth.py — build-mode gap-closing truth battery
+(contract BUILDER-2026-10-13T02:02:02Z-dad27f-factory-gap-bm).
+
+Closes the PARTIAL audit gaps: _load_state, _save_state, assess_complexity
+and the full start_* phase-transition family (start_research, start_feasible,
+start_consult, start_consent, start_plan, start_build_phase, start_verify,
+start_integrate, start_record) — the build-pipeline state machine, frozen
+transition-by-transition in declaration order (phase names and ordering are
+load-bearing for the pipeline).
+
+Uses the SHARED harness (scripts/freeze_truth.py): dotted fn names + recv
+receiver rebuild + RUNE_FROZEN_NOW pinned clock. The envelope carries its
+own frozen_now, so `--verify` needs no env:
+
+  PYTHONPATH=src python3 scripts/freeze_truth.py \
+      --verify src/go/build-mode/truth_buildmode_gap.json \
+      --module rune.build_mode
+
+NOTE: the oracle's _log_phase/_save_state write to BUILDS_DIR (module
+global from expanduser). Run generator AND verify with a sandboxed HOME to
+contain the incidental state files (truth never reads them back):
+
+  HOME=$(mktemp -d) PYTHONPATH=src python3 scripts/freeze_truth.py ...
+
+_load_state/_save_state are IO seams: _save_state's byte contract is frozen
+as ledger_load_fixture.json (exact bytes of json.dump(asdict(state),
+indent=2)); the generator also proves the oracle roundtrip (write fixture →
+_load_state → equal state). _load_state's FileNotFoundError path is pinned
+by the Go test (error text "Build <id> not found").
+
+Usage: python3 scripts/gen_build_mode_gap_truth.py   (from repo root)
+"""
+
+import dataclasses
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "scripts"))
+sys.path.insert(0, str(REPO / "src"))
+
+from freeze_truth import _frozen_clock, freeze, write_envelope  # noqa: E402
+from rune import build_mode as bm  # noqa: E402
+
+FROZEN_NOW = "2026-10-13T02:20:00"
+STAMP = FROZEN_NOW + "+00:00"
+BUILD_ID = "20261013-022000"
+
+os.environ["RUNE_FROZEN_NOW"] = FROZEN_NOW  # pinned clock for the harness
+os.environ["RUNE_BUILD_MODE_CORE"] = "python"  # poisoned-truth law
+bm.datetime = _frozen_clock(FROZEN_NOW)  # pin the SEED walk too (recv states
+# must not carry live-clock timestamps — the truth must be self-consistent)
+
+# BYTE-PARITY LAW for the IO fixture: Go maps marshal dict keys SORTED, so
+# every generator-supplied dict below uses sorted keys (Python insertion
+# order == sorted order) — then SaveState can reproduce json.dump bytes
+# exactly. The oracle-internal feasible payload (feasible, risk, details)
+# is the one unavoidable exception — fixture therefore comes from the
+# correction-cycle state (single-key payloads only); dict-heavy states are
+# parity-checked at the parsed level (truth replay + roundtrip test).
+
+
+def fresh_mgr() -> bm.BuildManager:
+    m = bm.BuildManager.__new__(bm.BuildManager)
+    m.state = bm.BuildState(
+        build_id=BUILD_ID, description="gap battery build", created_at=STAMP)
+    return m
+
+
+def seed(mgr) -> dict:
+    return dataclasses.asdict(mgr.state)
+
+
+def main():
+    cases = []
+    mgr = fresh_mgr()
+
+    def call(fn, *args, **kwargs):
+        """Freeze one transition: recv = PRE-state, then apply to the live
+        walk manager (the harness re-runs it on a rebuilt receiver)."""
+        cases.append({
+            "fn": f"BuildManager.{fn.__name__}",
+            "recv": {"cls": "BuildManager", "state": seed(mgr)},
+            "args": list(args), "kwargs": kwargs,
+        })
+        fn(mgr, *args, **kwargs)
+
+    # ── assess_complexity ────────────────────────────────────────────
+    # bare receiver (the fn never touches state) + a seeded receiver.
+    cases.append({"fn": "BuildManager.assess_complexity",
+                  "recv": {"cls": "BuildManager"}, "args": [], "kwargs": {}})
+    cases.append({"fn": "BuildManager.assess_complexity",
+                  "recv": {"cls": "BuildManager", "state": seed(mgr)},
+                  "args": [], "kwargs": {}})
+
+    # ── the canonical 11-phase happy path, IN DECLARATION ORDER ──────
+    call(bm.BuildManager.complete_assess,
+         bm.BuildManager.assess_complexity(mgr))
+    call(bm.BuildManager.start_research)
+    call(bm.BuildManager.complete_research,
+         {"findings": "gap battery pattern", "source": "oracle"})
+    call(bm.BuildManager.start_feasible)
+    call(bm.BuildManager.complete_feasible, True, "low",
+         {"loc_estimate": 120, "worktree_isolated": True})
+    call(bm.BuildManager.start_consult)
+    call(bm.BuildManager.complete_consult, ["gap-closing pattern"])
+    call(bm.BuildManager.start_consent)
+    call(bm.BuildManager.complete_consent, True)
+    call(bm.BuildManager.start_plan)
+    call(bm.BuildManager.complete_plan,
+         [{"batch": 1, "files": ["a.py"]}, {"batch": 2, "files": ["b.py"]}])
+    call(bm.BuildManager.start_build_phase)
+    call(bm.BuildManager.complete_build_phase,
+         [{"batch": 1, "ok": True}, {"batch": 2, "ok": True}])
+    call(bm.BuildManager.start_verify)
+    call(bm.BuildManager.complete_verify, True, [])
+    call(bm.BuildManager.start_integrate)
+    call(bm.BuildManager.complete_integrate, True, [])
+    call(bm.BuildManager.start_record)
+    call(bm.BuildManager.complete_record)
+
+    # ── _save_state branches ─────────────────────────────────────────
+    # state None → early return (no file, no dirs)
+    cases.append({"fn": "BuildManager._save_state",
+                  "recv": {"cls": "BuildManager", "state": None},
+                  "args": [], "kwargs": {}})
+    # state present → writes BUILDS_DIR/<id>/ledger.json (sandboxed HOME);
+    # result None + state unchanged.
+    call(bm.BuildManager._save_state)
+
+    # ── correction cycle (corrections increment; fixed False and True) ─
+    cMgr = fresh_mgr()
+    cMgr.state.corrections = 1
+    def ccall(fn, *args, **kwargs):
+        cases.append({
+            "fn": f"BuildManager.{fn.__name__}",
+            "recv": {"cls": "BuildManager", "state": seed(cMgr)},
+            "args": list(args), "kwargs": kwargs,
+        })
+        fn(cMgr, *args, **kwargs)
+    ccall(bm.BuildManager.start_correct)
+    ccall(bm.BuildManager.complete_correct, False)
+    ccall(bm.BuildManager.start_correct)
+    ccall(bm.BuildManager.complete_correct, True)
+
+    # ── verify/integrate FAILURE branches (failure lives in DETAILS —
+    #    status stays "complete", the pinned oracle quirk) ─────────────
+    fMgr = fresh_mgr()
+    def fcall(fn, *args, **kwargs):
+        cases.append({
+            "fn": f"BuildManager.{fn.__name__}",
+            "recv": {"cls": "BuildManager", "state": seed(fMgr)},
+            "args": list(args), "kwargs": kwargs,
+        })
+        fn(fMgr, *args, **kwargs)
+    fcall(bm.BuildManager.complete_verify, False, ["test_x failed"])
+    fcall(bm.BuildManager.start_integrate)
+    fcall(bm.BuildManager.complete_integrate, False, ["scope breach"])
+
+    # ── feasibility details=None branch (default {}) ─────────────────
+    dMgr = fresh_mgr()
+    cases.append({
+        "fn": "BuildManager.complete_feasible",
+        "recv": {"cls": "BuildManager", "state": seed(dMgr)},
+        "args": [True, "low"], "kwargs": {},
+    })
+
+    # ── resume scan branches (get_resume_point / can_resume) ─────────
+    # state None → None/False; empty phases → None; in_progress → found;
+    # failed → found; all complete → None.
+    def scan_case(state_dict):
+        cases.append({"fn": "BuildManager.get_resume_point",
+                      "recv": {"cls": "BuildManager", "state": state_dict},
+                      "args": [], "kwargs": {}})
+        cases.append({"fn": "BuildManager.can_resume",
+                      "recv": {"cls": "BuildManager", "state": state_dict},
+                      "args": [], "kwargs": {}})
+
+    scan_case(None)
+    scan_case(seed(fresh_mgr()))
+    p = seed(fresh_mgr())
+    p["phases"]["plan"] = "in_progress"
+    scan_case(p)
+    f = seed(fresh_mgr())
+    f["phases"]["verify"] = "failed"
+    scan_case(f)
+    scan_case(seed(mgr))  # fully walked: every phase complete
+
+    # ── get_status branches ──────────────────────────────────────────
+    cases.append({"fn": "BuildManager.get_status",
+                  "recv": {"cls": "BuildManager", "state": None},
+                  "args": [], "kwargs": {}})
+    cases.append({"fn": "BuildManager.get_status",
+                  "recv": {"cls": "BuildManager", "state": seed(mgr)},
+                  "args": [], "kwargs": {}})
+
+    # ── _load_state IO roundtrip + byte fixture ──────────────────────
+    # The fixture is the CORRECTION-CYCLE state: every dict in it is
+    # single-key or sorted-key (see BYTE-PARITY LAW) — the one state whose
+    # json.dump bytes Go can reproduce exactly. The full-walk state (with
+    # the oracle-internal feasible payload) is roundtrip-proven at the
+    # parsed level instead.
+    full_state = dataclasses.asdict(mgr.state)
+    fixture_state = dataclasses.asdict(cMgr.state)
+    with tempfile.TemporaryDirectory() as tmp:
+        bm.BUILDS_DIR = tmp  # oracle reads the module global at call time
+        build_dir = Path(tmp) / BUILD_ID
+        build_dir.mkdir(parents=True)
+        ledger_bytes = json.dumps(fixture_state, indent=2,
+                                  ensure_ascii=False).encode("utf-8")
+        (build_dir / "ledger.json").write_bytes(ledger_bytes)
+        full_bytes = json.dumps(full_state, indent=2,
+                                ensure_ascii=False).encode("utf-8")
+        (build_dir.parent / "full-walk-build").mkdir()
+        (build_dir.parent / "full-walk-build" / "ledger.json").write_bytes(
+            full_bytes)
+        m2 = bm.BuildManager.__new__(bm.BuildManager)
+        loaded = bm.BuildManager._load_state(m2, BUILD_ID)
+        assert dataclasses.asdict(loaded) == fixture_state, "roundtrip broke"
+        full_id = "full-walk-build"
+        m3 = bm.BuildManager.__new__(bm.BuildManager)
+        loaded_full = bm.BuildManager._load_state(m3, full_id)
+        assert dataclasses.asdict(loaded_full) == full_state, "full broke"
+        # missing build → FileNotFoundError("Build <id> not found")
+        try:
+            bm.BuildManager._load_state(m2, "no-such-build")
+            raise SystemExit("_load_state missing-path did NOT raise")
+        except FileNotFoundError as e:
+            assert str(e) == f"Build no-such-build not found", str(e)
+    fixture = REPO / "src/go/build-mode" / "ledger_load_fixture.json"
+    fixture.write_bytes(ledger_bytes)
+    print(f"wrote {fixture} ({len(ledger_bytes)} bytes, oracle roundtrip proven)")
+
+    truth = freeze("rune.build_mode", cases, "rune.build_mode")
+    out = REPO / "src/go/build-mode" / "truth_buildmode_gap.json"
+    write_envelope(FROZEN_NOW, truth, out)
+    print(f"wrote {out} ({len(truth)} frozen cases)")
+
+
+if __name__ == "__main__":
+    main()

scripts/gen_meaningful_events_truth.py

@@ -0,0 +1,133 @@
+#!/usr/bin/env python3
+"""Freeze the meaningful_events oracle truth (dichotomy wave, #2479).
+
+Generator law: RUNE_ME_CORE=python pinned, every input deep-copied,
+every public function exercised across ALL branches:
+  * should_fire — unknown trigger; each of the 8 stop conditions; each
+    of the 8 absent conditions; rate limiting incl. Python int()
+    truncation (fractional remaining), future last_fired, last_fired 0;
+    falsy condition/stop values (0, "", None) vs truthy
+  * fire_due — table-order listing, subsets, none due, stop-blocked
+  * mark_fired — key set on copy (original untouched), unknown trigger
+    still records, explicit now
+"""
+import copy
+import json
+import os
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "src" / "rune"))
+os.environ["RUNE_ME_CORE"] = "python"
+
+import meaningful_events as me  # noqa: E402
+
+cases_should_fire = []
+cases_fire_due = []
+cases_mark_fired = []
+
+
+def case_fire(name, trigger, state, now):
+    trigger = copy.deepcopy(trigger)
+    state = copy.deepcopy(state)
+    now = copy.deepcopy(now)
+    cases_should_fire.append({
+        "name": name,
+        "args": {"trigger": trigger, "state": state, "now": now},
+        "expect": me.should_fire(copy.deepcopy(trigger),
+                                 copy.deepcopy(state),
+                                 copy.deepcopy(now)),
+    })
+
+
+def case_due(name, state, now):
+    state = copy.deepcopy(state)
+    now = copy.deepcopy(now)
+    cases_fire_due.append({
+        "name": name,
+        "args": {"state": state, "now": now},
+        "expect": me.fire_due(copy.deepcopy(state), copy.deepcopy(now)),
+    })
+
+
+def case_mark(name, state, trigger, now):
+    state = copy.deepcopy(state)
+    trigger = copy.deepcopy(trigger)
+    now = copy.deepcopy(now)
+    st = copy.deepcopy(state)
+    expect = me.mark_fired(st, copy.deepcopy(trigger), copy.deepcopy(now))
+    cases_mark_fired.append({
+        "name": name,
+        "args": {"state": state, "trigger": trigger, "now": now},
+        "expect": expect,
+        "input_unchanged": st == state,
+    })
+
+
+# ── should_fire: every branch ───────────────────────────────────────────────
+case_fire("unknown-trigger", "nope", {}, 1000.0)
+for trig in me.TRIGGERS:
+    cond, limit, stop = me.TRIGGERS[trig]
+    case_fire(f"fires-{trig}", trig, {cond: True}, 1000.0)
+    case_fire(f"fires-with-old-last-{trig}", trig,
+              {cond: True, f"last_fired_{trig}": 0.0}, 1000.0)
+    case_fire(f"stop-met-{trig}", trig, {cond: True, stop: True}, 1000.0)
+    case_fire(f"stop-truthy-string-{trig}", trig, {cond: True, stop: "yes"}, 1000.0)
+    case_fire(f"stop-falsy-zero-not-met-{trig}", trig,
+              {cond: True, stop: 0}, 1000.0)
+    case_fire(f"condition-absent-{trig}", trig, {}, 1000.0)
+    case_fire(f"condition-falsy-empty-str-{trig}", trig, {cond: ""}, 1000.0)
+    case_fire(f"condition-falsy-null-{trig}", trig, {cond: None}, 1000.0)
+    case_fire(f"rate-limited-{trig}", trig,
+              {cond: True, f"last_fired_{trig}": 1000.0 - (limit - 100.0)}, 1000.0)
+    case_fire(f"rate-limited-exact-boundary-{trig}", trig,
+              {cond: True, f"last_fired_{trig}": 1000.0 - limit}, 1000.0)
+case_fire("rate-truncation-fractional", "task_blockers",
+          {"task_blocked": True, "last_fired_task_blockers": 400.5}, 1000.0)
+case_fire("rate-truncation-tiny-remaining", "owner_corrections",
+          {"correction_received": True, "last_fired_owner_corrections": 700.0001},
+          1000.0)
+case_fire("future-last-rate-limited", "task_blockers",
+          {"task_blocked": True, "last_fired_task_blockers": 5000.0}, 1000.0)
+case_fire("negative-now-last-fractional", "curiosity_candidates",
+          {"novel_material_present": True, "last_fired_curiosity_candidates": -100.25},
+          1000.0)
+case_fire("float-now-and-last", "repeated_failure",
+          {"same_failure_3x": True, "last_fired_repeated_failure": 123.456},
+          789.012)
+
+# ── fire_due: ordering + subsets ────────────────────────────────────────────
+ALL_DUE = {cond: True for cond, _, _ in me.TRIGGERS.values()}
+case_due("all-eight-due-table-order", ALL_DUE, 1000.0)
+case_due("none-due", {}, 1000.0)
+case_due("subset-due", {"task_blocked": True, "correction_received": 1}, 1000.0)
+case_due("stop-blocks-some", {**ALL_DUE, "task_unblocked": True}, 1000.0)
+case_due("rate-limits-block-some",
+         {cond: True for cond, _, _ in me.TRIGGERS.values()}
+         | {"last_fired_owner_corrections": 999.0}, 1000.0)
+case_due("falsy-conditions-only",
+         {cond: 0 for cond, _, _ in me.TRIGGERS.values()}, 1000.0)
+
+# ── mark_fired: copy semantics ──────────────────────────────────────────────
+case_mark("mark-explicit-now", {"task_blocked": True}, "task_blockers", 1000.0)
+case_mark("mark-unknown-trigger-still-records", {}, "mystery", 55.5)
+case_mark("mark-overwrites-existing", {"last_fired_task_blockers": 1.0},
+          "task_blockers", 2000.0)
+case_mark("mark-empty-state", {}, "owner_corrections", 0.25)
+
+truth = {
+    "module": "meaningful_events",
+    "core_pin": "RUNE_ME_CORE=python",
+    "should_fire_cases": cases_should_fire,
+    "fire_due_cases": cases_fire_due,
+    "mark_fired_cases": cases_mark_fired,
+}
+
+out = REPO / "src" / "rust" / "meaningful_events" / "frozen_truth.json"
+out.parent.mkdir(parents=True, exist_ok=True)
+out.write_text(json.dumps(truth, indent=1, sort_keys=True) + "\n",
+               encoding="utf-8")
+n = len(cases_should_fire) + len(cases_fire_due) + len(cases_mark_fired)
+print(f"frozen {n} cases -> {out}")

src/go/build-mode/build_mode.go

@@ -223,7 +223,12 @@ func StartBuildID(nowUTC string) string {
 
 // ResumePoint — reversed declaration-order scan: the first phase whose
 // status is in_progress or failed; "" when all complete (Python None).
+// Nil state → "" (the Python's `if not self.state: return None` guard,
+// pinned by truth_buildmode_gap.json).
 func ResumePoint(state *BuildState) string {
+	if state == nil {
+		return ""
+	}
 	for i := len(Phases) - 1; i >= 0; i-- {
 		status, has := state.Phases[Phases[i]]
 		if !has {

src/go/build-mode/build_mode_gap.go

@@ -0,0 +1,185 @@
+package main
+
+// build_mode_gap.go — gap-closing batch: the PARTIAL audit's missing fns of
+// src/rune/build_mode.py (contract BUILDER-2026-10-13T02:02:02Z-dad27f-
+// factory-gap-bm). _load_state/_save_state (the ledger IO seam, buildsDir
+// INJECTED — the Python reads a module global), assess_complexity, and the
+// full start_* phase-transition family. Phase names and ordering are
+// load-bearing for the pipeline: every transition here is frozen
+// transition-by-transition in declaration order in
+// truth_buildmode_gap.json (live oracle, clock pinned via the shared
+// freeze_truth harness).
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"sort"
+)
+
+// LoadState — _load_state: read BUILDS_DIR/<build_id>/ledger.json into a
+// BuildState. Missing file → the Python FileNotFoundError text.
+func LoadState(buildsDir, buildID string) (*BuildState, error) {
+	ledgerPath := filepath.Join(buildsDir, buildID, "ledger.json")
+	if _, err := os.Stat(ledgerPath); err != nil {
+		return nil, fmt.Errorf("Build %s not found", buildID)
+	}
+	data, err := os.ReadFile(ledgerPath)
+	if err != nil {
+		return nil, err
+	}
+	var state BuildState
+	if err := json.Unmarshal(data, &state); err != nil {
+		return nil, err
+	}
+	return &state, nil
+}
+
+// SaveState — _save_state: write the state as asdict-shape JSON (indent 2,
+// no HTML escaping, no trailing newline — Python json.dump byte contract,
+// pinned by ledger_load_fixture.json). Nil state → no-op (the Python's
+// `if not self.state: return`).
+//
+// BYTE PARITY TRAP: Python dict preserves INSERTION order and json.dump
+// emits it; Go maps marshal sorted. The live manager creates the phases
+// keys via _log_phase, so the ledger's first-occurrence order IS the dict
+// order — the phases object is spliced in that order (keys missing from
+// the ledger fall back to pipeline declaration order). The nil sentinel
+// replaces exactly the one top-level "phases" key.
+func SaveState(state *BuildState, buildsDir string) error {
+	if state == nil {
+		return nil
+	}
+	buildDir := filepath.Join(buildsDir, state.BuildID)
+	if err := os.MkdirAll(buildDir, 0o755); err != nil {
+		return err
+	}
+	data, err := stateJSON(state)
+	if err != nil {
+		return err
+	}
+	return os.WriteFile(filepath.Join(buildDir, "ledger.json"), data, 0o644)
+}
+
+// stateJSON — the Python json.dump byte contract (indent 2, no HTML
+// escaping, no trailing newline, phases in dict-insertion order).
+func stateJSON(state *BuildState) ([]byte, error) {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+	enc.SetEscapeHTML(false)
+	enc.SetIndent("", "  ")
+	saved := state.Phases
+	state.Phases = nil
+	err := enc.Encode(state)
+	state.Phases = saved
+	if err != nil {
+		return nil, err
+	}
+	out := bytes.Replace(buf.Bytes(), []byte(`"phases": null`),
+		orderedPhasesJSON(state), 1)
+	return bytes.TrimRight(out, "\n"), nil
+}
+
+// orderedPhasesJSON — the phases object in Python-dict insertion order
+// (ledger first-occurrence, then declaration order, then sorted rest).
+func orderedPhasesJSON(state *BuildState) []byte {
+	order := make([]string, 0, len(state.Phases))
+	seen := make(map[string]bool, len(state.Phases))
+	for _, e := range state.Ledger {
+		if _, has := state.Phases[e.Phase]; has && !seen[e.Phase] {
+			seen[e.Phase] = true
+			order = append(order, e.Phase)
+		}
+	}
+	for _, p := range Phases {
+		if _, has := state.Phases[p]; has && !seen[p] {
+			seen[p] = true
+			order = append(order, p)
+		}
+	}
+	rest := make([]string, 0)
+	for p := range state.Phases {
+		if !seen[p] {
+			rest = append(rest, p)
+		}
+	}
+	sort.Strings(rest)
+	order = append(order, rest...)
+
+	var b bytes.Buffer
+	// Python json.dump(indent=2) nests the phases entries one level deeper
+	// (4-space entries, 2-space closing brace).
+	b.WriteString(`"phases": {`)
+	for i, p := range order {
+		if i > 0 {
+			b.WriteByte(',')
+		}
+		b.WriteString("\n    ")
+		k, _ := json.Marshal(p)
+		v, _ := json.Marshal(state.Phases[p])
+		b.Write(k)
+		b.WriteString(": ")
+		b.Write(v)
+	}
+	if len(order) > 0 {
+		b.WriteString("\n  ")
+	}
+	b.WriteString("}")
+	return b.Bytes()
+}
+
+// AssessComplexity — assess_complexity: the placeholder report, verbatim.
+func AssessComplexity() map[string]interface{} {
+	return map[string]interface{}{
+		"complexity":          "complex", // simple/medium/complex
+		"estimated_steps":     10,
+		"requires_build_mode": true,
+	}
+}
+
+// StartResearch — start_research.
+func StartResearch(state *BuildState, nowISO string) {
+	LogPhase(state, "research", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartFeasible — start_feasible.
+func StartFeasible(state *BuildState, nowISO string) {
+	LogPhase(state, "feasible", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartConsult — start_consult.
+func StartConsult(state *BuildState, nowISO string) {
+	LogPhase(state, "consult", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartConsent — start_consent.
+func StartConsent(state *BuildState, nowISO string) {
+	LogPhase(state, "consent", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartPlan — start_plan.
+func StartPlan(state *BuildState, nowISO string) {
+	LogPhase(state, "plan", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartBuildPhase — start_build_phase.
+func StartBuildPhase(state *BuildState, nowISO string) {
+	LogPhase(state, "build", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartVerify — start_verify.
+func StartVerify(state *BuildState, nowISO string) {
+	LogPhase(state, "verify", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartIntegrate — start_integrate.
+func StartIntegrate(state *BuildState, nowISO string) {
+	LogPhase(state, "integrate", StatusInProgress, nil, nil, nowISO)
+}
+
+// StartRecord — start_record.
+func StartRecord(state *BuildState, nowISO string) {
+	LogPhase(state, "record", StatusInProgress, nil, nil, nowISO)
+}

src/go/build-mode/ledger_load_fixture.json

@@ -0,0 +1,55 @@
+{
+  "build_id": "20261013-022000",
+  "description": "gap battery build",
+  "created_at": "2026-10-13T02:20:00+00:00",
+  "worktree_path": null,
+  "current_phase": "assess",
+  "phases": {
+    "correct": "complete"
+  },
+  "ledger": [
+    {
+      "phase": "correct",
+      "status": "in_progress",
+      "timestamp": "2026-10-13T02:20:00+00:00",
+      "details": {},
+      "errors": []
+    },
+    {
+      "phase": "correct",
+      "status": "complete",
+      "timestamp": "2026-10-13T02:20:00+00:00",
+      "details": {
+        "fixed": false
+      },
+      "errors": []
+    },
+    {
+      "phase": "correct",
+      "status": "in_progress",
+      "timestamp": "2026-10-13T02:20:00+00:00",
+      "details": {},
+      "errors": []
+    },
+    {
+      "phase": "correct",
+      "status": "complete",
+      "timestamp": "2026-10-13T02:20:00+00:00",
+      "details": {
+        "fixed": true
+      },
+      "errors": []
+    }
+  ],
+  "batches": [],
+  "corrections": 3,
+  "max_corrections": 3,
+  "current_build_phase": "foundation",
+  "trajectory": [],
+  "failure_patterns": [],
+  "complexity_score": 0,
+  "estimated_steps": 0,
+  "actual_steps": 0,
+  "files_changed": [],
+  "contracts": []
+}

src/go/build-mode/truth_buildmode_gap_test.go

@@ -0,0 +1,257 @@
+package main
+
+// truth_buildmode_gap_test.go — gap-closing parity gate: replays
+// truth_buildmode_gap.json VERBATIM (go:embed — the shared freeze_truth
+// harness output, live oracle, clock pinned) through the ported functions.
+// Every case carries a recv receiver (the PRE state), the call args, and
+// the expected post-call state + result; the Go replay must match
+// byte-for-byte after canonical JSON. Contract
+// BUILDER-2026-10-13T02:02:02Z-dad27f-factory-gap-bm.
+
+import (
+	_ "embed"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+)
+
+//go:embed truth_buildmode_gap.json
+var gapTruth []byte
+
+//go:embed ledger_load_fixture.json
+var ledgerFixture []byte
+
+type gapEnvelope struct {
+	FrozenNow string `json:"frozen_now"`
+	Cases     []struct {
+		Fn   string `json:"fn"`
+		Recv struct {
+			Cls   string          `json:"cls"`
+			State json.RawMessage `json:"state"`
+		} `json:"recv"`
+		Args     []interface{} `json:"args"`
+		Expected struct {
+			Result json.RawMessage `json:"result"`
+			State  json.RawMessage `json:"state"`
+		} `json:"expected"`
+	} `json:"cases"`
+}
+
+func TestBuildModeGapParity(t *testing.T) {
+	var doc gapEnvelope
+	if err := json.Unmarshal(gapTruth, &doc); err != nil {
+		t.Fatalf("unparseable: %v", err)
+	}
+	stamp := doc.FrozenNow + "+00:00"
+	for i, c := range doc.Cases {
+		name := c.Fn
+		t.Run(name, func(t *testing.T) {
+			// Rebuild the receiver: recv.state null → nil state.
+			var state *BuildState
+			if string(c.Recv.State) != "null" && len(c.Recv.State) > 0 {
+				state = &BuildState{}
+				if err := json.Unmarshal(c.Recv.State, state); err != nil {
+					t.Fatalf("recv state unparseable: %v", err)
+				}
+			}
+			got, gotState := dispatchGap(t, state, c.Fn, c.Args, stamp)
+			canonEq(t, "result", got, c.Expected.Result)
+			// state: absent/null expected → no state attr or nil state;
+			// else canonical compare.
+			if len(c.Expected.State) == 0 || string(c.Expected.State) == "null" {
+				if len(c.Expected.State) == 0 {
+					return // receiver never had the attr — nothing to compare
+				}
+				if gotState != nil {
+					t.Fatalf("expected nil state, got one")
+				}
+			} else {
+				canonEq(t, "state", gotState, c.Expected.State)
+			}
+		})
+		_ = i
+	}
+}
+
+// dispatchGap — replay one frozen call through the ported function; return
+// the JSON-comparable result and the post-call state pointer.
+func dispatchGap(t *testing.T, state *BuildState, fn string, args []interface{}, stamp string) (result interface{}, outState *BuildState) {
+	t.Helper()
+	name := strings.TrimPrefix(fn, "BuildManager.")
+	arg := func(i int) interface{} {
+		if i < len(args) {
+			return args[i]
+		}
+		return nil
+	}
+	switch name {
+	case "assess_complexity":
+		return AssessComplexity(), state
+	case "complete_assess":
+		CompleteAssess(state, arg(0).(map[string]interface{}), stamp)
+	case "start_research":
+		StartResearch(state, stamp)
+	case "complete_research":
+		CompleteResearch(state, arg(0).(map[string]interface{}), stamp)
+	case "start_feasible":
+		StartFeasible(state, stamp)
+	case "complete_feasible":
+		CompleteFeasible(state, arg(0) == true, arg(1).(string),
+			argMap(arg(2)), stamp)
+	case "start_consult":
+		StartConsult(state, stamp)
+	case "complete_consult":
+		CompleteConsult(state, strList(arg(0)), stamp)
+	case "start_consent":
+		StartConsent(state, stamp)
+	case "complete_consent":
+		CompleteConsent(state, arg(0) == true, stamp)
+	case "start_plan":
+		StartPlan(state, stamp)
+	case "complete_plan":
+		CompletePlan(state, objList(arg(0)), stamp)
+	case "start_build_phase":
+		StartBuildPhase(state, stamp)
+	case "complete_build_phase":
+		CompleteBuildPhase(state, objList(arg(0)), stamp)
+	case "start_verify":
+		StartVerify(state, stamp)
+	case "complete_verify":
+		CompleteVerify(state, arg(0) == true, strList(arg(1)), stamp)
+	case "start_correct":
+		StartCorrect(state, stamp)
+	case "complete_correct":
+		CompleteCorrect(state, arg(0) == true, stamp)
+	case "start_integrate":
+		StartIntegrate(state, stamp)
+	case "complete_integrate":
+		CompleteIntegrate(state, arg(0) == true, strList(arg(1)), stamp)
+	case "start_record":
+		StartRecord(state, stamp)
+	case "complete_record":
+		CompleteRecord(state, stamp)
+	case "_save_state":
+		dir := t.TempDir()
+		if err := SaveState(state, dir); err != nil {
+			t.Fatalf("save: %v", err)
+		}
+		return nil, state
+	case "get_resume_point":
+		rp := ResumePoint(state)
+		if rp == "" {
+			return nil, state // Python None
+		}
+		return rp, state
+	case "can_resume":
+		return CanResume(state), state
+	case "get_status":
+		return BuildStatus(state), state
+	default:
+		t.Fatalf("unknown frozen fn %q", fn)
+	}
+	return nil, state
+}
+
+func argMap(v interface{}) map[string]interface{} {
+	m, _ := v.(map[string]interface{})
+	return m
+}
+
+// canonEq — canonical-JSON comparison (Go-built values vs the frozen
+// oracle snapshot).
+func canonEq(t *testing.T, what string, got interface{}, wantRaw json.RawMessage) {
+	t.Helper()
+	gb, err := json.Marshal(got)
+	if err != nil {
+		t.Fatalf("marshal got %s: %v", what, err)
+	}
+	var gotV, wantV interface{}
+	if err := json.Unmarshal(gb, &gotV); err != nil {
+		t.Fatalf("reparse got %s: %v", what, err)
+	}
+	if err := json.Unmarshal(wantRaw, &wantV); err != nil {
+		t.Fatalf("reparse want %s: %v", what, err)
+	}
+	gc, _ := json.Marshal(gotV)
+	wc, _ := json.Marshal(wantV)
+	if string(gc) != string(wc) {
+		t.Fatalf("%s divergence:\n got %s\nwant %s", what, gc, wc)
+	}
+}
+
+// TestLoadStateFixture — the IO seam: a build whose ledger.json holds the
+// oracle's exact save bytes must load into the identical state.
+func TestLoadStateFixture(t *testing.T) {
+	dir := t.TempDir()
+	buildDir := filepath.Join(dir, "20261013-022000")
+	if err := os.MkdirAll(buildDir, 0o755); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(filepath.Join(buildDir, "ledger.json"), ledgerFixture, 0o644); err != nil {
+		t.Fatal(err)
+	}
+	state, err := LoadState(dir, "20261013-022000")
+	if err != nil {
+		t.Fatalf("load: %v", err)
+	}
+	var want BuildState
+	if err := json.Unmarshal(ledgerFixture, &want); err != nil {
+		t.Fatal(err)
+	}
+	canonEq(t, "loadstate", state, ledgerFixture)
+
+	if _, err := LoadState(dir, "no-such-build"); err == nil ||
+		err.Error() != "Build no-such-build not found" {
+		t.Fatalf("missing-build error: got %v", err)
+	}
+}
+
+// TestSaveStateBytes — the byte contract: Python json.dump(asdict(state),
+// indent=2) — the fixture IS those bytes; SaveState must reproduce them.
+func TestSaveStateBytes(t *testing.T) {
+	var state BuildState
+	if err := json.Unmarshal(ledgerFixture, &state); err != nil {
+		t.Fatal(err)
+	}
+	dir := t.TempDir()
+	if err := SaveState(&state, dir); err != nil {
+		t.Fatal(err)
+	}
+	got, err := os.ReadFile(filepath.Join(dir, "20261013-022000", "ledger.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(got) != string(ledgerFixture) {
+		t.Fatalf("save bytes diverge from oracle fixture:\n got %d bytes\nwant %d bytes",
+			len(got), len(ledgerFixture))
+	}
+	// nil state → no-op, nothing written (fresh dir: nothing created).
+	nilDir := t.TempDir()
+	if err := SaveState(nil, nilDir); err != nil {
+		t.Fatalf("nil save: %v", err)
+	}
+	entries, _ := os.ReadDir(nilDir)
+	if len(entries) != 0 {
+		t.Fatalf("nil state must not create the build dir, found %d entries", len(entries))
+	}
+}
+
+// TestSaveStateRoundTrip — the phases splice must never corrupt content:
+// save the fixture state, load it back, require canonical equality.
+func TestSaveStateRoundTrip(t *testing.T) {
+	var state BuildState
+	if err := json.Unmarshal(ledgerFixture, &state); err != nil {
+		t.Fatal(err)
+	}
+	dir := t.TempDir()
+	if err := SaveState(&state, dir); err != nil {
+		t.Fatal(err)
+	}
+	back, err := LoadState(dir, state.BuildID)
+	if err != nil {
+		t.Fatal(err)
+	}
+	canonEq(t, "roundtrip", back, ledgerFixture)
+}

src/rune/__init__.py

@@ -6,7 +6,7 @@ This is the main entry point for the Rune program.
 import json
 from pathlib import Path
 
-__version__ = "0.262.0"
+__version__ = "0.264.0"
 
 # One build identity (handoff 2026-09-10 §4): the semantic version above is
 # the repo-level truth (auto-bumped at merge); the DEPLOYED revision (exact

src/rune/canary_gitea_automation.py

@@ -233,7 +233,7 @@ def main() -> int:
     )
     check(
         "Rune minor version advertises provisioning and restore verification",
-        '__version__ = "0.262.0"' in rune_version,
+        '__version__ = "0.264.0"' in rune_version,
         failures,
     )
     # #169 policy update: auto-bump on merge is sanctioned (Nick directive:

src/rune/me_loader.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python3
+"""ctypes loader for the Rust meaningful_events core (dichotomy wave,
+#2479 tranche). Mirrors broker_loader.py's contract:
+
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * me_should_fire({trigger, state, now}) -> {fires, reason}
+  * me_fire_due({state, now}) -> [trigger, ...]
+  * me_mark_fired({state, trigger, now}) -> state dict
+
+The wall CLOCK is host-side by nature (now resolved by the facade —
+null never crosses); the kernel owns the trigger table and the laws.
+Free: capability_scoring's cs_free_string (shared process). Override the
+.so with ME_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" / "meaningful_events"
+_LIB_PATH = Path(
+    os.environ.get("ME_RUST_LIB", _CRATE_DIR / "target" / "release" / "libmeaningful_events.so")
+)
+
+_ERR_TYPES = {
+    "ValueError": ValueError,
+    "KeyError": KeyError,
+    "AttributeError": AttributeError,
+    "TypeError": TypeError,
+}
+
+_lib = None
+
+
+class EventsCoreUnavailable(FileNotFoundError):
+    """The Rust meaningful-events core is not built. Callers fall back
+    to Python."""
+
+
+def available() -> bool:
+    return _LIB_PATH.is_file()
+
+
+def _load() -> ctypes.CDLL:
+    global _lib
+    if _lib is not None:
+        return _lib
+    if not _LIB_PATH.is_file():
+        raise EventsCoreUnavailable(str(_LIB_PATH))
+    lib = ctypes.CDLL(str(_LIB_PATH))
+    for sym in ("me_should_fire", "me_fire_due", "me_mark_fired"):
+        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 should_fire(trigger: str, state: dict, now: float) -> dict:
+    return _call("me_should_fire", {"trigger": trigger, "state": state,
+                                    "now": now})
+
+
+def fire_due(state: dict, now: float) -> list:
+    return _call("me_fire_due", {"state": state, "now": now})
+
+
+def mark_fired(state: dict, trigger: str, now: float) -> dict:
+    return _call("me_mark_fired", {"state": state, "trigger": trigger,
+                                   "now": now})

src/rune/meaningful_events.py

@@ -15,6 +15,7 @@ approaching, not because a timer fired.
 
 from __future__ import annotations
 
+import os
 import time
 
 # Trigger classes with (condition_key, default_rate_limit_s, stop_key)
@@ -30,10 +31,36 @@ TRIGGERS = {
 }
 
 
+def _kernel():
+    """The Rust meaningful-events core (me_loader) when RUNE_ME_CORE
+    allows it, else None — the trigger table, the should_fire law, the
+    fire_due ordering and the mark_fired copy law delegate there first
+    and fall back to the Python body below (same semantics). The wall
+    CLOCK is host-side by nature: `now` is resolved here (never null
+    across FFI), and an INT now keeps the Python floor in mark_fired
+    (the oracle stores the value as-is; the kernel wire is float)."""
+    if os.environ.get("RUNE_ME_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import me_loader
+
+        if me_loader.available():
+            return me_loader
+    except Exception:  # noqa: BLE001 — kernel failure → Python floor
+        return None
+    return None
+
+
 def should_fire(trigger: str, state: dict, now: float | None = None) -> dict:
     """Evaluate one trigger against current state. Returns
     {fires, reason}. Rate-limited, stop-conditioned."""
     now = now if now is not None else time.time()
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.should_fire(trigger, state, now)
+        except Exception:  # noqa: BLE001 — kernel floor
+            pass
     if trigger not in TRIGGERS:
         return {"fires": False, "reason": f"unknown trigger: {trigger}"}
     cond_key, rate_limit, stop_key = TRIGGERS[trigger]
@@ -50,12 +77,26 @@ def should_fire(trigger: str, state: dict, now: float | None = None) -> dict:
 
 def fire_due(state: dict, now: float | None = None) -> list:
     """All triggers due right now (the endogenous cognition tick)."""
+    resolved = now if now is not None else time.time()
+    k = _kernel()
+    if k is not None:
+        try:
+            return k.fire_due(state, resolved)
+        except Exception:  # noqa: BLE001 — kernel floor
+            pass
     return [t for t in TRIGGERS
-            if should_fire(t, state, now)["fires"]]
+            if should_fire(t, state, resolved)["fires"]]
 
 
 def mark_fired(state: dict, trigger: str, now: float | None = None) -> dict:
     """Record a firing (for rate limiting). Returns the updated state."""
+    k = _kernel()
+    if k is not None and (now is None or isinstance(now, float)):
+        try:
+            return k.mark_fired(state, trigger,
+                                now if now is not None else time.time())
+        except Exception:  # noqa: BLE001 — kernel floor
+            pass
     state = dict(state)
     state[f"last_fired_{trigger}"] = now if now is not None else time.time()
     return state

src/rune/rune_tui.py

@@ -8,7 +8,7 @@ mode system, subject header, shared memory and identity.
 
 This is not an interface. This is how I see, speak, and interact with the world.
 
-Version: v0.262.0
+Version: v0.264.0
 Features: 41 (24 original + 17 new native commands)
 
 Usage:
@@ -2037,7 +2037,7 @@ class RuneApp(App):
         # Show version info
         self.chat_channel.add_message(
             "system",
-            f"[dim]Rune v0.262.0 | Textual TUI | {len(loaded)} native modules[/dim]",
+            f"[dim]Rune v0.264.0 | Textual TUI | {len(loaded)} native modules[/dim]",
         )
 
         # Auto-show night mode suggestion if evening/night hours

src/rust/meaningful_events/Cargo.lock

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

src/rust/meaningful_events/Cargo.toml

@@ -0,0 +1,11 @@
+[package]
+name = "meaningful_events"
+version = "0.1.0"
+edition = "2021"
+description = "FULL parity port of src/rune/meaningful_events.py (dichotomy wave, #2479 tranche)"
+
+[lib]
+crate-type = ["rlib", "cdylib", "staticlib"]
+
+[dependencies]
+evidence_provenance = { path = "../evidence_provenance" }

src/rust/meaningful_events/src/branch_battery.rs

@@ -0,0 +1,153 @@
+//! branch_battery — replay the FROZEN oracle truth VERBATIM
+//! (frozen_truth.json, generated by scripts/gen_meaningful_events_truth.py
+//! against the LIVE Python oracle with RUNE_ME_CORE=python pinned and
+//! deep-copied inputs).
+
+use crate::{fire_due, mark_fired, should_fire};
+use evidence_provenance::pyjson::{self, Json};
+
+const TRUTH: &str = include_str!("../frozen_truth.json");
+
+fn truth() -> Json {
+    pyjson::parse(TRUTH).expect("frozen truth parses")
+}
+
+fn cases(group: &str) -> Vec<Json> {
+    match truth().get(group) {
+        Some(Json::Arr(items)) => items.clone(),
+        _ => panic!("truth group {group} missing"),
+    }
+}
+
+fn at<'a>(v: &'a Json, path: &[&str]) -> Option<&'a Json> {
+    let mut cur = v;
+    for k in path {
+        cur = match cur {
+            Json::Obj(pairs) => pairs
+                .iter()
+                .find(|(k2, _)| k2 == k)
+                .map(|(_, x)| x)?,
+            _ => return None,
+        };
+    }
+    Some(cur)
+}
+
+fn jstr(v: Option<&Json>) -> String {
+    match v {
+        Some(Json::Str(s)) => s.clone(),
+        _ => String::new(),
+    }
+}
+
+fn now_of(v: Option<&Json>) -> Option<f64> {
+    match v {
+        Some(Json::Num(n)) | Some(Json::Float(n)) => Some(*n),
+        _ => None,
+    }
+}
+
+/// Python-value equality on Json (order-insensitive objects).
+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)) | (Json::Float(x), Json::Float(y)) => x == y,
+        (Json::Float(x), Json::Num(y)) | (Json::Num(x), Json::Float(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_case(name: &str, want: &Json, got: &Json) {
+    assert!(
+        peq(want, got),
+        "case {name} diverged:\n want={}\n  got={}",
+        pyjson::dumps(want),
+        pyjson::dumps(got),
+    );
+}
+
+#[test]
+fn replay_should_fire() {
+    for c in cases("should_fire_cases") {
+        let got = should_fire(
+            &jstr(at(&c, &["args", "trigger"])),
+            at(&c, &["args", "state"]).expect("state"),
+            now_of(at(&c, &["args", "now"])),
+        );
+        assert_case(
+            &jstr(at(&c, &["name"])),
+            at(&c, &["expect"]).expect("expect"),
+            &got,
+        );
+    }
+}
+
+#[test]
+fn replay_fire_due() {
+    for c in cases("fire_due_cases") {
+        let got = fire_due(
+            at(&c, &["args", "state"]).expect("state"),
+            now_of(at(&c, &["args", "now"])),
+        );
+        assert_case(
+            &jstr(at(&c, &["name"])),
+            at(&c, &["expect"]).expect("expect"),
+            &got,
+        );
+    }
+}
+
+#[test]
+fn replay_mark_fired() {
+    for c in cases("mark_fired_cases") {
+        let got = mark_fired(
+            at(&c, &["args", "state"]).expect("state"),
+            &jstr(at(&c, &["args", "trigger"])),
+            now_of(at(&c, &["args", "now"])),
+        );
+        assert_case(
+            &jstr(at(&c, &["name"])),
+            at(&c, &["expect"]).expect("expect"),
+            &got,
+        );
+        // the copy law: the kernel takes &Json and cannot mutate, but
+        // the truth ALSO records the oracle's input-unchanged invariant
+        assert!(
+            matches!(at(&c, &["input_unchanged"]), Some(Json::Bool(true))),
+            "case {} oracle input was mutated",
+            jstr(at(&c, &["name"])),
+        );
+    }
+}
+
+#[test]
+fn all_eight_triggers_in_table_order() {
+    let names: Vec<&str> = crate::TRIGGERS.iter().map(|t| t.name).collect();
+    assert_eq!(
+        names,
+        vec![
+            "unresolved_commitments",
+            "approaching_deadlines",
+            "repeated_failure",
+            "contradictions",
+            "curiosity_candidates",
+            "prediction_mismatches",
+            "task_blockers",
+            "owner_corrections",
+        ]
+    );
+}

src/rust/meaningful_events/src/lib.rs

@@ -0,0 +1,268 @@
+//! meaningful_events — endogenous meaningful-event triggers (#48).
+//!
+//! FULL parity port of src/rune/meaningful_events.py (dichotomy wave,
+//! #2479 tranche). The wall CLOCK is host-side by nature — `now` is
+//! passed per op (null never reaches the kernel; the facade resolves
+//! time.time() first). The kernel owns the eight-trigger table, the
+//! should_fire law, the fire_due ordering and the mark_fired copy law.
+//!
+//! FFI (doc-envelope law): every call takes one JSON doc and returns
+//! {"ok": true, "value": ...} | {"ok": false, "err": ..., "type": ...}.
+//!   * me_should_fire({trigger, state, now})   -> {fires, reason}
+//!   * me_fire_due({state, now})               -> [trigger, ...]
+//!   * me_mark_fired({state, trigger, now})    -> state dict
+//!
+//! Python laws replicated exactly:
+//!   * should_fire ordering: unknown → stop → absent → rate → fires
+//!   * truthiness on state values (0, "", null are ABSENT/NOT-MET)
+//!   * last = state["last_fired_{trigger}"] or 0 (falsy → 0)
+//!   * rate gate is STRICTLY `now - last < rate_limit` (equal fires)
+//!   * remaining = int(rate_limit - (now - last)) — Python int()
+//!     truncation TOWARD ZERO (matches `as i64` for this domain)
+//!   * fire_due lists TABLE ORDER (insertion order, not sorted)
+//!   * mark_fired: dict(state) shallow copy + key set (unknown
+//!     triggers still record)
+//!
+//! NOTE: no me_free_string export — callers free through capability_
+//! scoring's cs_free_string (shared process allocator).
+
+use evidence_provenance::pyjson::{self, Json};
+use std::ffi::{CStr, CString};
+use std::os::raw::c_char;
+
+fn obj(pairs: Vec<(&str, Json)>) -> Json {
+    Json::Obj(pairs.into_iter().map(|(k, v)| (k.to_string(), v)).collect())
+}
+
+fn s(v: &str) -> Json {
+    Json::Str(v.to_string())
+}
+
+/// Phase B groundwork for the language-advantage pass: the trigger
+/// table is typed records, not tuples — the compiler names every field.
+pub struct Trigger {
+    pub name: &'static str,
+    pub condition_key: &'static str,
+    pub rate_limit_s: f64,
+    pub stop_key: &'static str,
+}
+
+/// The eight trigger classes, TABLE ORDER (the fire_due listing law).
+pub const TRIGGERS: [Trigger; 8] = [
+    Trigger { name: "unresolved_commitments", condition_key: "has_open_commitments", rate_limit_s: 3600.0, stop_key: "commitments_resolved" },
+    Trigger { name: "approaching_deadlines", condition_key: "deadline_within_24h", rate_limit_s: 1800.0, stop_key: "deadline_passed_or_met" },
+    Trigger { name: "repeated_failure", condition_key: "same_failure_3x", rate_limit_s: 900.0, stop_key: "failure_pattern_broken" },
+    Trigger { name: "contradictions", condition_key: "contradiction_detected", rate_limit_s: 1800.0, stop_key: "contradiction_resolved" },
+    Trigger { name: "curiosity_candidates", condition_key: "novel_material_present", rate_limit_s: 7200.0, stop_key: "material_absorbed" },
+    Trigger { name: "prediction_mismatches", condition_key: "prediction_wrong", rate_limit_s: 1800.0, stop_key: "model_updated" },
+    Trigger { name: "task_blockers", condition_key: "task_blocked", rate_limit_s: 600.0, stop_key: "task_unblocked" },
+    Trigger { name: "owner_corrections", condition_key: "correction_received", rate_limit_s: 300.0, stop_key: "correction_absorbed" },
+];
+
+fn find(trigger: &str) -> Option<&'static Trigger> {
+    TRIGGERS.iter().find(|t| t.name == trigger)
+}
+
+/// Python truthiness on a JSON value (state.get() laws).
+fn truthy(v: Option<&Json>) -> bool {
+    match v {
+        None | Some(Json::Null) => false,
+        Some(Json::Bool(b)) => *b,
+        Some(Json::Num(n)) => *n != 0.0,
+        Some(Json::Float(n)) => *n != 0.0,
+        Some(Json::Str(x)) => !x.is_empty(),
+        Some(Json::Arr(a)) => !a.is_empty(),
+        Some(Json::Obj(o)) => !o.is_empty(),
+    }
+}
+
+/// as_f64 over state/now values (Num and Float alike).
+fn f64_of(v: Option<&Json>) -> Option<f64> {
+    match v {
+        Some(Json::Num(n)) | Some(Json::Float(n)) => Some(*n),
+        _ => None,
+    }
+}
+
+/// Phase B (language advantage): the decision is a typed verdict
+/// matched EXHAUSTIVELY — the JSON envelope is built once at the edge,
+/// and fire_due filters on the enum instead of re-parsing decision
+/// JSON per trigger.
+#[derive(Debug, Clone, Copy, PartialEq)]
+enum FireDecision {
+    /// unreachable through should_fire (find() rules it out) but kept
+    /// so the match stays exhaustive over the decision lattice
+    #[allow(dead_code)]
+    Unknown,
+    Stopped(&'static str),
+    Absent(&'static str),
+    RateLimited(i64),
+    Fire(&'static str),
+}
+
+fn decide(t: &'static Trigger, state: &Json, now: f64) -> FireDecision {
+    if truthy(state.get(t.stop_key)) {
+        return FireDecision::Stopped(t.stop_key);
+    }
+    if !truthy(state.get(t.condition_key)) {
+        return FireDecision::Absent(t.condition_key);
+    }
+    // last = state.get(f"last_fired_{trigger}", 0) — falsy → 0
+    let last = f64_of(state.get(&format!("last_fired_{}", t.name))).unwrap_or(0.0);
+    if now - last < t.rate_limit_s {
+        return FireDecision::RateLimited((t.rate_limit_s - (now - last)) as i64);
+    }
+    FireDecision::Fire(t.condition_key)
+}
+
+/// should_fire law. `now` is always Some (facade-resolved clock).
+pub fn should_fire(trigger: &str, state: &Json, now: Option<f64>) -> Json {
+    let now = now.unwrap_or(0.0);
+    let Some(t) = find(trigger) else {
+        return obj(vec![
+            ("fires", Json::Bool(false)),
+            ("reason", s(&format!("unknown trigger: {trigger}"))),
+        ]);
+    };
+    match decide(t, state, now) {
+        FireDecision::Stopped(stop_key) => obj(vec![
+            ("fires", Json::Bool(false)),
+            ("reason", s(&format!("stop condition met: {stop_key}"))),
+        ]),
+        FireDecision::Absent(cond_key) => obj(vec![
+            ("fires", Json::Bool(false)),
+            ("reason", s(&format!("condition absent: {cond_key}"))),
+        ]),
+        FireDecision::RateLimited(remaining) => obj(vec![
+            ("fires", Json::Bool(false)),
+            ("reason", s(&format!("rate-limited ({remaining}s left)"))),
+        ]),
+        FireDecision::Fire(cond_key) => obj(vec![
+            ("fires", Json::Bool(true)),
+            ("reason", s(&format!("condition holds: {cond_key}"))),
+        ]),
+        FireDecision::Unknown => unreachable!("find() already ruled this out"),
+    }
+}
+
+/// fire_due law: TABLE ORDER listing of due triggers.
+pub fn fire_due(state: &Json, now: Option<f64>) -> Json {
+    let now = now.unwrap_or(0.0);
+    let due: Vec<Json> = TRIGGERS
+        .iter()
+        .filter(|t| matches!(decide(t, state, now), FireDecision::Fire(_)))
+        .map(|t| s(t.name))
+        .collect();
+    Json::Arr(due)
+}
+
+/// mark_fired law: dict(state) shallow copy + `last_fired_{trigger}`.
+pub fn mark_fired(state: &Json, trigger: &str, now: Option<f64>) -> Json {
+    let mut pairs: Vec<(String, Json)> = match state {
+        Json::Obj(pairs) => pairs.clone(),
+        _ => Vec::new(),
+    };
+    let stamp = Json::Float(now.unwrap_or(0.0));
+    let key = format!("last_fired_{trigger}");
+    if let Some(slot) = pairs.iter_mut().find(|(k, _)| *k == key) {
+        slot.1 = stamp;
+    } else {
+        pairs.push((key, stamp));
+    }
+    Json::Obj(pairs)
+}
+
+// ── 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()))
+}
+
+fn now_of(doc: &Json) -> Result<Option<f64>, (String, String)> {
+    match doc.get("now") {
+        None | Some(Json::Null) => Ok(None),
+        Some(v) => f64_of(Some(v))
+            .map(Some)
+            .ok_or_else(|| ("now must be a number or null".into(), "TypeError".into())),
+    }
+}
+
+fn state_of(doc: &Json) -> Result<&Json, (String, String)> {
+    match doc.get("state") {
+        Some(st @ Json::Obj(_)) => Ok(st),
+        _ => Err(("state must be an object".into(), "TypeError".into())),
+    }
+}
+
+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)
+        }
+    };
+}
+
+ffi!(me_should_fire, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let trigger = match doc.get("trigger") {
+        Some(Json::Str(t)) => t.clone(),
+        _ => return Err(("trigger must be a string".into(), "TypeError".into())),
+    };
+    let state = state_of(&doc)?.clone();
+    let now = now_of(&doc)?;
+    Ok(should_fire(&trigger, &state, now))
+});
+
+ffi!(me_fire_due, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let state = state_of(&doc)?.clone();
+    let now = now_of(&doc)?;
+    Ok(fire_due(&state, now))
+});
+
+ffi!(me_mark_fired, |raw: &str| -> Result<Json, (String, String)> {
+    let doc = parse_doc(raw)?;
+    let state = state_of(&doc)?.clone();
+    let trigger = match doc.get("trigger") {
+        Some(Json::Str(t)) => t.clone(),
+        _ => return Err(("trigger must be a string".into(), "TypeError".into())),
+    };
+    let now = now_of(&doc)?;
+    Ok(mark_fired(&state, &trigger, now))
+});
+
+#[cfg(test)]
+mod branch_battery;

tests/test_meaningful_events_port.py

@@ -0,0 +1,93 @@
+"""Tests: meaningful_events.py Rust core port (dichotomy wave, #2479).
+
+Frozen-truth battery: the 96 frozen-oracle cases replayed VERBATIM
+through the facade in BOTH core modes (RUNE_ME_CORE=python floor vs the
+Rust kernel), plus loader pin discipline, the fire→mark→quiet loop and
+input-copy semantics.
+"""
+
+from __future__ import annotations
+
+import copy
+import json
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "src"))
+sys.path.insert(0, str(ROOT / "src" / "rune"))
+
+import meaningful_events as me  # noqa: E402
+from rune import me_loader  # noqa: E402
+
+TRUTH = json.loads(
+    (ROOT / "src" / "rust" / "meaningful_events" / "frozen_truth.json")
+    .read_text())
+
+
+@pytest.fixture()
+def core_built():
+    if not me_loader.available():
+        pytest.skip("rust meaningful_events core not built")
+    return True
+
+
+def drive(mode, fn, *args):
+    os.environ["RUNE_ME_CORE"] = mode
+    try:
+        return fn(*copy.deepcopy(args))
+    finally:
+        os.environ["RUNE_ME_CORE"] = "auto"
+
+
+def test_loader_probe_and_pin(core_built, monkeypatch):
+    monkeypatch.delenv("RUNE_ME_CORE", raising=False)
+    assert me_loader.available()
+    assert me._kernel() is not None
+    monkeypatch.setenv("RUNE_ME_CORE", "python")
+    assert me._kernel() is None
+
+
+def test_frozen_should_fire_both_ways(core_built):
+    for c in TRUTH["should_fire_cases"]:
+        args = (c["args"]["trigger"], c["args"]["state"], c["args"]["now"])
+        py = drive("python", me.should_fire, *args)
+        rs = drive("auto", me.should_fire, *args)
+        assert py == rs == c["expect"], c["name"]
+
+
+def test_frozen_fire_due_both_ways(core_built):
+    for c in TRUTH["fire_due_cases"]:
+        args = (c["args"]["state"], c["args"]["now"])
+        py = drive("python", me.fire_due, *args)
+        rs = drive("auto", me.fire_due, *args)
+        assert py == rs == c["expect"], c["name"]
+
+
+def test_frozen_mark_fired_both_ways(core_built):
+    for c in TRUTH["mark_fired_cases"]:
+        args = (c["args"]["state"], c["args"]["trigger"], c["args"]["now"])
+        py = drive("python", me.mark_fired, *args)
+        rs = drive("auto", me.mark_fired, *args)
+        assert py == rs == c["expect"], c["name"]
+        assert c["input_unchanged"] is True, c["name"]
+
+
+def test_fire_mark_due_loop_both_ways(core_built):
+    for mode in ("python", "auto"):
+        state = {"task_blocked": True}
+        due = drive(mode, me.fire_due, state, 1000.0)
+        assert "task_blockers" in due
+        state = drive(mode, me.mark_fired, state, "task_blockers", 1000.0)
+        assert drive(mode, me.fire_due, state, 1000.0) == []
+
+
+def test_rate_limit_string_law(core_built):
+    # int() truncation surfaces IN the reason string — verbatim law
+    r = drive("auto", me.should_fire, "task_blockers",
+              {"task_blocked": True, "last_fired_task_blockers": 400.5},
+              1000.0)
+    assert r == {"fires": False, "reason": "rate-limited (0s left)"}