rune/scripts/gen_task_gate_truth.py

#!/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, 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):
    seed_rows([])
    gates = oracle_mod.list_gates()
    rec("list-empty", {}, gates, read_rows(), None)


def l_order_desc(rec):
    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", {}, gates, read_rows(), None)


def l_ties(rec):
    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", {}, gates, read_rows(), None)


def l_null_promoted_at(rec):
    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", {}, gates, read_rows(), None)


# ── 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"})


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]),
]

# 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 = {
        "frozen": datetime.now(timezone.utc).isoformat(),
        "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.",
        "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, sort_keys=True)
        f.write("\n")
    print(f"frozen {total} cases → {os.path.relpath(TRUTH_PATH, REPO)}")
    return 0


if __name__ == "__main__":
    sys.exit(main())