788 lines
31 KiB
Python
788 lines
31 KiB
Python
#!/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)
|
|
|
|
|
|
# ── probe (literal-100% residue: _now / _tm_core) ────────────────────────────
|
|
|
|
def c_now_epoch_zero(rec):
|
|
rec_facade(rec, "now-epoch-zero", {"epoch": 0},
|
|
{"stamp": "1970-01-01T00:00:00Z"}, oracle_ret=None)
|
|
|
|
|
|
def c_now_frozen_epoch(rec):
|
|
rec_facade(rec, "now-frozen-epoch", {"epoch": FROZEN_TS},
|
|
{"stamp": FROZEN_NOW}, oracle_ret=None)
|
|
|
|
|
|
def c_now_negative_epoch(rec):
|
|
import time as _t
|
|
rec_facade(rec, "now-negative-epoch", {"epoch": -1},
|
|
{"stamp": _t.strftime("%Y-%m-%dT%H:%M:%SZ", _t.gmtime(-1))},
|
|
oracle_ret=None)
|
|
|
|
|
|
def c_now_far_future(rec):
|
|
import time as _t
|
|
rec_facade(rec, "now-far-future", {"epoch": 4102444800},
|
|
{"stamp": _t.strftime("%Y-%m-%dT%H:%M:%SZ", _t.gmtime(4102444800))},
|
|
oracle_ret=None)
|
|
|
|
|
|
def c_now_oracle_shape(rec):
|
|
# the facade's own _now() under the frozen rig renders the FROZEN stamp
|
|
rec_facade(rec, "now-facade-stamp", {},
|
|
{"stamp": oracle_mod._now()}, oracle_ret=None)
|
|
|
|
|
|
def c_tm_core_pins(rec):
|
|
import os as _os
|
|
rec_facade(rec, "tm-core-pin-semantics",
|
|
{"pins": ["python", None, "auto", "rust", ""]},
|
|
{"live": [p != "python" for p in ["python", None, "auto", "rust", ""]]},
|
|
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]
|
|
PROBE = [c_now_epoch_zero, c_now_frozen_epoch, c_now_negative_epoch,
|
|
c_now_far_future, c_now_oracle_shape, c_tm_core_pins]
|
|
|
|
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), ("probe", PROBE)]
|
|
|
|
# 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,
|
|
"probe": "tm_now",
|
|
}
|
|
|
|
# loader wrapper + slice of `expected` the unwrapped FFI value maps to
|
|
def _probe_kernel_call(i, ld):
|
|
"""The now-* cases ride tm_now; the facade-stamp case has no epoch
|
|
(the facade's own _now() under the frozen rig is the oracle — the
|
|
kernel pass asserts the FFI stamp for the frozen epoch instead);
|
|
the pin-semantics case exercises the probe mirror."""
|
|
import time as _t
|
|
if "epoch" in i:
|
|
return {"stamp": ld.now(i["epoch"])}
|
|
if "pins" in i:
|
|
return {"live": [ld.core_live(p) for p in i["pins"]]}
|
|
return {"stamp": ld.now(int(_t.mktime(_t.strptime(
|
|
"2026-10-12T12:00:00Z", "%Y-%m-%dT%H:%M:%SZ"))) - _t.timezone
|
|
if False else 1791806400)}
|
|
|
|
|
|
|
|
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"])),
|
|
"probe": ("probe", lambda w: w, _probe_kernel_call),
|
|
}
|
|
|
|
|
|
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())
|