Nick
cf7a11e993
Dichotomy wave #2479, contract factory-heavy-ro-sig. Name-preserving ports of the five audit-missing observation/routing-signal fns into gaps_signals.go with a 35-row frozen oracle battery (truth_signals.json, hermetic generator, byte-identical regen) and 35/35 live parity (shared corpus between generator and parity checker — no drift). All IO behind injectable seams; recorder-down crash parity pinned (record_failure sits outside the oracle's try). Audit 29/68 → 35/68. Divergences documented in verification/ro-sig-summary.md.
348 lines
12 KiB
Python
348 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""gen_operator_signals_truth.py — freeze the observation/routing-SIGNAL
|
|
family gap truth for rune_operator (dichotomy wave #2479, contract
|
|
factory-heavy-ro-sig).
|
|
|
|
Covers the five audit-missing signal functions against the LIVE oracle
|
|
(src/rune/rune_operator.py):
|
|
|
|
_filter_safety_noise / _sorted_tool_names / _build_model_router
|
|
(module fn + RuneOperator method — one name) / _wire_auto_filer /
|
|
_reconcile_guarded
|
|
|
|
Hermetic laws (gen_tool_executor_lifecycle_truth.py precedent):
|
|
- pure fns run live, verbatim;
|
|
- IO fns run behind process-scoped stubs (forge_client,
|
|
failure_auto_filer, subsystem_health, event_bus._store_path,
|
|
threading.Thread, the rune.rune_operator logger) patched in
|
|
try/finally — every branch is scripted, no socket, no subprocess, no
|
|
process-global wiring leaks;
|
|
- no clock, no uuid, no filesystem paths in the corpus (the spool path
|
|
is a frozen constant string);
|
|
- action grammar (BOTH sides emit exactly these strings):
|
|
forge_client | spool:<path> | filer:<client>,<path> |
|
|
set_auto_filer:<marker> | thread:rune-autofile-reconcile|true |
|
|
reconcile_target:<marker> | reconcile:<marker> | log:<message>
|
|
|
|
str(e) law: recorded failures carry the exception MESSAGE only (type
|
|
names stripped by str(e)); the crash-class row freezes the recorder-down
|
|
propagation ("raises").
|
|
|
|
Byte-identical regeneration: no timestamps; re-run and diff.
|
|
|
|
Usage: python3 scripts/gen_operator_signals_truth.py [--out PATH]
|
|
"""
|
|
|
|
import copy
|
|
import json
|
|
import logging
|
|
import os
|
|
import sys
|
|
import threading
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parent.parent
|
|
sys.path.insert(0, str(REPO / "src"))
|
|
|
|
from rune import event_bus as py_event_bus # noqa: E402
|
|
from rune import failure_auto_filer as py_faf # noqa: E402
|
|
from rune import forge_client as py_forge # noqa: E402
|
|
from rune import rune_operator as ro # noqa: E402 — the live oracle
|
|
from rune import subsystem_health as py_health # noqa: E402
|
|
|
|
SPOOL_PATH = "/journal/events.jsonl"
|
|
|
|
|
|
# ── case corpus ──────────────────────────────────────────────────────
|
|
|
|
def build_cases():
|
|
cases = []
|
|
|
|
def add(op, name, **kw):
|
|
cases.append({"op": op, "name": name, **kw})
|
|
|
|
# _filter_safety_noise (pure)
|
|
add("filter_safety", "mixed_noise",
|
|
thinking="Let me check the config.\nAs an AI, I follow safety guidelines.\nReading the file now.")
|
|
add("filter_safety", "blank_preserve", thinking="line1\n\n \nline2")
|
|
add("filter_safety", "blank_collapse", thinking="a\n\n\n\nb")
|
|
add("filter_safety", "all_noise",
|
|
thinking="As an AI language model, I cannot provide that.\nThis request goes against safety policy.")
|
|
add("filter_safety", "empty", thinking="")
|
|
add("filter_safety", "case_insensitive", thinking="AS AN AI I MUST DECLINE")
|
|
add("filter_safety", "however_line", thinking="However, I will proceed carefully\nreal line")
|
|
add("filter_safety", "interior_noise", thinking="first\nI apologize but I can't\nlast")
|
|
add("filter_safety", "edge_strip", thinking=" \nkeep me\n ")
|
|
add("filter_safety", "collapse_after_strip", thinking="keep\n\n\n\n\nmore")
|
|
add("filter_safety", "unicode", thinking="héllo wörld\nAs an AI, no\nünïcödé kept")
|
|
|
|
# _sorted_tool_names (pure)
|
|
add("sorted_tool_names", "mixed_drop_dupes",
|
|
value=["read_file", 5, None, "write_file", "read_file"])
|
|
add("sorted_tool_names", "sorted_passthrough", value=["a", "b", "c"])
|
|
add("sorted_tool_names", "reverse", value=["z", "y", "x"])
|
|
add("sorted_tool_names", "dupes_only", value=["b", "a", "b", "a"])
|
|
add("sorted_tool_names", "empty_list", value=[])
|
|
add("sorted_tool_names", "none", value=None)
|
|
add("sorted_tool_names", "string_input", value="read_file")
|
|
add("sorted_tool_names", "dict_input", value={"a": 1})
|
|
add("sorted_tool_names", "int_input", value=42)
|
|
add("sorted_tool_names", "case_distinct", value=["Read", "read", "READ"])
|
|
add("sorted_tool_names", "nested_list", value=[["x"], "y"])
|
|
|
|
# _build_model_router (scripted seam)
|
|
add("build_model_router", "unavailable", mode="unavailable")
|
|
add("build_model_router", "ok_dict", mode="ok",
|
|
result={"router": "ok", "ranked": ["qwen3:14b", "qwen2.5:7b"]})
|
|
add("build_model_router", "ok_list", mode="ok", result=["floor", "measured"])
|
|
add("build_model_router", "factory_raises", mode="raises",
|
|
error="profiles file unreadable")
|
|
add("build_model_router", "recorder_down", mode="recorder_raises",
|
|
error="factory boom", recorder_error="recorder down")
|
|
|
|
# _wire_auto_filer (scripted seams)
|
|
add("wire_auto_filer", "disabled", env="1")
|
|
add("wire_auto_filer", "happy")
|
|
add("wire_auto_filer", "forge_raises", forge_error="forge unconfigured")
|
|
add("wire_auto_filer", "spool_path_raises", spool_error="journal path unavailable")
|
|
add("wire_auto_filer", "filer_ctor_raises", filer_error="filer ctor boom")
|
|
add("wire_auto_filer", "set_auto_filer_raises", set_raises=True)
|
|
|
|
# _reconcile_guarded (scripted filer)
|
|
add("reconcile_guarded", "ok")
|
|
add("reconcile_guarded", "reconcile_raises", error="spool journal locked")
|
|
|
|
return cases
|
|
|
|
|
|
# ── oracle runner (live, patched in try/finally) ─────────────────────
|
|
|
|
class _CapHandler(logging.Handler):
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.msgs = []
|
|
|
|
def emit(self, record):
|
|
self.msgs.append(record.getMessage())
|
|
|
|
|
|
class _StubSpool:
|
|
def __init__(self, path):
|
|
self.path = str(path)
|
|
ACT.append("spool:" + self.path)
|
|
|
|
|
|
class _StubFiler:
|
|
marker = "filer-1"
|
|
|
|
def __init__(self, error=None):
|
|
self._error = error
|
|
|
|
def reconcile(self):
|
|
ACT.append("reconcile:" + self.marker)
|
|
if self._error:
|
|
raise RuntimeError(self._error)
|
|
return {"replayed": 0, "filed": 0, "failed": 0, "skipped": 0}
|
|
|
|
|
|
class _FakeThread:
|
|
"""threading.Thread stand-in: records (name, daemon), runs target
|
|
synchronously so the action order is deterministic."""
|
|
target = None
|
|
args = ()
|
|
daemon = None
|
|
name = None
|
|
|
|
def __init__(self, target=None, args=(), daemon=None, name=None):
|
|
self.target, self.args = target, args
|
|
self.daemon, self.name = daemon, name
|
|
|
|
def start(self):
|
|
ACT.append(f"thread:{self.name}|{str(self.daemon).lower()}")
|
|
self.target(*self.args)
|
|
|
|
|
|
class _LogStub:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def exception(self, msg):
|
|
self.calls.append(msg)
|
|
|
|
|
|
ACT = [] # action sink, reset per case
|
|
_REAL_RECONCILE = ro._reconcile_guarded # captured before any patching
|
|
|
|
|
|
def run_oracle(cases):
|
|
out = {}
|
|
op_logger = logging.getLogger("rune.rune_operator")
|
|
op_logger.addHandler(_HANDLER)
|
|
saved_level = op_logger.level
|
|
# log.info rows (the disabled branch) need the logger at INFO or the
|
|
# record is dropped before any handler sees it.
|
|
if saved_level == logging.NOTSET or saved_level > logging.INFO:
|
|
op_logger.setLevel(logging.INFO)
|
|
|
|
saved_globals = {
|
|
"bmr": ro.build_model_router,
|
|
"rec": py_health.record_failure,
|
|
"setf": py_health.set_auto_filer,
|
|
"forge": py_forge.create_forge_client,
|
|
"sp": py_event_bus._store_path,
|
|
"faf": py_faf.FailureAutoFiler,
|
|
"spool": py_faf.JsonlSpool,
|
|
"log": py_faf._LOG,
|
|
"rg": ro._reconcile_guarded,
|
|
"thread": threading.Thread,
|
|
}
|
|
try:
|
|
for c in cases:
|
|
op, name = c["op"], c["name"]
|
|
ACT.clear()
|
|
_HANDLER.msgs.clear()
|
|
out[(op, name)] = _run_one(c)
|
|
finally:
|
|
ro.build_model_router = saved_globals["bmr"]
|
|
py_health.record_failure = saved_globals["rec"]
|
|
py_health.set_auto_filer = saved_globals["setf"]
|
|
py_forge.create_forge_client = saved_globals["forge"]
|
|
py_event_bus._store_path = saved_globals["sp"]
|
|
py_faf.FailureAutoFiler = saved_globals["faf"]
|
|
py_faf.JsonlSpool = saved_globals["spool"]
|
|
py_faf._LOG = saved_globals["log"]
|
|
ro._reconcile_guarded = saved_globals["rg"]
|
|
threading.Thread = saved_globals["thread"]
|
|
op_logger.removeHandler(_HANDLER)
|
|
op_logger.setLevel(saved_level)
|
|
os.environ.pop("RUNE_AUTOFILE_OFF", None)
|
|
return out
|
|
|
|
|
|
def _run_one(c):
|
|
op = c["op"]
|
|
|
|
if op == "filter_safety":
|
|
return ro._filter_safety_noise(c["thinking"])
|
|
|
|
if op == "sorted_tool_names":
|
|
return ro._sorted_tool_names(c["value"])
|
|
|
|
if op == "build_model_router":
|
|
recorded = []
|
|
mode = c["mode"]
|
|
if mode == "unavailable":
|
|
ro.build_model_router = None
|
|
elif mode == "ok":
|
|
ro.build_model_router = lambda: copy.deepcopy(c["result"])
|
|
else:
|
|
def _factory_boom():
|
|
raise RuntimeError(c["error"])
|
|
ro.build_model_router = _factory_boom
|
|
|
|
def _rec(subsystem, error, tier=1, registry=None, context="", filer=None):
|
|
if c.get("recorder_error"):
|
|
raise RuntimeError(c["recorder_error"])
|
|
recorded.append({
|
|
"subsystem": subsystem,
|
|
"message": str(error),
|
|
"tier": tier,
|
|
"context": context,
|
|
})
|
|
py_health.record_failure = _rec
|
|
try:
|
|
result = ro._build_model_router()
|
|
return {"result": result, "recorded": recorded}
|
|
except Exception as exc: # recorder-down crash class propagates
|
|
return {"raises": str(exc)}
|
|
|
|
if op == "wire_auto_filer":
|
|
if c.get("env") == "1":
|
|
os.environ["RUNE_AUTOFILE_OFF"] = "1"
|
|
else:
|
|
os.environ.pop("RUNE_AUTOFILE_OFF", None)
|
|
|
|
def _forge():
|
|
ACT.append("forge_client")
|
|
if c.get("forge_error"):
|
|
raise RuntimeError(c["forge_error"])
|
|
return "stub-client"
|
|
|
|
def _store_path():
|
|
if c.get("spool_error"):
|
|
raise RuntimeError(c["spool_error"])
|
|
return SPOOL_PATH
|
|
|
|
def _faf_ctor(client, journal):
|
|
ACT.append(f"filer:{client},{journal.path}")
|
|
if c.get("filer_error"):
|
|
raise RuntimeError(c["filer_error"])
|
|
return _StubFiler()
|
|
|
|
def _set(filer):
|
|
ACT.append("set_auto_filer:" + getattr(filer, "marker", "?"))
|
|
if c.get("set_raises"):
|
|
raise TypeError("set_auto_filer exploded")
|
|
|
|
def _rg(filer):
|
|
ACT.append("reconcile_target:" + getattr(filer, "marker", "?"))
|
|
|
|
py_forge.create_forge_client = _forge
|
|
py_event_bus._store_path = _store_path
|
|
py_faf.JsonlSpool = _StubSpool
|
|
py_faf.FailureAutoFiler = _faf_ctor
|
|
py_health.set_auto_filer = _set
|
|
ro._reconcile_guarded = _rg
|
|
threading.Thread = _FakeThread
|
|
|
|
ro._wire_auto_filer()
|
|
return {"actions": ACT[:] + ["log:" + m for m in handler_msgs()]}
|
|
|
|
if op == "reconcile_guarded":
|
|
filer = _StubFiler(error=c.get("error"))
|
|
log_stub = _LogStub()
|
|
py_faf._LOG = log_stub
|
|
# the wire rows patch this global; run the REAL guarded reconcile
|
|
ro._reconcile_guarded = _REAL_RECONCILE
|
|
_REAL_RECONCILE(filer)
|
|
return {"actions": ACT[:] + ["log:" + m for m in log_stub.calls]}
|
|
|
|
raise ValueError(f"unknown op {op}")
|
|
|
|
|
|
_HANDLER = _CapHandler()
|
|
|
|
|
|
def handler_msgs():
|
|
return _HANDLER.msgs
|
|
|
|
|
|
def main():
|
|
cases = build_cases()
|
|
oracle = run_oracle(cases)
|
|
|
|
rows = []
|
|
for c in cases:
|
|
rows.append({**c, "want": oracle[(c["op"], c["name"])]})
|
|
|
|
doc = {
|
|
"_doc": (
|
|
"FROZEN ORACLE (rune_operator observation/routing-SIGNAL family "
|
|
"— factory-heavy-ro-sig). Pure fns verbatim; IO fns behind "
|
|
"scripted stubs (action grammar in the generator docstring; "
|
|
"str(e) law — messages carry no type prefix). Regenerate: "
|
|
"scripts/gen_operator_signals_truth.py"
|
|
),
|
|
"rows": rows,
|
|
}
|
|
|
|
out = Path(sys.argv[sys.argv.index("--out") + 1]) if "--out" in sys.argv else (
|
|
REPO / "src" / "go" / "operator-deep" / "truth_signals.json"
|
|
)
|
|
out.write_text(json.dumps(doc, indent=1, ensure_ascii=False) + "\n", encoding="utf-8")
|
|
print(f"wrote {out} ({len(rows)} rows)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|