Nick
19f93200c1
Port the last four audit-missing fns name-preserving into the crate: * _tg_core — the facade loader-probe twin (_cs_core law; 4 pin rows). * _now — the aware-UTC isoformat stamp with the clock as an argument (epoch micros; .ffffff omitted at microsecond 0; +00:00; 6 rows pin epoch-zero, pre-epoch negatives, leap-day-2024, whole-second). * list_gates — the listing projection moved kernel-side: count DESC stable sort (sqlite's tie order = id order, proven by the tie rows); truth group now carries rows-before input + expected. * main — the full argparse CLI dispatcher as a lib fn (task_persist MainOutcome law): choices/arg/--event/--data/--json parsing with the COLUMNS=80 usage wrap and exit-2 error classes, the required-cmd- before-unrecognized precedence quirk, both output modes per verb (PROMOTED/Noted/Observing, GATE lines, the 10-wide status/3-wide id list format, hook compact-json echo), and the in-memory DB-state application (matched UPDATE / autoincrement INSERT with promoted_at NULL column shape). 25 main rows frozen from the live oracle. serde_json gains preserve_order (row dicts ride sqlite column order — the --json renderer replays it); the rig's frozen stamp is now deterministic and the truth file keeps insertion order (sort_keys would lie about row key order). Battery 8/8 green (was 4); live parity 982/982 (32 truth + 950 fuzz); --replay with delegation live facade 67/67 kernel 67/67; clippy -D warnings clean; gen x2 byte-identical. No adjacent pytest exists for task_gate (plugin surface) — parity/replay/battery are the gates. Fn coverage 5/9 -> 9/9.
736 lines
27 KiB
Python
736 lines
27 KiB
Python
#!/usr/bin/env python3
|
|
"""gen_task_gate_truth.py — freeze/replay parity truth for task_gate
|
|
(dichotomy wave, #2479 tranche).
|
|
|
|
Freeze mode (default): exercises EVERY public function of the LIVE Python
|
|
oracle (src/rune/task_gate.py) across every branch and records
|
|
(kernel_input, expected_kernel_output, oracle_return) triples into
|
|
src/rust/task_gate/src/frozen_tg_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_TG_CORE=python BEFORE any import.
|
|
No-clock law: the wall clock is frozen by patching the datetime symbol in
|
|
task_gate, so every kernel payload carries its own `now` and replay is
|
|
deterministic regardless of run date.
|
|
|
|
Boundary: the sqlite IO (connect, the row reads, UPDATE/INSERT) and the
|
|
CLI main() stay in the Python facade — the kernel owns the verdict logic
|
|
(pattern extraction, regex gate matching, count/threshold/status
|
|
transitions, promotion decision, listing projection, hook echo).
|
|
list_gates' ORDER BY is decided by sqlite in the facade; the kernel
|
|
receives already-ordered rows.
|
|
"""
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
TRUTH_PATH = os.path.join(
|
|
REPO, "src", "rust", "task_gate", "src", "frozen_tg_truth.json"
|
|
)
|
|
REPLAY = "--replay" in sys.argv
|
|
|
|
if not REPLAY:
|
|
os.environ["RUNE_TG_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 task_gate as oracle_mod # noqa: E402
|
|
|
|
if REPLAY:
|
|
import tg_loader # noqa: E402 (kernel-level replay side)
|
|
|
|
FROZEN_NOW = "2026-10-11T23:00:00.123456+00:00"
|
|
|
|
|
|
class FrozenDatetime(datetime):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
assert tz is timezone.utc or tz is None
|
|
return datetime(2026, 10, 11, 23, 0, 0, 123456, tzinfo=timezone.utc)
|
|
|
|
|
|
oracle_mod.datetime = FrozenDatetime
|
|
|
|
_COLS = ("id", "instruction", "pattern", "count", "threshold", "status",
|
|
"promoted_at", "created_at", "last_seen")
|
|
|
|
|
|
def fresh_db():
|
|
tmp = tempfile.mkdtemp(prefix="tg_truth_")
|
|
oracle_mod.DB_PATH = os.path.join(tmp, "task_gate.db")
|
|
return oracle_mod.connect()
|
|
|
|
|
|
def seed_rows(pre_rows):
|
|
"""Seed explicit rows (dicts with any subset of the columns; count/
|
|
threshold/status/pattern/created_at/last_seen default like the oracle
|
|
would for a fresh row). Returns rows as full dicts in id order."""
|
|
db = fresh_db()
|
|
for i, r in enumerate(pre_rows, start=1):
|
|
db.execute(
|
|
"INSERT INTO instructions (id, instruction, pattern, count,"
|
|
" threshold, status, promoted_at, created_at, last_seen)"
|
|
" VALUES (?,?,?,?,?,?,?,?,?)",
|
|
(
|
|
r.get("id", i),
|
|
r.get("instruction", "seed instruction %d" % i),
|
|
r.get("pattern", r"\bseed\b"),
|
|
r.get("count", 1),
|
|
r.get("threshold", 3),
|
|
r.get("status", "observing"),
|
|
r.get("promoted_at"),
|
|
r.get("created_at", FROZEN_NOW),
|
|
r.get("last_seen", FROZEN_NOW),
|
|
),
|
|
)
|
|
db.commit()
|
|
rows = [dict(r) for r in db.execute(
|
|
"SELECT * FROM instructions ORDER BY id").fetchall()]
|
|
db.close()
|
|
return rows
|
|
|
|
|
|
def read_rows():
|
|
db = sqlite3.connect(oracle_mod.DB_PATH)
|
|
db.row_factory = sqlite3.Row
|
|
rows = [dict(r) for r in db.execute(
|
|
"SELECT * FROM instructions ORDER BY id").fetchall()]
|
|
db.close()
|
|
return rows
|
|
|
|
|
|
def kernel_input(rows, text):
|
|
return {"instructions": rows, "text": text, "now": FROZEN_NOW}
|
|
|
|
|
|
def expected_record(rows_after, result):
|
|
"""Derive the kernel's expected doc purely from oracle behavior."""
|
|
if result["matched"]:
|
|
row = next(r for r in rows_after if r["id"] == result["id"])
|
|
return {"matched": True, "id": row["id"], "count": row["count"],
|
|
"status": row["status"], "promoted": result["promoted"],
|
|
"last_seen": row["last_seen"]}
|
|
row = next(r for r in rows_after if r["id"] == result["id"])
|
|
return {"matched": False,
|
|
"new_row": {k: row[k] for k in
|
|
("instruction", "pattern", "count", "threshold",
|
|
"status", "created_at", "last_seen")}}
|
|
|
|
|
|
# ── record_instruction ───────────────────────────────────────────────────────
|
|
|
|
def r_new_keywords(rec):
|
|
rows = seed_rows([])
|
|
text = "Always CHECK the benchmark before building"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-keywords-pattern",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_new_no_keywords(rec):
|
|
rows = seed_rows([])
|
|
text = "do it"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-no-keywords-fallback-escape",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_new_fallback_truncate50(rec):
|
|
rows = seed_rows([])
|
|
text = " AB CD EF GH IJ KL MN OP QR ST UV WX YZ 12 34 56 "
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-fallback-truncated-50-stripped",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_new_regex_special(rec):
|
|
rows = seed_rows([])
|
|
text = "fix the bug (now)!"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-regex-special-chars",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_new_unicode(rec):
|
|
rows = seed_rows([])
|
|
text = "café über always"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-unicode-keywords",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_new_underscore_digit(rec):
|
|
rows = seed_rows([])
|
|
text = "v1_2 test_it now"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-underscore-digit-keywords",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_new_instruction_verbatim(rec):
|
|
# the INSERT stores the ORIGINAL text, not the lowered/stripped form
|
|
rows = seed_rows([])
|
|
text = " ALWAYS check "
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-instruction-verbatim-not-stripped",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_observing(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 1, "threshold": 3,
|
|
"status": "observing"},
|
|
])
|
|
text = "please CHECK everything"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("match-observing-increment",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_promote(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 2, "threshold": 3,
|
|
"status": "observing"},
|
|
])
|
|
text = "check again"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("match-promotes-at-threshold",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_already_active(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 5, "threshold": 3,
|
|
"status": "active", "promoted_at": FROZEN_NOW},
|
|
])
|
|
text = "check once more"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("match-already-active-no-promote",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_first_wins(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 1, "threshold": 3},
|
|
{"pattern": r"\bcheck|everything\b", "count": 1, "threshold": 3},
|
|
])
|
|
text = "check everything"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("match-first-row-wins",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_skips_retired(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 1, "threshold": 3,
|
|
"status": "retired"},
|
|
{"pattern": r"\bbenchmark\b", "count": 1, "threshold": 3},
|
|
])
|
|
text = "check the benchmark"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("match-skips-retired-row",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_word_boundary(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 1, "threshold": 3},
|
|
])
|
|
text = "the checks all pass"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("no-match-word-boundary-plural",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_fallback_pattern(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"do\ it", "count": 1, "threshold": 3},
|
|
])
|
|
text = "Do IT now"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("match-fallback-escaped-space-pattern",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_match_digit_threshold(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bship\b", "count": 3, "threshold": 4},
|
|
])
|
|
text = "ship it"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("match-count-below-higher-threshold",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
def r_new_after_nonmatch(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bdeploy\b", "count": 1, "threshold": 3},
|
|
])
|
|
text = "release the kraken"
|
|
result = oracle_mod.record_instruction(text)
|
|
after = read_rows()
|
|
rec("new-row-after-existing-nonmatch",
|
|
kernel_input(rows, text), result, after,
|
|
expected_record(after, result))
|
|
|
|
|
|
# ── check_context ────────────────────────────────────────────────────────────
|
|
|
|
def c_no_active(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 1, "threshold": 3},
|
|
{"pattern": r"\bship\b", "count": 3, "threshold": 3,
|
|
"status": "retired"},
|
|
])
|
|
context = "check the ship"
|
|
matches = oracle_mod.check_context(context)
|
|
rec("ctx-no-active-gates",
|
|
{"instructions": rows, "context": context},
|
|
matches, None, {"matches": matches})
|
|
|
|
|
|
def c_one_match(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 3, "threshold": 3,
|
|
"status": "active", "promoted_at": FROZEN_NOW,
|
|
"instruction": "always check"},
|
|
])
|
|
context = "about to check the benchmark"
|
|
matches = oracle_mod.check_context(context)
|
|
rec("ctx-one-match",
|
|
{"instructions": rows, "context": context},
|
|
matches, None, {"matches": matches})
|
|
|
|
|
|
def c_multi_match_order(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 3, "threshold": 3,
|
|
"status": "active", "instruction": "always check"},
|
|
{"pattern": r"\bship\b", "count": 7, "threshold": 3,
|
|
"status": "active", "instruction": "ship weekly"},
|
|
{"pattern": r"\bbenchmark\b", "count": 4, "threshold": 3,
|
|
"status": "active", "instruction": "run the benchmark"},
|
|
])
|
|
context = "check, ship, benchmark"
|
|
matches = oracle_mod.check_context(context)
|
|
rec("ctx-multi-match-rowid-order",
|
|
{"instructions": rows, "context": context},
|
|
matches, None, {"matches": matches})
|
|
|
|
|
|
def c_case_insensitive(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 3, "threshold": 3,
|
|
"status": "active", "instruction": "always check"},
|
|
])
|
|
context = "CHECK EVERYTHING NOW"
|
|
matches = oracle_mod.check_context(context)
|
|
rec("ctx-case-insensitive-context",
|
|
{"instructions": rows, "context": context},
|
|
matches, None, {"matches": matches})
|
|
|
|
|
|
def c_no_match(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 3, "threshold": 3,
|
|
"status": "active", "instruction": "always check"},
|
|
])
|
|
context = "deploy to production"
|
|
matches = oracle_mod.check_context(context)
|
|
rec("ctx-no-match",
|
|
{"instructions": rows, "context": context},
|
|
matches, None, {"matches": matches})
|
|
|
|
|
|
def c_word_boundary(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\bcheck\b", "count": 3, "threshold": 3,
|
|
"status": "active", "instruction": "always check"},
|
|
])
|
|
context = "the checks pass"
|
|
matches = oracle_mod.check_context(context)
|
|
rec("ctx-word-boundary-plural-no-match",
|
|
{"instructions": rows, "context": context},
|
|
matches, None, {"matches": matches})
|
|
|
|
|
|
# ── list_gates ───────────────────────────────────────────────────────────────
|
|
|
|
def l_empty(rec):
|
|
rows = seed_rows([])
|
|
gates = oracle_mod.list_gates()
|
|
rec("list-empty", {"instructions": rows}, gates, read_rows(), gates)
|
|
|
|
|
|
def l_order_desc(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\ba\b", "count": 5, "instruction": "five"},
|
|
{"pattern": r"\bb\b", "count": 1, "instruction": "one"},
|
|
{"pattern": r"\bc\b", "count": 3, "instruction": "three"},
|
|
])
|
|
gates = oracle_mod.list_gates()
|
|
rec("list-ordered-count-desc", {"instructions": rows}, gates, read_rows(), gates)
|
|
|
|
|
|
def l_ties(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\ba\b", "count": 2, "instruction": "first"},
|
|
{"pattern": r"\bb\b", "count": 4, "instruction": "second"},
|
|
{"pattern": r"\bc\b", "count": 2, "instruction": "third"},
|
|
])
|
|
gates = oracle_mod.list_gates()
|
|
rec("list-tie-order", {"instructions": rows}, gates, read_rows(), gates)
|
|
|
|
|
|
def l_null_promoted_at(rec):
|
|
rows = seed_rows([
|
|
{"pattern": r"\ba\b", "count": 2, "status": "active",
|
|
"promoted_at": FROZEN_NOW, "instruction": "active one"},
|
|
{"pattern": r"\bb\b", "count": 1, "instruction": "observing one"},
|
|
])
|
|
gates = oracle_mod.list_gates()
|
|
rec("list-null-and-set-promoted-at", {"instructions": rows}, gates,
|
|
read_rows(), gates)
|
|
|
|
|
|
# ── promote_gate ─────────────────────────────────────────────────────────────
|
|
|
|
def p_first(rec):
|
|
seed_rows([
|
|
{"pattern": r"\ba\b", "count": 1, "instruction": "observing"},
|
|
])
|
|
ret = oracle_mod.promote_gate(1)
|
|
rec("promote-existing-id",
|
|
{"id": 1, "now": FROZEN_NOW}, ret, read_rows(),
|
|
{"status": "active", "promoted_at": FROZEN_NOW})
|
|
|
|
|
|
def p_unknown_id(rec):
|
|
seed_rows([])
|
|
ret = oracle_mod.promote_gate(42)
|
|
rec("promote-unknown-id",
|
|
{"id": 42, "now": FROZEN_NOW}, ret, read_rows(),
|
|
{"status": "active", "promoted_at": FROZEN_NOW})
|
|
|
|
|
|
def p_overrides_retired(rec):
|
|
seed_rows([
|
|
{"pattern": r"\ba\b", "count": 1, "status": "retired",
|
|
"instruction": "retired one"},
|
|
])
|
|
ret = oracle_mod.promote_gate(1)
|
|
rec("promote-retired-row-becomes-active",
|
|
{"id": 1, "now": FROZEN_NOW}, ret, read_rows(),
|
|
{"status": "active", "promoted_at": FROZEN_NOW})
|
|
|
|
|
|
# ── handle_hook ──────────────────────────────────────────────────────────────
|
|
|
|
def h_default(rec):
|
|
ret = oracle_mod.handle_hook("post-build")
|
|
rec("hook-plain-event", {"event": "post-build"}, ret, None,
|
|
{"status": "ok", "event": "post-build", "handler": "task_gate"})
|
|
|
|
|
|
def h_with_data(rec):
|
|
ret = oracle_mod.handle_hook("pre-commit", {"sha": "abc"})
|
|
rec("hook-with-data", {"event": "pre-commit"}, ret, None,
|
|
{"status": "ok", "event": "pre-commit", "handler": "task_gate"})
|
|
|
|
|
|
def h_empty(rec):
|
|
ret = oracle_mod.handle_hook("")
|
|
rec("hook-empty-event", {"event": ""}, ret, None,
|
|
{"status": "ok", "event": "", "handler": "task_gate"})
|
|
|
|
|
|
# ── now / probe / main (literal-100% residue groups) ─────────────────────────
|
|
|
|
_EPOCH = datetime(1970, 1, 1)
|
|
|
|
|
|
def _micros(dt):
|
|
return (dt - _EPOCH) // timedelta(microseconds=1)
|
|
|
|
|
|
NOW_STAMPS = [
|
|
("epoch-zero", _micros(datetime(1970, 1, 1))),
|
|
("frozen-now-micros", _micros(datetime(2026, 10, 11, 23, 0, 0, 123456))),
|
|
("whole-second-no-frac", _micros(datetime(2026, 10, 11, 23, 0, 0))),
|
|
("leap-day-2024", _micros(datetime(2024, 2, 29, 12, 34, 56, 500000))),
|
|
("pre-epoch-negative", -1),
|
|
("pre-epoch-day-floor", _micros(datetime(1969, 12, 30, 23, 59, 59, 999999))),
|
|
]
|
|
|
|
|
|
def build_now_cases(rec):
|
|
from datetime import datetime as _dt
|
|
|
|
for name, micros in NOW_STAMPS:
|
|
class _F(_dt):
|
|
@classmethod
|
|
def now(cls, tz=None):
|
|
base = _dt(1970, 1, 1) + timedelta(microseconds=micros)
|
|
if tz is not None:
|
|
return base.replace(tzinfo=tz)
|
|
return base
|
|
|
|
old = oracle_mod.datetime
|
|
oracle_mod.datetime = _F
|
|
try:
|
|
got = oracle_mod._now()
|
|
finally:
|
|
oracle_mod.datetime = old
|
|
rec(f"now-{name}", {"micros": micros}, got, None, {"now": got})
|
|
|
|
|
|
def build_probe_cases(rec):
|
|
for name, pin in [("pin-python", "python"), ("pin-auto", "auto"),
|
|
("pin-other-resolves", "rust"),
|
|
("pin-unset-defaults-auto", None)]:
|
|
old = os.environ.get("RUNE_TG_CORE")
|
|
if pin is None:
|
|
os.environ.pop("RUNE_TG_CORE", None)
|
|
else:
|
|
os.environ["RUNE_TG_CORE"] = pin
|
|
try:
|
|
resolved = oracle_mod._tg_core() is not None
|
|
finally:
|
|
if old is None:
|
|
os.environ.pop("RUNE_TG_CORE", None)
|
|
else:
|
|
os.environ["RUNE_TG_CORE"] = old
|
|
rec(f"probe-{name}", {"pin": pin}, resolved, None,
|
|
{"resolved": resolved})
|
|
|
|
|
|
MAIN_CASES = [
|
|
# (name, pre_rows, argv)
|
|
("record-new-observing", [], ["record", "always check the benchmark"]),
|
|
("record-match-noted", [{"pattern": r"\bcheck\b", "count": 1,
|
|
"threshold": 3, "status": "observing"}],
|
|
["record", "please check it"]),
|
|
("record-promoted", [{"pattern": r"\bcheck\b", "count": 2,
|
|
"threshold": 3, "status": "observing"}],
|
|
["record", "check again"]),
|
|
("record-json-new", [], ["record", "--json", "ship it now"]),
|
|
("record-json-promoted", [{"pattern": r"\bship\b", "count": 2,
|
|
"threshold": 3, "status": "observing"}],
|
|
["record", "--json", "ship it"]),
|
|
("record-missing-arg", [], ["record"]),
|
|
("check-matches", [{"pattern": r"\bcheck\b", "count": 4,
|
|
"status": "active", "promoted_at": FROZEN_NOW}],
|
|
["check", "always CHECK the build"]),
|
|
("check-no-match", [{"pattern": r"\bcheck\b", "count": 4,
|
|
"status": "active", "promoted_at": FROZEN_NOW}],
|
|
["check", "unrelated words"]),
|
|
("check-observing-not-listed", [{"pattern": r"\bcheck\b", "count": 1,
|
|
"status": "observing"}],
|
|
["check", "check this"]),
|
|
("check-missing-arg", [], ["check"]),
|
|
("check-json", [{"pattern": r"\bcheck\b", "count": 4,
|
|
"status": "active", "promoted_at": FROZEN_NOW}],
|
|
["check", "--json", "check this"]),
|
|
("list-empty", [], ["list"]),
|
|
("list-rows", [{"pattern": r"\ba\b", "count": 5, "instruction": "five"},
|
|
{"pattern": r"\bb\b", "count": 1, "instruction": "one"},
|
|
{"pattern": r"\bc\b", "count": 3, "instruction": "three",
|
|
"status": "active", "promoted_at": FROZEN_NOW}],
|
|
["list"]),
|
|
("list-json", [{"pattern": r"\ba\b", "count": 5, "instruction": "five"},
|
|
{"pattern": r"\bb\b", "count": 1, "instruction": "one"}],
|
|
["list", "--json"]),
|
|
("promote", [{"pattern": r"\ba\b", "count": 1, "instruction": "obs"}],
|
|
["promote", "1"]),
|
|
("promote-unknown-id", [], ["promote", "42"]),
|
|
("promote-missing-arg", [], ["promote"]),
|
|
("hook-default", [], ["hook"]),
|
|
("hook-event-flag", [], ["hook", "--event", "on_session_end"]),
|
|
("hook-data-json", [], ["hook", "--data", '{"k": 1}']),
|
|
("missing-cmd-exit-2", [], []),
|
|
("invalid-choice-exit-2", [], ["bogus"]),
|
|
("unknown-flag-required-cmd-exit-2", [], ["--zz"]),
|
|
("unknown-flag-after-cmd-exit-2", [], ["record", "--zz", "hello"]),
|
|
("extra-positional-exit-2", [], ["record", "a", "b"]),
|
|
]
|
|
|
|
|
|
def build_main_cases(rec):
|
|
import contextlib
|
|
import io
|
|
|
|
for name, pre_rows, argv in MAIN_CASES:
|
|
rows = seed_rows(pre_rows)
|
|
old_argv = sys.argv
|
|
sys.argv = ["task_gate.py"] + list(argv)
|
|
buf_out, buf_err = io.StringIO(), io.StringIO()
|
|
code = 0
|
|
try:
|
|
with contextlib.redirect_stdout(buf_out), contextlib.redirect_stderr(buf_err):
|
|
try:
|
|
oracle_mod.main()
|
|
except SystemExit as e:
|
|
code = e.code if isinstance(e.code, int) else 0
|
|
finally:
|
|
sys.argv = old_argv
|
|
after = read_rows()
|
|
rec(f"main-{name}",
|
|
{"argv": argv, "instructions": rows, "now": FROZEN_NOW},
|
|
None, after,
|
|
{"exit": code, "stdout": buf_out.getvalue(),
|
|
"stderr": buf_err.getvalue()})
|
|
|
|
|
|
GROUPS = [
|
|
("record_instruction", [r_new_keywords, r_new_no_keywords,
|
|
r_new_fallback_truncate50, r_new_regex_special,
|
|
r_new_unicode, r_new_underscore_digit,
|
|
r_new_instruction_verbatim, r_match_observing,
|
|
r_match_promote, r_match_already_active,
|
|
r_match_first_wins, r_match_skips_retired,
|
|
r_match_word_boundary, r_match_fallback_pattern,
|
|
r_match_digit_threshold, r_new_after_nonmatch]),
|
|
("check_context", [c_no_active, c_one_match, c_multi_match_order,
|
|
c_case_insensitive, c_no_match, c_word_boundary]),
|
|
("list_gates", [l_empty, l_order_desc, l_ties, l_null_promoted_at]),
|
|
("promote_gate", [p_first, p_unknown_id, p_overrides_retired]),
|
|
("handle_hook", [h_default, h_with_data, h_empty]),
|
|
("now", [build_now_cases]),
|
|
("probe", [build_probe_cases]),
|
|
("main", [build_main_cases]),
|
|
]
|
|
|
|
# groups where the kernel is involved (list_gates' ordering is sqlite's)
|
|
KERNEL_GROUPS = {"record_instruction", "check_context", "promote_gate",
|
|
"handle_hook"}
|
|
|
|
|
|
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, kinput, oracle_ret, rows_after, kexpected, _r=recs):
|
|
entry = {"name": name, "input": kinput, "return": oracle_ret,
|
|
"expected": kexpected}
|
|
if rows_after is not None:
|
|
entry["rows_after"] = rows_after
|
|
if REPLAY:
|
|
old = next(c for c in frozen[group] if c["name"] == name)
|
|
for key in ("input", "return", "expected", "rows_after"):
|
|
if key not in entry and key not in old:
|
|
continue
|
|
got, want = entry.get(key), old.get(key)
|
|
if json.dumps(got, sort_keys=True) != json.dumps(
|
|
want, sort_keys=True):
|
|
print(f"FACADE DIVERGE [{group}] {name} {key}")
|
|
print(" got: ", json.dumps(got, sort_keys=True)[:600])
|
|
print(" want:", json.dumps(want, sort_keys=True)[: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
|
|
for group, cases in frozen.items():
|
|
if group not in KERNEL_GROUPS:
|
|
continue
|
|
for case in cases:
|
|
inp = case["input"]
|
|
if group == "record_instruction":
|
|
got = tg_loader.record_instruction(
|
|
inp["instructions"], inp["text"], inp["now"])
|
|
elif group == "check_context":
|
|
got = tg_loader.check_context(
|
|
inp["instructions"], inp["context"])
|
|
elif group == "promote_gate":
|
|
got = tg_loader.promote_gate(inp["id"], inp["now"])
|
|
else:
|
|
got = tg_loader.handle_hook(inp["event"])
|
|
if json.dumps(got, sort_keys=True) != json.dumps(
|
|
case["expected"], sort_keys=True):
|
|
failures += 1
|
|
print(f"KERNEL DIVERGE [{group}] {case['name']}")
|
|
print(" got: ", json.dumps(got, sort_keys=True)[:600])
|
|
print(" want:", json.dumps(
|
|
case["expected"], sort_keys=True)[:600])
|
|
print(f"replay: facade {total - facade_failures}/{total}, "
|
|
f"kernel {total - failures}/{total}")
|
|
return 1 if (failures or facade_failures) else 0
|
|
|
|
doc = {
|
|
# deterministic stamp (regen x2 must be byte-identical; a wall
|
|
# clock here would break the freeze protocol)
|
|
"frozen": FROZEN_NOW,
|
|
"oracle": "src/rune/task_gate.py",
|
|
"pin": "RUNE_TG_CORE=python",
|
|
"note": "Frozen via scripts/gen_task_gate_truth.py against the live "
|
|
"Python oracle; wall clock frozen at " + FROZEN_NOW + ". "
|
|
"Battery embeds this VERBATIM. Row dicts keep sqlite "
|
|
"column order (main --json re-renders it).",
|
|
"cases": groups,
|
|
}
|
|
os.makedirs(os.path.dirname(TRUTH_PATH), exist_ok=True)
|
|
with open(TRUTH_PATH, "w") as f:
|
|
# insertion order preserved: row-doc key order IS the truth the
|
|
# Rust --json renderer replays (sort_keys would lie about it)
|
|
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())
|