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

scripts/gen_testing_mode_truth.py

@@ -0,0 +1,727 @@
+#!/usr/bin/env python3
+"""gen_testing_mode_truth.py — freeze/replay parity truth for testing_mode
+(dichotomy wave, #2479 tranche).
+
+Freeze mode (default): exercises EVERY public function of the LIVE Python
+oracle (src/rune/testing_mode.py) across every branch and records
+(kernel_input, expected_kernel_output) pairs into
+src/rust/testing_mode/src/frozen_tm_truth.json.
+
+Replay mode (--replay): re-runs the SAME scenarios with the Rust core
+delegation active and asserts both kernel-level (loader direct) and
+facade-level (oracle functions delegate) parity against the frozen file.
+
+Poisoned-truth law: freeze pins RUNE_TM_CORE=python BEFORE any import (a
+no-op pre-port, load-bearing post-port — the facade must never delegate
+into the kernel while truth is being frozen).
+
+No-clock law: the wall clock is frozen by patching the `time` symbol in
+testing_mode, so every kernel payload carries its own `now`/`now_ts` and
+replay is deterministic regardless of run date.
+
+Boundary law: the session FILE IO (_load hit branch, save), the
+observations.jsonl append/rewrite, the forge adapter call, the ledger
+read (outstanding/record_real_world) and the registry IO all stay in the
+Python facade — the kernel is a pure function of their results. The
+`load`, `save` and `forge_dry_run` truth groups are facade-only (no
+kernel symbol) and are skipped by the kernel-level replay pass.
+
+Ledger/registry law: behaviours and registry are stubbed with
+deterministic fakes (the oracle treats them as duck-typed boundaries);
+their canned responses are deep-copied into the kernel inputs BEFORE the
+oracle runs (the oracle mutates the session state in place).
+"""
+import copy
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+TRUTH_PATH = os.path.join(REPO, "src", "rust", "testing_mode", "src", "frozen_tm_truth.json")
+REPLAY = "--replay" in sys.argv
+
+os.environ["RUNE_TM_CORE"] = "python"  # poisoned-truth law
+
+sys.path.insert(0, os.path.join(REPO, "src"))
+sys.path.insert(0, os.path.join(REPO, "src", "rune"))
+
+import testing_mode as oracle_mod  # noqa: E402
+
+if REPLAY:
+    from rune import tm_loader  # noqa: E402  (kernel-level replay side)
+    # replay law: the facade pass must exercise the LIVE Rust delegation
+    # path, so the pin is lifted AFTER import (the facade reads the env
+    # at call time). The kernel pass calls the loader directly.
+    del os.environ["RUNE_TM_CORE"]
+
+FROZEN_TS = 1791806400  # 2026-10-12T12:00:00Z
+FROZEN_NOW = "2026-10-12T12:00:00Z"
+
+
+class FakeTime:
+    @staticmethod
+    def strftime(fmt, t):
+        import time as _t
+        return _t.strftime(fmt, t)
+
+    @staticmethod
+    def gmtime():
+        import time as _t
+        return _t.gmtime(FROZEN_TS)
+
+    @staticmethod
+    def time():
+        return FROZEN_TS
+
+
+oracle_mod.time = FakeTime
+
+
+# ── deterministic boundary stubs ─────────────────────────────────────────────
+
+class FakeBehaviours:
+    """Duck-typed BehaviourLedger stub — canned outstanding/records."""
+
+    def __init__(self, outstanding=None, rec=None):
+        self.outstanding_value = outstanding or []
+        self.rec = rec or {"behaviour_id": "behaviour:T-1", "state": "USER_CONFIRMED"}
+        self.outstanding_kwargs = None
+
+    def outstanding(self, *, component=None, project=None):
+        self.outstanding_kwargs = {"component": component, "project": project}
+        return copy.deepcopy(self.outstanding_value)
+
+    def record_real_world(self, behaviour_id, *, passed, detail="",
+                          confirmed_by="user", generation=""):
+        rec = dict(self.rec)
+        rec["behaviour_id"] = rec.get("behaviour_id", f"behaviour:{behaviour_id}")
+        return rec
+
+
+class FakeRegistry:
+    def __init__(self, entries):
+        self._entries = entries
+        self.set_calls = []
+        self.saved = False
+
+    def all_projects(self):
+        return copy.deepcopy(self._entries)
+
+    def set_outstanding_validations(self, project_id, count):
+        self.set_calls.append((project_id, count))
+
+    def save(self):
+        self.saved = True
+
+
+class FakeLedger:
+    """Patched-in BehaviourLedger — summary canned per location."""
+
+    summaries = {}
+
+    def __init__(self, loc):
+        self.loc = loc
+
+    def summary(self):
+        return copy.deepcopy(FakeLedger.summaries.get(str(self.loc),
+                                                      {"outstanding": 0, "states": {}}))
+
+
+def fresh_session_dir():
+    return tempfile.mkdtemp(prefix="tm_truth_")
+
+
+def fresh_state():
+    s = oracle_mod.TestingSession(fresh_session_dir())
+    return copy.deepcopy(s.state)
+
+
+# ── case helpers ─────────────────────────────────────────────────────────────
+
+def rec_kernel(rec, name, inp, expected, oracle_ret=None):
+    """Record a kernel case; inputs deep-copied BEFORE oracle mutation."""
+    rec(name, copy.deepcopy(inp), oracle_ret, expected)
+
+
+def rec_facade(rec, name, inp, expected, oracle_ret=None):
+    rec(name, inp, oracle_ret, expected)
+
+
+# ── default_state (kernel) ───────────────────────────────────────────────────
+
+def _ds_case(rec, name, project, scope_component):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project=project, scope_component=scope_component)
+    rec_kernel(rec, name,
+               {"project": project, "scope_component": scope_component, "now": FROZEN_NOW},
+               {"state": s.state})
+
+
+def c_ds_default(rec):
+    _ds_case(rec, "default-no-scope", "p", None)
+
+
+def c_ds_scope(rec):
+    _ds_case(rec, "default-with-scope", "p", "component:shared")
+
+
+def c_ds_empty_project(rec):
+    _ds_case(rec, "default-empty-project", "", "component:x")
+
+
+# ── next_item (kernel) ───────────────────────────────────────────────────────
+
+def _next_case(rec, name, state_over, outstanding, expect_item):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project="p")
+    s.state.update(copy.deepcopy(state_over))
+    state_before = copy.deepcopy(s.state)
+    beh = FakeBehaviours(outstanding=outstanding)
+    item = s.next_item(beh)
+    assert beh.outstanding_kwargs == {"component": state_before.get("scope_component"),
+                                      "project": state_before.get("project") or None}
+    rec_kernel(rec, name,
+               {"state": state_before, "outstanding": outstanding},
+               {"state": s.state, "item": item},
+               oracle_ret=item)
+
+
+def c_next_paused(rec):
+    _next_case(rec, "next-paused-null", {"paused": True},
+               [{"behaviour_id": "behaviour:T-1"}], None)
+
+
+def c_next_empty(rec):
+    _next_case(rec, "next-empty-queue", {}, [], None)
+
+
+def c_next_first(rec):
+    _next_case(rec, "next-first-of-two", {},
+               [{"behaviour_id": "behaviour:T-1", "state": "UNVALIDATED"},
+                {"behaviour_id": "behaviour:T-2", "state": "RETEST_REQUIRED"}],
+               {"behaviour_id": "behaviour:T-1", "state": "UNVALIDATED"})
+
+
+def c_next_overwrites_current(rec):
+    _next_case(rec, "next-overwrites-current-item", {"current_item": "behaviour:OLD"},
+               [{"behaviour_id": "behaviour:T-2"}], {"behaviour_id": "behaviour:T-2"})
+
+
+def c_next_empty_clears_current(rec):
+    _next_case(rec, "next-empty-clears-current", {"current_item": "behaviour:T-1"},
+               [], None)
+
+
+# ── record (kernel) ──────────────────────────────────────────────────────────
+
+def _record_case(rec, name, completed_pre, behaviour_id, passed, rec_state):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project="p")
+    s.state["completed_this_session"] = copy.deepcopy(completed_pre)
+    state_before = copy.deepcopy(s.state)
+    beh = FakeBehaviours(rec={"behaviour_id": f"behaviour:{behaviour_id}",
+                              "state": rec_state})
+    ret = s.record(beh, behaviour_id, passed=passed, detail="checked")
+    rec_kernel(rec, name,
+               {"state": state_before,
+                "rec": ret, "passed": passed, "now": FROZEN_NOW},
+               {"state": s.state},
+               oracle_ret={"behaviour_id": ret["behaviour_id"], "state": ret["state"]})
+
+
+def c_record_pass(rec):
+    _record_case(rec, "record-pass-appends", [], "T-1", True, "USER_CONFIRMED")
+
+
+def c_record_fail(rec):
+    _record_case(rec, "record-fail-appends", [], "T-2", False, "TEST_FAILED")
+
+
+def c_record_appends_to_existing(rec):
+    pre = [{"behaviour_id": "behaviour:T-0", "passed": True, "at": "2026-10-12T11:00:00Z"}]
+    _record_case(rec, "record-appends-preserves-order", pre, "T-1", True, "USER_CONFIRMED")
+
+
+# ── pause / resume (kernel) ──────────────────────────────────────────────────
+
+def _pause_case(rec, name, paused_pre):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project="p")
+    s.state["paused"] = paused_pre
+    state_before = copy.deepcopy(s.state)
+    s.pause()
+    rec_kernel(rec, name,
+               {"state": state_before},
+               {"state": s.state})
+
+
+def _resume_case(rec, name, paused_pre):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project="p")
+    s.state["paused"] = paused_pre
+    state_before = copy.deepcopy(s.state)
+    s.resume()
+    rec_kernel(rec, name,
+               {"state": state_before},
+               {"state": s.state})
+
+
+def c_pause_from_active(rec):
+    _pause_case(rec, "pause-from-active", False)
+
+
+def c_pause_already_paused(rec):
+    _pause_case(rec, "pause-idempotent", True)
+
+
+def c_resume_from_paused(rec):
+    _resume_case(rec, "resume-from-paused", True)
+
+
+def c_resume_already_active(rec):
+    _resume_case(rec, "resume-idempotent", False)
+
+
+# ── note_blocker (kernel) ────────────────────────────────────────────────────
+
+def _blocker_case(rec, name, blockers_pre, note):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project="p")
+    s.state["blockers"] = copy.deepcopy(blockers_pre)
+    state_before = copy.deepcopy(s.state)
+    s.note_blocker("behaviour:T-9", note)
+    rec_kernel(rec, name,
+               {"state": state_before,
+                "behaviour_id": "behaviour:T-9", "note": note, "now": FROZEN_NOW},
+               {"state": s.state})
+
+
+def c_blocker_normal(rec):
+    _blocker_case(rec, "blocker-normal", [], "blocked on upstream")
+
+
+def c_blocker_truncates(rec):
+    _blocker_case(rec, "blocker-note-truncated-200", [], "x" * 260)
+
+
+def c_blocker_appends(rec):
+    pre = [{"behaviour_id": "behaviour:T-1", "note": "old", "at": "2026-10-12T11:00:00Z"}]
+    _blocker_case(rec, "blocker-appends-preserves-order", pre, "new")
+
+
+# ── record_observation (kernel) ──────────────────────────────────────────────
+
+def _obs_case(rec, name, **kw):
+    d = fresh_session_dir()
+    obs = oracle_mod.record_observation(d, **kw)
+    inp = {"now": FROZEN_NOW, "now_ts": FROZEN_TS,
+           "observed": kw.get("observed", ""),
+           "expected": kw.get("expected", ""),
+           "behaviour_id": kw.get("behaviour_id", ""),
+           "environment": kw.get("environment", ""),
+           "reproduction": kw.get("reproduction", ""),
+           "candidate_areas": list(kw.get("candidate_areas") or []),
+           "confidence": kw.get("confidence", "low")}
+    rec_kernel(rec, name, inp, {"observation": obs}, oracle_ret=obs)
+
+
+def c_obs_defaults(rec):
+    _obs_case(rec, "obs-defaults", observed="button dead", expected="button saves")
+
+
+def c_obs_full(rec):
+    _obs_case(rec, "obs-full-fields", observed="500 on save", expected="201 created",
+              behaviour_id="behaviour:T-1", environment="production",
+              reproduction="POST /save with empty body",
+              candidate_areas=["component:api"], confidence="high")
+
+
+def c_obs_candidate_areas_none(rec):
+    _obs_case(rec, "obs-candidate-areas-none-coerced", observed="x", expected="y",
+              candidate_areas=None)
+
+
+def c_obs_candidate_areas_copy(rec):
+    src = ["component:a", "component:b"]
+    _obs_case(rec, "obs-candidate-areas-hypotheses", observed="x", expected="y",
+              candidate_areas=src)
+
+
+# ── close_observation (kernel) ───────────────────────────────────────────────
+
+def _obs_lines(d, n=1):
+    lines = []
+    for i in range(n):
+        obs = oracle_mod.record_observation(
+            d, observed=f"obs-{i}", expected=f"exp-{i}",
+            behaviour_id=f"behaviour:T-{i}")
+        lines.append(json.loads(json.dumps(obs, sort_keys=True)))
+    return lines
+
+
+def _close_case(rec, name, records, observation_id, retest, confirmed,
+                contract_ref="", expect_raise=False):
+    d = fresh_session_dir()
+    with (Path(d) / "observations.jsonl").open("w") as f:
+        for r in records:
+            f.write(json.dumps(r, sort_keys=True) + "\n")
+    try:
+        result = oracle_mod.close_observation(
+            d, observation_id, behaviour_retest_passed=retest,
+            user_confirmed=confirmed, contract_ref=contract_ref)
+        ret, expected = result, {"records": _redump(records, observation_id, retest,
+                                                   confirmed, contract_ref, FROZEN_NOW),
+                                 "result": result}
+    except ValueError as e:
+        ret = f"ValueError: {e}"
+        expected = {"records": [json.loads(json.dumps(r, sort_keys=True)) for r in records],
+                    "result": None}
+    rec_kernel(rec, name,
+               {"records": copy.deepcopy(records), "observation_id": observation_id,
+                "behaviour_retest_passed": retest, "user_confirmed": confirmed,
+                "contract_ref": contract_ref, "now": FROZEN_NOW},
+               expected, oracle_ret=ret)
+
+
+def _redump(records, observation_id, retest, confirmed, contract_ref, now):
+    out = []
+    for r in records:
+        r = copy.deepcopy(r)
+        if r.get("observation_id") == observation_id:
+            r["closure"] = {"behaviour_retest_passed": retest, "user_confirmed": confirmed,
+                            "contract_ref": contract_ref,
+                            "closed": retest and confirmed, "at": now}
+        out.append(r)
+    return out
+
+
+def c_close_full(rec):
+    d = fresh_session_dir()
+    records = _obs_lines(d, 1)
+    _close_case(rec, "close-retest-and-confirmed", records,
+                records[0]["observation_id"], True, True)
+
+
+def c_close_no_user_confirm(rec):
+    d = fresh_session_dir()
+    records = _obs_lines(d, 1)
+    _close_case(rec, "close-green-without-user-not-closed", records,
+                records[0]["observation_id"], True, False)
+
+
+def c_close_no_retest(rec):
+    d = fresh_session_dir()
+    records = _obs_lines(d, 1)
+    _close_case(rec, "close-confirm-without-retest-not-closed", records,
+                records[0]["observation_id"], False, True)
+
+
+def c_close_with_contract_ref(rec):
+    d = fresh_session_dir()
+    records = _obs_lines(d, 1)
+    _close_case(rec, "close-with-contract-ref", records,
+                records[0]["observation_id"], True, True,
+                contract_ref="BUILDER-2026-10-12-factory-wave")
+
+
+def c_close_unknown(rec):
+    d = fresh_session_dir()
+    records = _obs_lines(d, 1)
+    _close_case(rec, "close-unknown-id-raises", records, "obs-404", True, True,
+                expect_raise=True)
+
+
+def c_close_preserves_others(rec):
+    d = fresh_session_dir()
+    records = _obs_lines(d, 3)
+    _close_case(rec, "close-preserves-sibling-records", records,
+                records[1]["observation_id"], True, True)
+
+
+# ── project_failure_to_forge (kernel: title/body; dry_run facade-only) ──────
+
+def _project_case(rec, name, behaviour_id, observation):
+    obs = oracle_mod.project_failure_to_forge(
+        None, "conductor/rune", behaviour_id=behaviour_id,
+        observation=observation, dry_run=True)
+    from rune.testing_mode import _py_project
+    title, body = _py_project(behaviour_id, observation)
+    assert title == obs["title"]
+    rec_kernel(rec, name,
+               {"behaviour_id": behaviour_id, "observation": copy.deepcopy(observation)},
+               {"title": obs["title"], "body": body}, oracle_ret=obs)
+
+
+def c_project_normal(rec):
+    _project_case(rec, "project-normal", "T-1",
+                  {"observation_id": "obs-1", "observed": "500", "expected": "201",
+                   "reproduction": "POST /x"})
+
+
+def c_project_truncates_expected(rec):
+    _project_case(rec, "project-expected-truncated-60", "T-2",
+                  {"observation_id": "obs-2", "observed": "o", "expected": "e" * 90,
+                   "reproduction": "r"})
+
+
+def c_project_missing_fields(rec):
+    _project_case(rec, "project-missing-fields-empty-defaults", "T-3", {})
+
+
+def c_project_observation_id_none(rec):
+    # missing observation_id renders Python str(None) → "None"
+    _project_case(rec, "project-observation-id-none-string", "T-4",
+                  {"observed": "x", "expected": "y", "reproduction": "z"})
+
+
+def c_project_dry_run(rec):
+    obs = {"observation_id": "obs-9", "observed": "x", "expected": "y",
+           "reproduction": "z"}
+    ret = oracle_mod.project_failure_to_forge(
+        None, "conductor/rune", behaviour_id="T-9", observation=obs, dry_run=True)
+    rec_facade(rec, "forge-dry-run-no-adapter-call",
+               {"behaviour_id": "T-9", "observation": obs, "dry_run": True},
+               {"projected": ret["projected"], "dry_run": ret["dry_run"],
+                "title": ret["title"]}, oracle_ret=ret)
+
+
+# ── aggregate (kernel shaping over facade-fetched summaries) ────────────────
+
+def _agg_case(rec, name, entries, ledger_flags, summaries):
+    d = fresh_session_dir()
+    locs = []
+    for i, present in enumerate(ledger_flags):
+        loc = Path(d) / f"p{i + 1}"
+        loc.mkdir(exist_ok=True)
+        if present:
+            (loc / "behaviour_ledger.json").write_text("{}")
+        locs.append(str(loc))
+    reg_entries = [{"project_id": e["project_id"], "scaffold_location": loc}
+                   for e, loc in zip(entries, locs)]
+    FakeLedger.summaries = {loc: s for loc, s in zip(locs, summaries) if s is not None}
+    oracle_mod.BehaviourLedger = FakeLedger
+    try:
+        rows = oracle_mod.aggregate(FakeRegistry(reg_entries), Path(d))
+    finally:
+        from rune.behaviour_ledger import BehaviourLedger as RealLedger
+        oracle_mod.BehaviourLedger = RealLedger
+    kernel_entries = [{"project_id": e["project_id"],
+                       "summary": (copy.deepcopy(s) if present else None)}
+                      for e, present, s in zip(entries, ledger_flags, summaries)]
+    rec_kernel(rec, name, {"entries": kernel_entries}, {"rows": rows}, oracle_ret=rows)
+
+
+def c_agg_no_ledger(rec):
+    _agg_case(rec, "agg-no-ledger-note", [{"project_id": "p1"}], [False], [None])
+
+
+def c_agg_with_summary(rec):
+    _agg_case(rec, "agg-with-summary", [{"project_id": "p1"}], [True],
+              [{"outstanding": 2, "states": {"UNVALIDATED": 2}}])
+
+
+def c_agg_mixed(rec):
+    _agg_case(rec, "agg-mixed-projects", [{"project_id": "p1"}, {"project_id": "p2"},
+                                          {"project_id": "p3"}],
+              [True, False, True],
+              [{"outstanding": 1, "states": {"RETEST_REQUIRED": 1}}, None,
+               {"outstanding": 0, "states": {}}])
+
+
+def c_agg_empty(rec):
+    _agg_case(rec, "agg-empty-registry", [], [], [])
+
+
+# ── load / save (facade-only: FILE IO stays in Python) ──────────────────────
+
+def c_load_missing(rec):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project="p", scope_component="component:c")
+    rec_facade(rec, "load-missing-file-default-state", {"path": "missing"},
+               {"state": s.state})
+
+
+def c_load_valid(rec):
+    d = fresh_session_dir()
+    pre = dict(fresh_state(), current_item="behaviour:T-1", paused=True)
+    Path(d, "testing_session.json").write_text(json.dumps(pre, indent=1, sort_keys=True))
+    s = oracle_mod.TestingSession(d)
+    rec_facade(rec, "load-valid-file-roundtrip", {"path": "valid"},
+               {"state": s.state})
+
+
+def c_save(rec):
+    d = fresh_session_dir()
+    s = oracle_mod.TestingSession(d, project="p")
+    s.state["current_item"] = "behaviour:T-1"
+    s.state["blockers"].append({"behaviour_id": "behaviour:T-2", "note": "n", "at": FROZEN_NOW})
+    s.pause()
+    s.save()
+    text = Path(d, "testing_session.json").read_text()
+    rec_facade(rec, "save-stamps-updated-at-indent1-sortkeys",
+               {"state": copy.deepcopy(s.state)}, {"file_text": text},
+               oracle_ret=None)
+
+
+# ── driver ───────────────────────────────────────────────────────────────────
+
+DEFAULT_STATE = [c_ds_default, c_ds_scope, c_ds_empty_project]
+NEXT = [c_next_paused, c_next_empty, c_next_first, c_next_overwrites_current,
+        c_next_empty_clears_current]
+RECORD = [c_record_pass, c_record_fail, c_record_appends_to_existing]
+PAUSE = [c_pause_from_active, c_pause_already_paused]
+RESUME = [c_resume_from_paused, c_resume_already_active]
+BLOCKER = [c_blocker_normal, c_blocker_truncates, c_blocker_appends]
+OBS = [c_obs_defaults, c_obs_full, c_obs_candidate_areas_none, c_obs_candidate_areas_copy]
+CLOSE = [c_close_full, c_close_no_user_confirm, c_close_no_retest,
+         c_close_with_contract_ref, c_close_unknown, c_close_preserves_others]
+PROJECT = [c_project_normal, c_project_truncates_expected, c_project_missing_fields,
+           c_project_observation_id_none]
+FORGE = [c_project_dry_run]
+AGG = [c_agg_no_ledger, c_agg_with_summary, c_agg_mixed, c_agg_empty]
+LOAD = [c_load_missing, c_load_valid]
+SAVE = [c_save]
+
+GROUPS = [("default_state", DEFAULT_STATE), ("next_item", NEXT), ("record", RECORD),
+          ("pause", PAUSE), ("resume", RESUME), ("note_blocker", BLOCKER),
+          ("record_observation", OBS), ("close_observation", CLOSE),
+          ("project", PROJECT), ("forge_dry_run", FORGE), ("aggregate", AGG),
+          ("load", LOAD), ("save", SAVE)]
+
+# kernel symbol per group (None → facade-only group, skipped kernel-side)
+KERNEL = {
+    "default_state": "tm_default_state",
+    "next_item": "tm_next_item",
+    "record": "tm_record",
+    "pause": "tm_pause",
+    "resume": "tm_resume",
+    "note_blocker": "tm_note_blocker",
+    "record_observation": "tm_record_observation",
+    "close_observation": "tm_close_observation",
+    "project": "tm_project",
+    "forge_dry_run": None,
+    "aggregate": "tm_aggregate",
+    "load": None,
+    "save": None,
+}
+
+# loader wrapper + slice of `expected` the unwrapped FFI value maps to
+CALLS = {
+    "default_state": ("default_state", lambda w: w["state"],
+                      lambda i, ld: ld.default_state(project=i["project"],
+                                                     scope_component=i["scope_component"],
+                                                     now=i["now"])),
+    "next_item": ("next_item", lambda w: w,
+                  lambda i, ld: ld.next_item(state=i["state"], outstanding=i["outstanding"])),
+    "record": ("record", lambda w: w["state"],
+               lambda i, ld: ld.record(state=i["state"], rec=i["rec"],
+                                       passed=i["passed"], now=i["now"])),
+    "pause": ("pause", lambda w: w["state"], lambda i, ld: ld.pause(state=i["state"])),
+    "resume": ("resume", lambda w: w["state"], lambda i, ld: ld.resume(state=i["state"])),
+    "note_blocker": ("note_blocker", lambda w: w["state"],
+                     lambda i, ld: ld.note_blocker(state=i["state"],
+                                                   behaviour_id=i["behaviour_id"],
+                                                   note=i["note"], now=i["now"])),
+    "record_observation": ("record_observation", lambda w: w["observation"],
+                           lambda i, ld: ld.record_observation(
+                               now=i["now"], now_ts=i["now_ts"], observed=i["observed"],
+                               expected=i["expected"], behaviour_id=i["behaviour_id"],
+                               environment=i["environment"], reproduction=i["reproduction"],
+                               candidate_areas=i["candidate_areas"],
+                               confidence=i["confidence"])),
+    "close_observation": ("close_observation", lambda w: w,
+                          lambda i, ld: ld.close_observation(
+                              records=i["records"], observation_id=i["observation_id"],
+                              behaviour_retest_passed=i["behaviour_retest_passed"],
+                              user_confirmed=i["user_confirmed"],
+                              contract_ref=i["contract_ref"], now=i["now"])),
+    "project": ("project", lambda w: w,
+                lambda i, ld: ld.project(behaviour_id=i["behaviour_id"],
+                                         observation=i["observation"])),
+    "aggregate": ("aggregate", lambda w: w["rows"],
+                  lambda i, ld: ld.aggregate(entries=i["entries"])),
+}
+
+
+def canon(x) -> str:
+    """Content canonicalization: a sorted dump makes the comparison
+    key-order independent (int keys never occur here)."""
+    return json.dumps(json.loads(json.dumps(x)), sort_keys=True)
+
+
+def main():
+    frozen = None
+    if REPLAY:
+        with open(TRUTH_PATH) as f:
+            frozen = json.load(f)["cases"]
+
+    groups = {}
+    for group, cases in GROUPS:
+        recs = []
+
+        def rec(name, inp, oracle_ret, expected, _r=recs):
+            entry = {"name": name, "input": inp, "return": oracle_ret,
+                     "expected": expected}
+            if REPLAY:
+                old = next(c for c in frozen[group] if c["name"] == name)
+                for key in ("input", "return", "expected"):
+                    got, want = entry[key], old[key]
+                    if canon(got) != canon(want):
+                        print(f"FACADE DIVERGE [{group}] {name} {key}")
+                        print("  got: ", json.dumps(got, sort_keys=True, default=str)[:600])
+                        print("  want:", json.dumps(want, sort_keys=True, default=str)[:600])
+                        recs.append({"name": name, "diverged": key})
+                        return
+            _r.append(entry)
+
+        for c in cases:
+            c(rec)
+        groups[group] = recs
+
+    total = sum(len(v) for v in groups.values())
+    if REPLAY:
+        facade_failures = sum(1 for v in groups.values() for c in v if "diverged" in c)
+
+        failures = 0
+        checked = 0
+        for group, cases in frozen.items():
+            spec = CALLS.get(group)
+            if spec is None:
+                continue
+            _, unwrap, call = spec
+            for case in cases:
+                checked += 1
+                got = call(copy.deepcopy(case["input"]), tm_loader)
+                want = unwrap(case["expected"])
+                if canon(got) != canon(want):
+                    failures += 1
+                    print(f"KERNEL DIVERGE [{group}] {case['name']}")
+                    print("  got: ", json.dumps(got, sort_keys=True)[:600])
+                    print("  want:", json.dumps(want, sort_keys=True, default=str)[:600])
+        print(f"replay: facade {total - facade_failures}/{total}, "
+              f"kernel {checked - failures}/{checked}")
+        return 1 if (failures or facade_failures) else 0
+
+    doc = {
+        "frozen": "2026-10-12T12:00:00Z",
+        "oracle": "src/rune/testing_mode.py",
+        "pin": "RUNE_TM_CORE=python",
+        "note": "Frozen via scripts/gen_testing_mode_truth.py against the live "
+                "Python oracle; wall clock frozen at " + FROZEN_NOW +
+                " (FakeTime patched into testing_mode.time; epoch " + str(FROZEN_TS) +
+                "). Battery embeds this VERBATIM. Groups `forge_dry_run`, `load` "
+                "and `save` are facade-only (FILE IO + adapter call stay in Python).",
+        "cases": groups,
+    }
+    os.makedirs(os.path.dirname(TRUTH_PATH), exist_ok=True)
+    with open(TRUTH_PATH, "w") as f:
+        json.dump(doc, f, indent=1)
+        f.write("\n")
+    print(f"frozen {total} cases → {os.path.relpath(TRUTH_PATH, REPO)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/rune/testing_mode.py

@@ -10,9 +10,19 @@ hardened behaviours stay trusted.
 Forge projection (19P): failures in active iteration create/update a
 human-visible issue through the T18 forge boundary. Forge failure never
 rewrites canonical truth.
+
+Dichotomy wave #2479: the session state machine (next-item selection,
+record bookkeeping, pause/resume, blockers), the observation
+record/close shaping, the forge projection title/body and the aggregate
+row shaping live in the Rust kernel (src/rust/testing_mode) behind
+doc-envelope FFI; the session/observations FILE IO, the ledger read,
+the registry IO and the forge adapter call stay here — the kernel is a
+pure function of their results. RUNE_TM_CORE=python pins the Python
+fallback (parity harnesses).
 """
 from __future__ import annotations
 
+import os
 import time
 from pathlib import Path
 
@@ -27,6 +37,129 @@ def _now() -> str:
     return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
 
 
+def _tm_core():
+    """Dichotomy wave #2479: Rust core first (RUNE_TM_CORE=python pins the
+    Python fallback — parity harnesses). The kernel owns the session
+    state machine + observation/forge/aggregate shaping; the FILE IO,
+    the ledger read and the registry IO stay in this facade."""
+    if os.environ.get("RUNE_TM_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import tm_loader
+
+        if tm_loader.available():
+            return tm_loader
+    except ImportError:
+        pass
+    return None
+
+
+# ── pure shaping helpers (the Python fallback + parity oracle) ───────────
+
+def _py_default_state(project, scope_component, now) -> dict:
+    return {"schema_version": SESSION_SCHEMA_VERSION,
+            "project": project, "scope_component": scope_component,
+            "current_item": None, "completed_this_session": [],
+            "newly_invalidated": [], "blockers": [],
+            "paused": False, "updated_at": now}
+
+
+def _py_next_item(state: dict, outstanding: list) -> tuple[dict, dict | None]:
+    """19L next eligible item — paused sessions dequeue nothing."""
+    if state["paused"]:
+        return state, None
+    item = outstanding[0] if outstanding else None
+    state["current_item"] = (item["behaviour_id"] if item else None)
+    return state, item
+
+
+def _py_record(state: dict, rec: dict, passed: bool, now: str) -> dict:
+    state["completed_this_session"].append(
+        {"behaviour_id": rec["behaviour_id"], "passed": passed,
+         "at": now})
+    if passed and rec["state"] == USER_CONFIRMED:
+        pass  # hardened; queue re-derives
+    return state
+
+
+def _py_note_blocker(state: dict, behaviour_id: str, note: str, now: str) -> dict:
+    state["blockers"].append({"behaviour_id": behaviour_id,
+                              "note": note[:200], "at": now})
+    return state
+
+
+def _py_record_observation(*, now: str, now_ts: int, observed: str,
+                           expected: str, behaviour_id: str = "",
+                           environment: str = "", reproduction: str = "",
+                           candidate_areas: list[str] | None = None,
+                           confidence: str = "low") -> dict:
+    """Durable real-world failure report. Candidate areas are
+    HYPOTHESES, not declared root cause (rule 19O)."""
+    obs = {
+        "observation_id": f"obs-{now_ts}",
+        "observed": observed, "expected": expected,
+        "behaviour_id": behaviour_id,
+        "environment": environment, "reproduction": reproduction,
+        "candidate_areas": list(candidate_areas or []),  # hypotheses
+        "confidence": confidence,
+        "created_at": now,
+        "closure": None,  # set by close_observation
+    }
+    return obs
+
+
+def _py_close_observation(records: list, observation_id: str, *,
+                          behaviour_retest_passed: bool,
+                          user_confirmed: bool,
+                          contract_ref: str = "",
+                          now: str = "") -> tuple[list, dict | None]:
+    """Closure returns to the ORIGINAL observed scenario (rule 19O):
+    code changed ≠ closure; automated green ≠ closure. The observation
+    closes only when the behaviour was retested AND user-confirmed."""
+    out = []
+    result = None
+    for rec in records:
+        if rec.get("observation_id") == observation_id:
+            rec["closure"] = {
+                "behaviour_retest_passed": behaviour_retest_passed,
+                "user_confirmed": user_confirmed,
+                "contract_ref": contract_ref,
+                "closed": behaviour_retest_passed and user_confirmed,
+                "at": now}
+            result = rec
+        out.append(rec)
+    return out, result
+
+
+def _py_project(behaviour_id: str, observation: dict) -> tuple[str, str]:
+    """19P forge projection title/body (the projection itself is a
+    PROJECTION — canonical truth stays in the Behaviour Ledger)."""
+    title = f"[testing] {behaviour_id}: {observation.get('expected', '')[:60]}"
+    body = (f"Observed: {observation.get('observed', '')}\n"
+            f"Expected: {observation.get('expected', '')}\n"
+            f"Behaviour: {behaviour_id}\n"
+            f"Observation: {observation.get('observation_id')}\n"
+            f"Reproduction: {observation.get('reproduction', '')}")
+    return title, body
+
+
+def _py_aggregate_rows(entries: list) -> list[dict]:
+    """19L aggregate row shaping over facade-fetched per-project
+    summaries (summary None → no behaviour ledger yet)."""
+    out = []
+    for entry in entries:
+        if entry["summary"] is None:
+            out.append({"project": entry["project_id"],
+                        "outstanding": None,
+                        "note": "no behaviour ledger yet"})
+            continue
+        s = entry["summary"]
+        out.append({"project": entry["project_id"],
+                    "outstanding": s["outstanding"],
+                    "states": s["states"]})
+    return out
+
+
 # ── 19N: session = coordination only ─────────────────────────────────────
 
 class TestingSession:
@@ -40,11 +173,12 @@ class TestingSession:
         import json
         if self.path.is_file():
             return json.loads(self.path.read_text())
-        return {"schema_version": SESSION_SCHEMA_VERSION,
-                "project": project, "scope_component": scope_component,
-                "current_item": None, "completed_this_session": [],
-                "newly_invalidated": [], "blockers": [],
-                "paused": False, "updated_at": _now()}
+        core = _tm_core()
+        if core is not None:
+            return core.default_state(project=project,
+                                      scope_component=scope_component,
+                                      now=_now())
+        return _py_default_state(project, scope_component, _now())
 
     def save(self) -> None:
         import json
@@ -55,14 +189,17 @@ class TestingSession:
     # ── 19L: next eligible item ──────────────────────────────────────
 
     def next_item(self, behaviours: BehaviourLedger) -> dict | None:
-        if self.state["paused"]:
-            return None
-        outstanding = behaviours.outstanding(
+        core = _tm_core()
+        if core is not None:
+            outstanding = behaviours.outstanding(
+                component=self.state.get("scope_component"),
+                project=self.state.get("project") or None)
+            out = core.next_item(state=self.state, outstanding=outstanding)
+            self.state = out["state"]
+            return out["item"]
+        _, item = _py_next_item(self.state, behaviours.outstanding(
             component=self.state.get("scope_component"),
-            project=self.state.get("project") or None)
-        item = outstanding[0] if outstanding else None
-        self.state["current_item"] = (item["behaviour_id"]
-                                      if item else None)
+            project=self.state.get("project") or None))
         return item
 
     # ── 19M: record result; dynamic queue re-derives from states ─────
@@ -73,25 +210,40 @@ class TestingSession:
         rec = behaviours.record_real_world(
             behaviour_id, passed=passed, detail=detail, confirmed_by=by,
             generation=generation)
-        self.state["completed_this_session"].append(
-            {"behaviour_id": rec["behaviour_id"], "passed": passed,
-             "at": _now()})
-        if passed and rec["state"] == USER_CONFIRMED:
-            pass  # hardened; queue re-derives
+        core = _tm_core()
+        if core is not None:
+            self.state = core.record(state=self.state, rec=rec,
+                                     passed=passed, now=_now())
+        else:
+            self.state = _py_record(self.state, rec, passed, _now())
         self.save()
         return rec
 
     def pause(self) -> None:
-        self.state["paused"] = True
+        core = _tm_core()
+        if core is not None:
+            self.state = core.pause(state=self.state)
+        else:
+            self.state["paused"] = True
         self.save()
 
     def resume(self) -> None:
-        self.state["paused"] = False
+        core = _tm_core()
+        if core is not None:
+            self.state = core.resume(state=self.state)
+        else:
+            self.state["paused"] = False
         self.save()
 
     def note_blocker(self, behaviour_id: str, note: str) -> None:
-        self.state["blockers"].append({"behaviour_id": behaviour_id,
-                                       "note": note[:200], "at": _now()})
+        core = _tm_core()
+        if core is not None:
+            self.state = core.note_blocker(state=self.state,
+                                           behaviour_id=behaviour_id,
+                                           note=note, now=_now())
+        else:
+            self.state = _py_note_blocker(self.state, behaviour_id, note,
+                                          _now())
         self.save()
 
 
@@ -105,16 +257,23 @@ def record_observation(store_dir: Path | str, *, observed: str,
     """Durable real-world failure report. Candidate areas are
     HYPOTHESES, not declared root cause (rule 19O)."""
     import json
-    obs = {
-        "observation_id": f"obs-{int(time.time())}",
-        "observed": observed, "expected": expected,
-        "behaviour_id": behaviour_id,
-        "environment": environment, "reproduction": reproduction,
-        "candidate_areas": list(candidate_areas or []),  # hypotheses
-        "confidence": confidence,
-        "created_at": _now(),
-        "closure": None,  # set by close_observation
-    }
+    core = _tm_core()
+    if core is not None:
+        obs = core.record_observation(now=_now(), now_ts=int(time.time()),
+                                      observed=observed, expected=expected,
+                                      behaviour_id=behaviour_id,
+                                      environment=environment,
+                                      reproduction=reproduction,
+                                      candidate_areas=list(candidate_areas or []),
+                                      confidence=confidence)
+    else:
+        obs = _py_record_observation(now=_now(), now_ts=int(time.time()),
+                                     observed=observed, expected=expected,
+                                     behaviour_id=behaviour_id,
+                                     environment=environment,
+                                     reproduction=reproduction,
+                                     candidate_areas=candidate_areas,
+                                     confidence=confidence)
     p = Path(store_dir) / "observations.jsonl"
     with p.open("a") as f:
         f.write(json.dumps(obs, sort_keys=True) + "\n")
@@ -131,20 +290,24 @@ def close_observation(store_dir: Path | str, observation_id: str, *,
     import json
     p = Path(store_dir) / "observations.jsonl"
     lines = p.read_text().strip().splitlines()
-    out = []
-    result = None
-    for line in lines:
-        rec = json.loads(line)
-        if rec.get("observation_id") == observation_id:
-            rec["closure"] = {
-                "behaviour_retest_passed": behaviour_retest_passed,
-                "user_confirmed": user_confirmed,
-                "contract_ref": contract_ref,
-                "closed": behaviour_retest_passed and user_confirmed,
-                "at": _now()}
-            result = rec
-        out.append(json.dumps(rec, sort_keys=True))
-    p.write_text("\n".join(out) + "\n")
+    records = [json.loads(line) for line in lines]
+    core = _tm_core()
+    if core is not None:
+        out = core.close_observation(records=records,
+                                     observation_id=observation_id,
+                                     behaviour_retest_passed=behaviour_retest_passed,
+                                     user_confirmed=user_confirmed,
+                                     contract_ref=contract_ref,
+                                     now=_now())
+        out_records, result = out["records"], out["result"]
+    else:
+        out_records, result = _py_close_observation(
+            records, observation_id,
+            behaviour_retest_passed=behaviour_retest_passed,
+            user_confirmed=user_confirmed, contract_ref=contract_ref,
+            now=_now())
+    p.write_text("\n".join(json.dumps(rec, sort_keys=True)
+                           for rec in out_records) + "\n")
     if result is None:
         raise ValueError(f"unknown observation {observation_id!r}")
     return result
@@ -160,12 +323,13 @@ def project_failure_to_forge(forge_adapter, repo: str, *,
     in the Behaviour Ledger. Forge failure never rewrites truth
     (rule 19P/19S-14): exceptions propagate to the caller as
     ForgeUnavailable; the ledger state is untouched."""
-    title = f"[testing] {behaviour_id}: {observation.get('expected', '')[:60]}"
-    body = (f"Observed: {observation.get('observed', '')}\n"
-            f"Expected: {observation.get('expected', '')}\n"
-            f"Behaviour: {behaviour_id}\n"
-            f"Observation: {observation.get('observation_id')}\n"
-            f"Reproduction: {observation.get('reproduction', '')}")
+    core = _tm_core()
+    if core is not None:
+        proj = core.project(behaviour_id=behaviour_id,
+                            observation=observation)
+        title, body = proj["title"], proj["body"]
+    else:
+        title, body = _py_project(behaviour_id, observation)
     if dry_run:
         return {"projected": False, "dry_run": True, "title": title}
     return forge_adapter.create_or_patch_release(
@@ -177,20 +341,22 @@ def project_failure_to_forge(forge_adapter, repo: str, *,
 def aggregate(registry: ProjectRegistry, store_root: Path) -> list[dict]:
     """Outstanding validation across ALL registered scaffolded projects
     (the /testing aggregate)."""
-    out = []
+    core = _tm_core()
+    entries = []
     for entry in registry.all_projects():
         loc = Path(entry["scaffold_location"])
         if not (loc / "behaviour_ledger.json").is_file():
-            out.append({"project": entry["project_id"],
-                        "outstanding": None,
-                        "note": "no behaviour ledger yet"})
+            entries.append({"project_id": entry["project_id"],
+                            "summary": None})
             continue
         beh = BehaviourLedger(loc)
-        s = beh.summary()
-        registry.set_outstanding_validations(entry["project_id"],
-                                             s["outstanding"])
-        out.append({"project": entry["project_id"],
-                    "outstanding": s["outstanding"],
-                    "states": s["states"]})
+        entries.append({"project_id": entry["project_id"],
+                        "summary": beh.summary()})
+    rows = (core.aggregate(entries=entries) if core is not None
+            else _py_aggregate_rows(entries))
+    for row in rows:
+        if row["outstanding"] is not None:
+            registry.set_outstanding_validations(row["project"],
+                                                 row["outstanding"])
     registry.save()
-    return out
+    return rows

src/rune/tm_loader.py

@@ -0,0 +1,178 @@
+#!/usr/bin/env python3
+"""ctypes loader for the Rust testing_mode core (dichotomy wave, #2479
+tranche). Mirrors broker_loader.py's contract:
+
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * tm_default_state({project, scope_component, now}) -> {state}
+  * tm_next_item({state, outstanding}) -> {state, item}
+  * tm_record({state, rec, passed, now}) -> {state}
+  * tm_pause({state}) -> {state}
+  * tm_resume({state}) -> {state}
+  * tm_note_blocker({state, behaviour_id, note, now}) -> {state}
+  * tm_record_observation({now, now_ts, observed, expected, behaviour_id,
+      environment, reproduction, candidate_areas, confidence})
+    -> {observation}
+  * tm_close_observation({records, observation_id,
+      behaviour_retest_passed, user_confirmed, contract_ref, now})
+    -> {records, result}
+  * tm_project({behaviour_id, observation}) -> {title, body}
+  * tm_aggregate({entries}) -> {rows}
+
+No-clock law: `now` (UTC ISO-8601) and `now_ts` (epoch seconds, for the
+oracle's obs-{int(time.time())} id) are always arguments to the stamping
+calls — the kernel never reads the wall clock.
+
+Boundary: the session FILE IO (TestingSession._load hit branch / save),
+the observations.jsonl append/rewrite, the BehaviourLedger read
+(outstanding / record_real_world), the ProjectRegistry IO and the forge
+adapter call stay in the Python facade (testing_mode.py); the kernel is
+a pure function of their results.
+
+FFI segfault law: every symbol gets explicit argtypes + restype (c_void_p
+for returned char* — the default c_int truncates 64-bit pointers).
+Strings are freed via capability_scoring's cs_free_string — the kernel
+exports NO free symbol of its own (duplicate exports break the cdylib
+link when sibling crates share one process).
+
+Override the .so with TM_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" / "testing_mode"
+_LIB_PATH = Path(
+    os.environ.get(
+        "TM_RUST_LIB", _CRATE_DIR / "target" / "release" / "libtesting_mode.so"
+    )
+)
+
+_SYMBOLS = (
+    "tm_default_state",
+    "tm_next_item",
+    "tm_record",
+    "tm_pause",
+    "tm_resume",
+    "tm_note_blocker",
+    "tm_record_observation",
+    "tm_close_observation",
+    "tm_project",
+    "tm_aggregate",
+)
+
+_lib = None
+
+
+class TestingModeCoreUnavailable(FileNotFoundError):
+    """The Rust testing_mode 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 TestingModeCoreUnavailable(
+            f"Rust testing_mode core not built: {_LIB_PATH} missing — run "
+            f"`cargo build --release` in {_CRATE_DIR}"
+        )
+    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
+    _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 via any sibling free symbol already in-process (capability_scoring)
+    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
+    doc = json.loads(out)
+    if not doc.get("ok"):
+        raise RuntimeError(doc.get("err", f"{sym} failed"))
+    return doc["value"]
+
+
+def default_state(project: str, scope_component: str | None, now: str) -> dict:
+    return _call("tm_default_state", {"project": project,
+                                      "scope_component": scope_component,
+                                      "now": now})["state"]
+
+
+def next_item(state: dict, outstanding: list) -> dict:
+    return _call("tm_next_item", {"state": state, "outstanding": outstanding})
+
+
+def record(state: dict, rec: dict, passed: bool, now: str) -> dict:
+    return _call("tm_record", {"state": state, "rec": rec, "passed": passed,
+                               "now": now})["state"]
+
+
+def pause(state: dict) -> dict:
+    return _call("tm_pause", {"state": state})["state"]
+
+
+def resume(state: dict) -> dict:
+    return _call("tm_resume", {"state": state})["state"]
+
+
+def note_blocker(state: dict, behaviour_id: str, note: str, now: str) -> dict:
+    return _call("tm_note_blocker", {"state": state,
+                                     "behaviour_id": behaviour_id,
+                                     "note": note, "now": now})["state"]
+
+
+def record_observation(now: str, now_ts: int, *, observed: str,
+                       expected: str, behaviour_id: str = "",
+                       environment: str = "", reproduction: str = "",
+                       candidate_areas: list | None = None,
+                       confidence: str = "low") -> dict:
+    return _call("tm_record_observation", {
+        "now": now, "now_ts": now_ts, "observed": observed,
+        "expected": expected, "behaviour_id": behaviour_id,
+        "environment": environment, "reproduction": reproduction,
+        "candidate_areas": candidate_areas or [],
+        "confidence": confidence})["observation"]
+
+
+def close_observation(records: list, observation_id: str, *,
+                      behaviour_retest_passed: bool,
+                      user_confirmed: bool, contract_ref: str = "",
+                      now: str = "") -> dict:
+    return _call("tm_close_observation", {
+        "records": records, "observation_id": observation_id,
+        "behaviour_retest_passed": behaviour_retest_passed,
+        "user_confirmed": user_confirmed, "contract_ref": contract_ref,
+        "now": now})
+
+
+def project(behaviour_id: str, observation: dict) -> dict:
+    return _call("tm_project", {"behaviour_id": behaviour_id,
+                                "observation": observation})
+
+
+def aggregate(entries: list) -> list:
+    return _call("tm_aggregate", {"entries": entries})["rows"]

src/rust/testing_mode/Cargo.lock

@@ -2,6 +2,28 @@
 # It is not intended for manual editing.
 version = 4
 
+[[package]]
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
+[[package]]
+name = "hashbrown"
+version = "0.17.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+
+[[package]]
+name = "indexmap"
+version = "2.14.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
+dependencies = [
+ "equivalent",
+ "hashbrown",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.18"
@@ -67,6 +89,7 @@ version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
+ "indexmap",
  "itoa",
  "memchr",
  "serde",

src/rust/testing_mode/Cargo.toml

@@ -7,4 +7,8 @@ edition = "2021"
 crate-type = ["rlib", "cdylib", "staticlib"]
 
 [dependencies]
-serde_json = "1"
+# preserve_order: Python-dict insertion-order semantics for the session
+# state map and the ledger payloads passed through verbatim;
+# float_roundtrip: correctly-rounded float parsing (floats ride along in
+# pass-through state and ledger records).
+serde_json = { version = "1", features = ["preserve_order", "float_roundtrip"] }

src/rust/testing_mode/src/branch_battery.rs

@@ -0,0 +1,140 @@
+//! testing_mode branch battery — parity vs the frozen Python-oracle truth
+//! (frozen_tm_truth.json: 40 cases across every public function and every
+//! branch — default-state shape (schema/scope/empty-project), 19L next-item
+//! selection (paused gate, empty queue, first-of-two, current_item
+//! overwrite/clear), 19M record bookkeeping (pass/fail append, order
+//! preservation), pause/resume idempotence, blocker notes with the
+//! 200-CHAR truncation, observation records (defaults, full fields,
+//! candidate-areas None coercion), close_observation closure lattice
+//! (retest AND user-confirmed, green-without-user NOT closed,
+//! confirm-without-retest NOT closed, contract_ref, unknown id → result
+//! null, sibling records preserved), forge projection title/body (60-char
+//! expected truncation, missing-field defaults, missing observation_id →
+//! Python str(None) "None"), aggregate row shaping (no-ledger note row,
+//! summary pass-through, mixed, empty)).
+//! Frozen 2026-10-12 via scripts/gen_testing_mode_truth.py with
+//! RUNE_TM_CORE=python pinned (poisoned-truth law) and the wall clock
+//! frozen at 2026-10-12T12:00:00Z (epoch 1791806400). Embedded VERBATIM —
+//! never transcribed. The `forge_dry_run`, `load` and `save` groups are
+//! facade-only (FILE IO + adapter call stay in Python) and are not
+//! replayed here.
+
+use crate::{
+    aggregate, close_observation, default_state, next_item, note_blocker, pause, project,
+    record, record_observation, resume,
+};
+use serde_json::{json, Map, Value};
+
+const TRUTH: &str = include_str!("frozen_tm_truth.json");
+
+fn cases(group: &str) -> Vec<Value> {
+    let truth: Value = serde_json::from_str(TRUTH).unwrap();
+    truth["cases"][group].as_array().unwrap().clone()
+}
+
+fn state_of(i: &Value) -> Map<String, Value> {
+    i["state"].as_object().cloned().unwrap_or_default()
+}
+
+fn replay(group: &str, f: impl Fn(&Value) -> Value) {
+    for case in cases(group) {
+        let got = f(&case["input"]);
+        assert_eq!(
+            &got,
+            &case["expected"],
+            "[{}] case '{}' diverged",
+            group,
+            case["name"].as_str().unwrap()
+        );
+    }
+}
+
+#[test]
+fn default_state_parity_battery() {
+    replay("default_state", |i| {
+        json!({"state": default_state(
+            &i["project"], &i["scope_component"], i["now"].as_str().unwrap(),
+        )})
+    });
+}
+
+#[test]
+fn next_item_parity_battery() {
+    replay("next_item", |i| {
+        let state = state_of(i);
+        let outstanding: Vec<Value> = i["outstanding"].as_array().unwrap().clone();
+        let (state, item) = next_item(&state, &outstanding);
+        json!({"state": state, "item": item})
+    });
+}
+
+#[test]
+fn record_parity_battery() {
+    replay("record", |i| {
+        let state = state_of(i);
+        json!({"state": record(&state, &i["rec"], i["passed"].as_bool().unwrap(),
+                               i["now"].as_str().unwrap())})
+    });
+}
+
+#[test]
+fn pause_parity_battery() {
+    replay("pause", |i| json!({"state": pause(&state_of(i))}));
+}
+
+#[test]
+fn resume_parity_battery() {
+    replay("resume", |i| json!({"state": resume(&state_of(i))}));
+}
+
+#[test]
+fn note_blocker_parity_battery() {
+    replay("note_blocker", |i| {
+        json!({"state": note_blocker(&state_of(i), &i["behaviour_id"],
+                                     i["note"].as_str().unwrap(), i["now"].as_str().unwrap())})
+    });
+}
+
+#[test]
+fn record_observation_parity_battery() {
+    replay("record_observation", |i| {
+        let s = |k: &str| i[k].as_str().unwrap_or("");
+        json!({"observation": record_observation(
+            s("now"), i["now_ts"].as_i64().unwrap_or(0),
+            s("observed"), s("expected"), s("behaviour_id"),
+            s("environment"), s("reproduction"),
+            &i["candidate_areas"], s("confidence"),
+        )})
+    });
+}
+
+#[test]
+fn close_observation_parity_battery() {
+    replay("close_observation", |i| {
+        let records: Vec<Value> = i["records"].as_array().unwrap().clone();
+        let (records, result) = close_observation(
+            &records, &i["observation_id"],
+            i["behaviour_retest_passed"].as_bool().unwrap(),
+            i["user_confirmed"].as_bool().unwrap(),
+            &i["contract_ref"], i["now"].as_str().unwrap(),
+        );
+        json!({"records": records, "result": result})
+    });
+}
+
+#[test]
+fn project_parity_battery() {
+    replay("project", |i| {
+        let observation = i["observation"].as_object().cloned().unwrap_or_default();
+        let (title, body) = project(&i["behaviour_id"], &observation);
+        json!({"title": title, "body": body})
+    });
+}
+
+#[test]
+fn aggregate_parity_battery() {
+    replay("aggregate", |i| {
+        let entries: Vec<Value> = i["entries"].as_array().unwrap().clone();
+        json!({"rows": aggregate(&entries)})
+    });
+}

src/rust/testing_mode/src/lib.rs

@@ -1,8 +1,483 @@
 //! testing_mode — testing mode state and project scope (W6).
-pub fn get_scope(project: &str, scope_component: &str) -> serde_json::Value {
-    serde_json::json!({
+//!
+//! FULL parity port of src/rune/testing_mode.py (dichotomy wave, #2479
+//! tranche). The Behaviour Validation Ledger is DURABLE TRUTH; a testing
+//! session is only coordination state (rule 19N) — pausable, resumable,
+//! never the authority. The queue is DYNAMIC (rule 19M): a fix for
+//! behaviour B may invalidate confirmed behaviour A via structural
+//! closure; unrelated hardened behaviours stay trusted.
+//!
+//! FFI (doc-envelope law): every symbol takes one JSON doc and returns
+//! {"ok": true, "value": ...} | {"ok": false, "err": ...}:
+//!   * tm_default_state({project, scope_component, now}) → {state}
+//!   * tm_next_item({state, outstanding}) → {state, item}
+//!   * tm_record({state, rec, passed, now}) → {state}
+//!   * tm_pause({state}) → {state}
+//!   * tm_resume({state}) → {state}
+//!   * tm_note_blocker({state, behaviour_id, note, now}) → {state}
+//!   * tm_record_observation({now, now_ts, observed, expected,
+//!     behaviour_id, environment, reproduction, candidate_areas,
+//!     confidence}) → {observation}
+//!   * tm_close_observation({records, observation_id,
+//!     behaviour_retest_passed, user_confirmed, contract_ref, now})
+//!     → {records, result}   (result null → unknown observation)
+//!   * tm_project({behaviour_id, observation}) → {title, body}
+//!   * tm_aggregate({entries}) → {rows}
+//!
+//! No-clock law: `now` (UTC ISO-8601) and `now_ts` (epoch seconds —
+//! the oracle's obs-{int(time.time())} id) are always arguments to the
+//! stamping calls; the kernel never reads the wall clock.
+//!
+//! Boundary: the session FILE IO (TestingSession._load hit branch /
+//! save), the observations.jsonl append/rewrite, the BehaviourLedger
+//! read (outstanding / record_real_world), the ProjectRegistry IO and
+//! the forge adapter call stay in the Python facade — the kernel is a
+//! pure function of their results.
+//!
+//! No tm_free_string export — sibling crates sharing the Python process
+//! already export identical free symbols; callers free through
+//! capability_scoring's cs_free_string.
+
+use serde_json::{json, Map, Value};
+use std::os::raw::c_char;
+
+pub const SESSION_SCHEMA_VERSION: i64 = 1;
+
+// ── Python scalar semantics ─────────────────────────────────────────────────
+
+/// Python truthiness over a JSON value.
+fn py_truthy(v: &Value) -> bool {
+    match v {
+        Value::Null => false,
+        Value::Bool(b) => *b,
+        Value::Number(n) => n.as_f64().map(|f| f != 0.0).unwrap_or(false),
+        Value::String(s) => !s.is_empty(),
+        Value::Array(a) => !a.is_empty(),
+        Value::Object(o) => !o.is_empty(),
+    }
+}
+
+/// Python str() over a JSON value for f-string interpolation (the
+/// Observation: line renders str(None) → "None" when the key is absent).
+fn py_str(v: &Value) -> String {
+    match v {
+        Value::Null => "None".to_string(),
+        Value::Bool(true) => "True".to_string(),
+        Value::Bool(false) => "False".to_string(),
+        Value::String(s) => s.clone(),
+        other => other.to_string(),
+    }
+}
+
+/// `observation.get(key, '')` rendered through Python str semantics —
+/// missing → "", present null → "None" (str(None)), present string → the
+/// string verbatim.
+fn py_get_str_or_empty(observation: &Map<String, Value>, key: &str) -> String {
+    match observation.get(key) {
+        Some(v) => py_str(v),
+        None => String::new(),
+    }
+}
+
+/// `observation.get(key)` with NO default, rendered through Python str
+/// semantics — missing → "None" (str(None)), present null → "None".
+fn py_get_str_or_none(observation: &Map<String, Value>, key: &str) -> String {
+    match observation.get(key) {
+        Some(v) => py_str(v),
+        None => "None".to_string(),
+    }
+}
+
+/// `observation.get(key, '')` when the value is REQUIRED to be a string
+/// by the oracle (the [:60] title slice) — null/missing collapse to "".
+fn py_str_field(observation: &Map<String, Value>, key: &str) -> String {
+    observation
+        .get(key)
+        .and_then(Value::as_str)
+        .unwrap_or("")
+        .to_string()
+}
+
+// ── session state machine (19N/19L/19M) ─────────────────────────────────────
+
+/// `_load` miss branch — the fresh coordination state. Session state is
+/// never spec truth: it carries only ids, flags and bookkeeping lists.
+pub fn default_state(project: &Value, scope_component: &Value, now: &str) -> Value {
+    json!({
+        "schema_version": SESSION_SCHEMA_VERSION,
         "project": project,
         "scope_component": scope_component,
-        "forge_available": false,
+        "current_item": Value::Null,
+        "completed_this_session": [],
+        "newly_invalidated": [],
+        "blockers": [],
+        "paused": false,
+        "updated_at": now,
     })
 }
+
+/// `next_item` — 19L next eligible item. Paused sessions dequeue
+/// nothing (state untouched); otherwise the first outstanding item is
+/// selected and its behaviour_id becomes current_item.
+pub fn next_item(
+    state: &Map<String, Value>,
+    outstanding: &[Value],
+) -> (Map<String, Value>, Value) {
+    if py_truthy(state.get("paused").unwrap_or(&Value::Null)) {
+        return (state.clone(), Value::Null);
+    }
+    let mut out = state.clone();
+    let item = outstanding.first().cloned().unwrap_or(Value::Null);
+    let current = item
+        .as_object()
+        .and_then(|o| o.get("behaviour_id"))
+        .cloned()
+        .unwrap_or(Value::Null);
+    out.insert("current_item".to_string(), current);
+    (out, item)
+}
+
+/// `record` — append the completed-session bookkeeping entry (the
+/// USER_CONFIRMED branch is a deliberate no-op: the queue re-derives).
+pub fn record(
+    state: &Map<String, Value>,
+    rec: &Value,
+    passed: bool,
+    now: &str,
+) -> Map<String, Value> {
+    let mut out = state.clone();
+    if let Some(list) = out
+        .get_mut("completed_this_session")
+        .and_then(Value::as_array_mut)
+    {
+        list.push(json!({
+            "behaviour_id": rec.get("behaviour_id").cloned().unwrap_or(Value::Null),
+            "passed": passed,
+            "at": now,
+        }));
+    }
+    out
+}
+
+/// `pause` — 19N coordination only.
+pub fn pause(state: &Map<String, Value>) -> Map<String, Value> {
+    let mut out = state.clone();
+    out.insert("paused".to_string(), Value::Bool(true));
+    out
+}
+
+/// `resume` — 19N coordination only.
+pub fn resume(state: &Map<String, Value>) -> Map<String, Value> {
+    let mut out = state.clone();
+    out.insert("paused".to_string(), Value::Bool(false));
+    out
+}
+
+/// `note_blocker` — note truncated to the oracle's 200 chars (chars,
+/// not bytes: note[:200] over a Python str).
+pub fn note_blocker(
+    state: &Map<String, Value>,
+    behaviour_id: &Value,
+    note: &str,
+    now: &str,
+) -> Map<String, Value> {
+    let mut out = state.clone();
+    if let Some(list) = out.get_mut("blockers").and_then(Value::as_array_mut) {
+        list.push(json!({
+            "behaviour_id": behaviour_id,
+            "note": note.chars().take(200).collect::<String>(),
+            "at": now,
+        }));
+    }
+    out
+}
+
+// ── observations (19O) ──────────────────────────────────────────────────────
+
+/// `record_observation` — durable real-world failure report; candidate
+/// areas are HYPOTHESES, not declared root cause (rule 19O). closure is
+/// set only by close_observation.
+#[allow(clippy::too_many_arguments)]
+pub fn record_observation(
+    now: &str,
+    now_ts: i64,
+    observed: &str,
+    expected: &str,
+    behaviour_id: &str,
+    environment: &str,
+    reproduction: &str,
+    candidate_areas: &Value,
+    confidence: &str,
+) -> Value {
+    json!({
+        "observation_id": format!("obs-{}", now_ts),
+        "observed": observed,
+        "expected": expected,
+        "behaviour_id": behaviour_id,
+        "environment": environment,
+        "reproduction": reproduction,
+        "candidate_areas": candidate_areas,
+        "confidence": confidence,
+        "created_at": now,
+        "closure": Value::Null,
+    })
+}
+
+/// `close_observation` — closure returns to the ORIGINAL observed
+/// scenario (rule 19O): code changed ≠ closure; automated green ≠
+/// closure. Closes only when retested AND user-confirmed. Unknown ids
+/// yield result null (the facade raises the ValueError).
+pub fn close_observation(
+    records: &[Value],
+    observation_id: &Value,
+    behaviour_retest_passed: bool,
+    user_confirmed: bool,
+    contract_ref: &Value,
+    now: &str,
+) -> (Vec<Value>, Value) {
+    let mut out = Vec::with_capacity(records.len());
+    let mut result = Value::Null;
+    for rec in records {
+        let mut rec = rec.clone();
+        if rec.get("observation_id") == Some(observation_id) {
+            rec["closure"] = json!({
+                "behaviour_retest_passed": behaviour_retest_passed,
+                "user_confirmed": user_confirmed,
+                "contract_ref": contract_ref,
+                "closed": behaviour_retest_passed && user_confirmed,
+                "at": now,
+            });
+            result = rec.clone();
+        }
+        out.push(rec);
+    }
+    (out, result)
+}
+
+// ── forge projection (19P) ──────────────────────────────────────────────────
+
+/// `project_failure_to_forge` title/body construction. The forge record
+/// is a PROJECTION — canonical truth stays in the Behaviour Ledger; the
+/// adapter call itself stays in the facade.
+pub fn project(behaviour_id: &Value, observation: &Map<String, Value>) -> (String, String) {
+    let behaviour = py_str(behaviour_id);
+    let title = format!(
+        "[testing] {}: {}",
+        behaviour,
+        truncate_chars(&py_str_field(observation, "expected"), 60)
+    );
+    let body = format!(
+        "Observed: {}\nExpected: {}\nBehaviour: {}\nObservation: {}\nReproduction: {}",
+        py_get_str_or_empty(observation, "observed"),
+        py_get_str_or_empty(observation, "expected"),
+        behaviour,
+        py_get_str_or_none(observation, "observation_id"),
+        py_get_str_or_empty(observation, "reproduction"),
+    );
+    (title, body)
+}
+
+fn truncate_chars(s: &str, n: usize) -> String {
+    s.chars().take(n).collect()
+}
+
+// ── aggregate (19L) ─────────────────────────────────────────────────────────
+
+/// `aggregate` row shaping over facade-fetched per-project summaries —
+/// summary null → "no behaviour ledger yet"; otherwise the summary's
+/// outstanding count + state breakdown pass through.
+pub fn aggregate(entries: &[Value]) -> Vec<Value> {
+    entries
+        .iter()
+        .map(|entry| {
+            let project = entry
+                .get("project_id")
+                .cloned()
+                .unwrap_or(Value::Null);
+            match entry.get("summary") {
+                None | Some(Value::Null) => json!({
+                    "project": project,
+                    "outstanding": Value::Null,
+                    "note": "no behaviour ledger yet",
+                }),
+                Some(s) => json!({
+                    "project": project,
+                    "outstanding": s.get("outstanding").cloned().unwrap_or(Value::Null),
+                    "states": s.get("states").cloned().unwrap_or(Value::Null),
+                }),
+            }
+        })
+        .collect()
+}
+
+// ── FFI (doc-envelope law) ──────────────────────────────────────────────────
+
+fn read_arg(ptr: *const c_char) -> Result<Value, String> {
+    if ptr.is_null() {
+        return Err("null argument pointer".to_string());
+    }
+    let s = unsafe { std::ffi::CStr::from_ptr(ptr) }
+        .to_str()
+        .map_err(|e| e.to_string())?;
+    serde_json::from_str(s).map_err(|e| e.to_string())
+}
+
+fn envelope(r: Result<Value, String>) -> *mut c_char {
+    let v = match r {
+        Ok(x) => json!({"ok": true, "value": x}),
+        Err(e) => json!({"ok": false, "err": e}),
+    };
+    std::ffi::CString::new(serde_json::to_string(&v).unwrap_or_default())
+        .map(|c| c.into_raw())
+        .unwrap_or(std::ptr::null_mut())
+}
+
+fn state_of(a: &Value) -> Map<String, Value> {
+    a.get("state")
+        .and_then(Value::as_object)
+        .cloned()
+        .unwrap_or_default()
+}
+
+/// tm_default_state({project, scope_component, now}) → {state}
+#[no_mangle]
+pub extern "C" fn tm_default_state(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        json!({"state": default_state(
+            a.get("project").unwrap_or(&Value::Null),
+            a.get("scope_component").unwrap_or(&Value::Null),
+            a.get("now").and_then(Value::as_str).unwrap_or(""),
+        )})
+    }))
+}
+
+/// tm_next_item({state, outstanding}) → {state, item}
+#[no_mangle]
+pub extern "C" fn tm_next_item(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        let state = state_of(&a);
+        let outstanding: Vec<Value> = a
+            .get("outstanding")
+            .and_then(Value::as_array)
+            .cloned()
+            .unwrap_or_default();
+        let (state, item) = next_item(&state, &outstanding);
+        json!({"state": state, "item": item})
+    }))
+}
+
+/// tm_record({state, rec, passed, now}) → {state}
+#[no_mangle]
+pub extern "C" fn tm_record(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        let state = state_of(&a);
+        let rec = a.get("rec").cloned().unwrap_or(Value::Null);
+        let passed = a.get("passed").map(py_truthy).unwrap_or(false);
+        let now = a.get("now").and_then(Value::as_str).unwrap_or("");
+        json!({"state": record(&state, &rec, passed, now)})
+    }))
+}
+
+/// tm_pause({state}) → {state}
+#[no_mangle]
+pub extern "C" fn tm_pause(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| json!({"state": pause(&state_of(&a))})))
+}
+
+/// tm_resume({state}) → {state}
+#[no_mangle]
+pub extern "C" fn tm_resume(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| json!({"state": resume(&state_of(&a))})))
+}
+
+/// tm_note_blocker({state, behaviour_id, note, now}) → {state}
+#[no_mangle]
+pub extern "C" fn tm_note_blocker(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        let state = state_of(&a);
+        let behaviour_id = a.get("behaviour_id").cloned().unwrap_or(Value::Null);
+        let note = a.get("note").and_then(Value::as_str).unwrap_or("");
+        let now = a.get("now").and_then(Value::as_str).unwrap_or("");
+        json!({"state": note_blocker(&state, &behaviour_id, note, now)})
+    }))
+}
+
+/// tm_record_observation({now, now_ts, observed, expected, behaviour_id,
+/// environment, reproduction, candidate_areas, confidence}) → {observation}
+#[no_mangle]
+pub extern "C" fn tm_record_observation(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        let s = |k: &str| a.get(k).and_then(Value::as_str).unwrap_or("");
+        let default_areas = json!([]);
+        json!({"observation": record_observation(
+            s("now"),
+            a.get("now_ts").and_then(Value::as_i64).unwrap_or(0),
+            s("observed"),
+            s("expected"),
+            s("behaviour_id"),
+            s("environment"),
+            s("reproduction"),
+            a.get("candidate_areas").unwrap_or(&default_areas),
+            s("confidence"),
+        )})
+    }))
+}
+
+/// tm_close_observation({records, observation_id, behaviour_retest_passed,
+/// user_confirmed, contract_ref, now}) → {records, result}
+#[no_mangle]
+pub extern "C" fn tm_close_observation(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        let records: Vec<Value> = a
+            .get("records")
+            .and_then(Value::as_array)
+            .cloned()
+            .unwrap_or_default();
+        let (records, result) = close_observation(
+            &records,
+            a.get("observation_id").unwrap_or(&Value::Null),
+            a.get("behaviour_retest_passed")
+                .and_then(Value::as_bool)
+                .unwrap_or(false),
+            a.get("user_confirmed")
+                .and_then(Value::as_bool)
+                .unwrap_or(false),
+            a.get("contract_ref").unwrap_or(&Value::Null),
+            a.get("now").and_then(Value::as_str).unwrap_or(""),
+        );
+        json!({"records": records, "result": result})
+    }))
+}
+
+/// tm_project({behaviour_id, observation}) → {title, body}
+#[no_mangle]
+pub extern "C" fn tm_project(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        let observation = a
+            .get("observation")
+            .and_then(Value::as_object)
+            .cloned()
+            .unwrap_or_default();
+        let (title, body) = project(
+            a.get("behaviour_id").unwrap_or(&Value::Null),
+            &observation,
+        );
+        json!({"title": title, "body": body})
+    }))
+}
+
+/// tm_aggregate({entries}) → {rows}
+#[no_mangle]
+pub extern "C" fn tm_aggregate(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        let entries: Vec<Value> = a
+            .get("entries")
+            .and_then(Value::as_array)
+            .cloned()
+            .unwrap_or_default();
+        json!({"rows": aggregate(&entries)})
+    }))
+}
+
+#[cfg(test)]
+mod branch_battery;