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

scripts/check_task_gate_parity.py

@@ -0,0 +1,259 @@
+#!/usr/bin/env python3
+"""Live parity: Rust task_gate kernel vs the Python oracle (dichotomy
+wave #2479). Replays the frozen truth corpus PLUS seeded fuzz over every
+public function and branch. Timestamps are frozen (no-clock law — the
+kernel takes `now` as an argument), so nothing is masked.
+
+The facade reads RUNE_TG_CORE per call, so one process can flip between
+the pinned oracle (RUNE_TG_CORE=python) and the kernel (unset).
+"""
+from __future__ import annotations
+
+import json
+import os
+import random
+import sqlite3
+import sys
+import tempfile
+from datetime import datetime, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src" / "rune"))
+
+import task_gate as tg  # noqa: E402
+
+FROZEN_NOW = "2026-10-11T23:00:00.123456+00:00"
+
+
+class FrozenDatetime(datetime):
+    """No-clock law: the facade's `now` stamps are wall clock, not logic —
+    freeze them so the two core runs stamp identically."""
+    @classmethod
+    def now(cls, tz=None):
+        return datetime(2026, 10, 11, 23, 0, 0, 123456, tzinfo=timezone.utc)
+
+
+tg.datetime = FrozenDatetime
+
+TRUTH_PATH = Path(__file__).resolve().parents[1] / (
+    "src/rust/task_gate/src/frozen_tg_truth.json")
+
+
+def call_core(core, fn, *args, **kwargs):
+    old = os.environ.get("RUNE_TG_CORE")
+    os.environ["RUNE_TG_CORE"] = core
+    try:
+        return getattr(tg, fn)(*args, **kwargs)
+    finally:
+        if old is None:
+            os.environ.pop("RUNE_TG_CORE", None)
+        else:
+            os.environ["RUNE_TG_CORE"] = old
+
+
+def fresh_db():
+    tmp = tempfile.mkdtemp(prefix="tg_parity_")
+    tg.DB_PATH = os.path.join(tmp, "task_gate.db")
+    db = tg.connect()
+    db.close()
+
+
+def read_rows():
+    db = sqlite3.connect(tg.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 seed_rows(rows):
+    fresh_db()
+    db = tg.connect()
+    for i, r in enumerate(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 %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()
+    db.close()
+
+
+def scenario_record(rows_before, text):
+    """Run record under both cores on identical pre-state; compare
+    (return, rows_after)."""
+    seed_rows(rows_before)
+    py_ret = call_core("python", "record_instruction", text)
+    py_rows = read_rows()
+    seed_rows(rows_before)
+    rs_ret = call_core("auto", "record_instruction", text)
+    rs_rows = read_rows()
+    ok = (json.dumps(py_ret, sort_keys=True)
+          == json.dumps(rs_ret, sort_keys=True)
+          and json.dumps(py_rows, sort_keys=True)
+          == json.dumps(rs_rows, sort_keys=True))
+    return ok, py_ret, rs_ret, py_rows, rs_rows
+
+
+def scenario_ctx(rows, context):
+    seed_rows(rows)
+    py = call_core("python", "check_context", context)
+    rs = call_core("auto", "check_context", context)
+    return (json.dumps(py, sort_keys=True)
+            == json.dumps(rs, sort_keys=True)), py, rs
+
+
+def scenario_list(rows):
+    seed_rows(rows)
+    py = call_core("python", "list_gates")
+    rs = call_core("auto", "list_gates")
+    return (json.dumps(py, sort_keys=True)
+            == json.dumps(rs, sort_keys=True)), py, rs
+
+
+def scenario_promote(rows, gate_id):
+    seed_rows(rows)
+    py = call_core("python", "promote_gate", gate_id)
+    py_rows = read_rows()
+    seed_rows(rows)
+    rs = call_core("auto", "promote_gate", gate_id)
+    rs_rows = read_rows()
+    return (json.dumps(py, sort_keys=True)
+            == json.dumps(rs, sort_keys=True)
+            and json.dumps(py_rows, sort_keys=True)
+            == json.dumps(rs_rows, sort_keys=True)), py, rs, py_rows, rs_rows
+
+
+def scenario_hook(event):
+    py = call_core("python", "handle_hook", event, {"ignored": True})
+    rs = call_core("auto", "handle_hook", event, {"ignored": True})
+    return (json.dumps(py, sort_keys=True)
+            == json.dumps(rs, sort_keys=True)), py, rs
+
+
+# ── fuzz pools ─────────────────────────────────────────────────────────
+WORDS = ["always", "check", "the", "benchmark", "before", "building",
+         "ship", "it", "do", "stop", "now", "café", "über", "v1_2",
+         "test_it", "fix", "bug", "release", "deploy", "kraken", "ab",
+         "cd", "research", "first", "wash", "spin", "gate", "task"]
+PATTERNS = [r"\bcheck\b", r"\bship\b", r"\bbenchmark\b", r"do\ it",
+            r"\balways\b|\bcheck\b", r"\bcafé\b", r"\bresearch\b",
+            r"\bgate\b|\btask\b", r"\bfirst\b"]
+STATUSES = ["observing", "active", "retired"]
+
+
+def rand_text(rng):
+    n = rng.randint(1, 8)
+    sep = rng.choice([" ", " ", " ", "  ", "! ", " ("])
+    return sep.join(rng.choice(WORDS) for _ in range(n)) + rng.choice(
+        ["", "!", "?", " (now)"])
+
+
+def rand_rows(rng, max_rows=4):
+    rows = []
+    for _ in range(rng.randint(0, max_rows)):
+        rows.append({
+            "pattern": rng.choice(PATTERNS),
+            "count": rng.randint(1, 5),
+            "threshold": rng.randint(2, 4),
+            "status": rng.choice(STATUSES),
+            "instruction": rand_text(rng),
+            "promoted_at": FROZEN_NOW if rng.random() < 0.3 else None,
+        })
+    return rows
+
+
+def fuzz_cases(rng):
+    cases = []
+    for _ in range(300):
+        cases.append(("record", rand_rows(rng), rand_text(rng)))
+    for _ in range(250):
+        cases.append(("ctx", rand_rows(rng), rand_text(rng)))
+    for _ in range(200):
+        cases.append(("list", rand_rows(rng, 5), None))
+    for _ in range(150):
+        cases.append(("promote", rand_rows(rng), rng.randint(1, 5)))
+    for _ in range(50):
+        cases.append(("hook", None, rand_text(rng)))
+    return cases
+
+
+def main() -> int:
+    failures = []
+
+    def check(label, ok, detail=""):
+        if not ok:
+            failures.append((label, detail))
+            print(f"FAIL {label}\n{detail}")
+
+    # 1. frozen truth corpus (facade level, both cores already proven by
+    #    gen_task_gate_truth.py --replay; re-run kernel level here live)
+    truth = json.loads(TRUTH_PATH.read_text())["cases"]
+    n_truth = 0
+    for group, cases in truth.items():
+        for case in cases:
+            if group == "record_instruction":
+                inp = case["input"]
+                ok, p, r, pr, rr = scenario_record(
+                    inp["instructions"], inp["text"])
+            elif group == "check_context":
+                inp = case["input"]
+                ok, p, r = scenario_ctx(inp["instructions"], inp["context"])
+            elif group == "list_gates":
+                ok, p, r = scenario_list(
+                    case["rows_after"] and _rows_before(case))
+            elif group == "promote_gate":
+                inp = case["input"]
+                ok, p, r, pr, rr = scenario_promote(
+                    [{"pattern": r"\bx\b"}], inp["id"])
+            else:  # handle_hook
+                ok, p, r = scenario_hook(case["input"]["event"])
+            n_truth += 1
+            check(f"truth:{group}:{case['name']}", ok,
+                  f"py={json.dumps(p, sort_keys=True)[:300]}\n"
+                  f"rs={json.dumps(r, sort_keys=True)[:300]}")
+
+    # 2. seeded fuzz
+    rng = random.Random(2479)
+    n_fuzz = 0
+    for kind, rows, arg in fuzz_cases(rng):
+        n_fuzz += 1
+        label = f"fuzz:{kind}:{n_fuzz}"
+        if kind == "record":
+            ok, p, r, pr, rr = scenario_record(rows, arg)
+        elif kind == "ctx":
+            ok, p, r = scenario_ctx(rows, arg)
+        elif kind == "list":
+            ok, p, r = scenario_list(rows)
+        elif kind == "promote":
+            ok, p, r, pr, rr = scenario_promote(rows, arg)
+        else:
+            ok, p, r = scenario_hook(arg)
+        check(label, ok,
+              f"py={json.dumps(p, sort_keys=True)[:300]}\n"
+              f"rs={json.dumps(r, sort_keys=True)[:300]}\n"
+              f"rows={json.dumps(rows, sort_keys=True)[:200]}")
+
+    total = n_truth + n_fuzz
+    print(f"parity: {total - len(failures)}/{total} "
+          f"({n_truth} truth + {n_fuzz} fuzz seed 2479)")
+    return 1 if failures else 0
+
+
+def _rows_before(case):
+    """list_gates truth carries rows_after (the full pre-state — listing
+    never mutates); replay it as the seed state."""
+    return case["rows_after"]
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/gen_task_gate_truth.py

@@ -0,0 +1,589 @@
+#!/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())

src/rune/task_gate.py

@@ -31,6 +31,24 @@ import memory_env
 DB_PATH = os.path.join(memory_env.memory_dir(), "task_gate.db")
 
 
+def _tg_core():
+    """Dichotomy wave #2479: Rust core first (RUNE_TG_CORE=python pins the
+    Python fallback — parity harnesses). The kernel owns the verdict logic
+    (pattern extraction, gate matching, count/threshold/status transitions,
+    promotion, active-gate projection, hook echo); the sqlite IO and the
+    CLI stay here."""
+    if os.environ.get("RUNE_TG_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import tg_loader
+
+        if tg_loader.available():
+            return tg_loader
+    except ImportError:
+        pass
+    return None
+
+
 def _now():
     return datetime.now(timezone.utc).isoformat()
 
@@ -67,6 +85,36 @@ def record_instruction(text):
     """Record an instruction. If it matches an existing pattern, increment count."""
     db = connect()
     try:
+        core = _tg_core()
+        if core is not None:
+            now = _now()
+            rows = [dict(r) for r in
+                    db.execute("SELECT * FROM instructions").fetchall()]
+            out = core.record_instruction(rows, text, now)
+            if out.get("matched"):
+                db.execute(
+                    "UPDATE instructions SET count=?, status=?, last_seen=? WHERE id=?",
+                    (out["count"], out["status"], out["last_seen"], out["id"]),
+                )
+                db.commit()
+                return {
+                    "matched": True,
+                    "id": out["id"],
+                    "count": out["count"],
+                    "status": out["status"],
+                    "promoted": out["promoted"],
+                }
+            row = out["new_row"]
+            cur = db.execute(
+                "INSERT INTO instructions (instruction, pattern, count, threshold, status, created_at, last_seen) "
+                "VALUES (?, ?, ?, ?, ?, ?, ?)",
+                (row["instruction"], row["pattern"], row["count"],
+                 row["threshold"], row["status"], row["created_at"],
+                 row["last_seen"]),
+            )
+            db.commit()
+            return {"matched": False, "id": cur.lastrowid, "count": 1, "status": "observing"}
+
         text_lower = text.lower().strip()
 
         # Check if this matches an existing instruction
@@ -112,6 +160,12 @@ def check_context(context):
     """Check if context matches any active gates."""
     db = connect()
     try:
+        core = _tg_core()
+        if core is not None:
+            rows = [dict(r) for r in
+                    db.execute("SELECT * FROM instructions").fetchall()]
+            return core.check_context(rows, context)["matches"]
+
         context_lower = context.lower()
 
         active = db.execute("SELECT * FROM instructions WHERE status='active'").fetchall()
@@ -146,6 +200,15 @@ def promote_gate(gate_id):
     """Manually promote a gate to active."""
     db = connect()
     try:
+        core = _tg_core()
+        if core is not None:
+            out = core.promote_gate(gate_id, _now())
+            db.execute(
+                "UPDATE instructions SET status='active', promoted_at=? WHERE id=?",
+                (out["promoted_at"], gate_id),
+            )
+            db.commit()
+            return {"promoted": gate_id}
         db.execute("UPDATE instructions SET status='active', promoted_at=? WHERE id=?", (_now(), gate_id))
         db.commit()
         return {"promoted": gate_id}
@@ -155,6 +218,9 @@ def promote_gate(gate_id):
 
 def handle_hook(event, data=None):
     """Hook handler — check for repeated instructions."""
+    core = _tg_core()
+    if core is not None:
+        return core.handle_hook(event)
     return {"status": "ok", "event": event, "handler": "task_gate"}
 
 

src/rune/tg_loader.py

@@ -0,0 +1,117 @@
+#!/usr/bin/env python3
+"""ctypes loader for the Rust task_gate core (dichotomy wave, #2479
+tranche). Mirrors broker_loader.py's contract:
+
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
+  * tg_record_instruction({instructions, text, now})
+      → {matched, id, count, status, promoted, last_seen | new_row}
+  * tg_check_context({instructions, context}) → {matches}
+  * tg_promote_gate({id, now}) → {status, promoted_at}
+  * tg_handle_hook({event}) → {status, event, handler}
+
+No-clock law: `now` (aware UTC ISO-8601) is always an argument — the
+kernel never reads the wall clock.
+
+Boundary: the sqlite IO (connect, the row reads, the UPDATE/INSERT) and
+the CLI main() stay in the Python facade (task_gate.py); the kernel owns
+the verdict logic (pattern extraction, gate matching, count/threshold/
+status transitions, promotion, active-gate projection, hook echo).
+
+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 TG_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" / "task_gate"
+_LIB_PATH = Path(
+    os.environ.get("TG_RUST_LIB", _CRATE_DIR / "target" / "release" / "libtask_gate.so")
+)
+
+_SYMBOLS = (
+    "tg_record_instruction",
+    "tg_check_context",
+    "tg_promote_gate",
+    "tg_handle_hook",
+)
+
+_lib = None
+
+
+class TaskGateCoreUnavailable(FileNotFoundError):
+    """The Rust task_gate 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 TaskGateCoreUnavailable(
+            f"Rust task_gate 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 record_instruction(instructions, text, now) -> dict:
+    return _call("tg_record_instruction", {
+        "instructions": instructions, "text": text, "now": now,
+    })
+
+
+def check_context(instructions, context) -> dict:
+    return _call("tg_check_context", {
+        "instructions": instructions, "context": context,
+    })
+
+
+def promote_gate(gate_id, now) -> dict:
+    return _call("tg_promote_gate", {"id": gate_id, "now": now})
+
+
+def handle_hook(event) -> dict:
+    return _call("tg_handle_hook", {"event": event})

src/rust/task_gate/src/branch_battery.rs

@@ -0,0 +1,68 @@
+//! task_gate branch battery — parity vs the frozen Python-oracle truth
+//! (frozen_tg_truth.json: 32 cases across all four kernel-owned public
+//! functions and every branch — keyword lattice with the 5-cap, len>3
+//! codepoint test, Python-3.7 re.escape fallback with the 50-codepoint
+//! truncation and escaped-space normalization, verbatim instruction
+//! storage, observing→active promotion at threshold, already-active and
+//! retired skips, first-row-wins, word boundaries, active-gate context
+//! projection with rowid order, retired-row promotion, hook echo).
+//! Frozen 2026-10-11 via scripts/gen_task_gate_truth.py with
+//! RUNE_TG_CORE=python pinned (poisoned-truth law). Embedded VERBATIM —
+//! never transcribed.
+
+use crate::{check_context, handle_hook, promote_gate, record_instruction};
+use serde_json::Value;
+
+const TRUTH: &str = include_str!("frozen_tg_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 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 record_instruction_parity_battery() {
+    replay("record_instruction", |i| {
+        record_instruction(
+            i["instructions"].as_array().unwrap(),
+            i["text"].as_str().unwrap(),
+            i["now"].as_str().unwrap(),
+        )
+    });
+}
+
+#[test]
+fn check_context_parity_battery() {
+    replay("check_context", |i| {
+        check_context(
+            i["instructions"].as_array().unwrap(),
+            i["context"].as_str().unwrap(),
+        )
+    });
+}
+
+#[test]
+fn promote_gate_parity_battery() {
+    replay("promote_gate", |i| promote_gate(i["now"].as_str().unwrap()));
+}
+
+#[test]
+fn handle_hook_parity_battery() {
+    replay("handle_hook", |i| handle_hook(i["event"].as_str().unwrap()));
+}

src/rust/task_gate/src/lib.rs

@@ -1,111 +1,334 @@
-//! task_gate — repeated task gate: repeated instructions become standards (W6).
-//! PARITY port of src/rune/task_gate.py record_instruction.
-use serde::{Deserialize, Serialize};
-
-#[derive(Debug, Clone, Serialize, Deserialize)]
-pub struct GateInstruction {
-    pub instruction: String,
-    pub pattern: String,
-    pub count: i64,
-    pub threshold: i64,
-    pub status: String,
-}
-
-pub fn extract_pattern(text: &str) -> String {
-    let lower = text.to_lowercase();
-    let words: Vec<&str> = lower
-        .split_whitespace()
-        .filter(|w| w.len() > 3)
-        .take(5)
-        .collect();
-    if words.is_empty() {
-        return regex::escape(&lower[..std::cmp::min(50, lower.len())]);
-    }
-    words
-        .iter()
-        .map(|w| format!(r"\b{}\b", regex::escape(w)))
-        .collect::<Vec<_>>()
-        .join("|")
-}
-
-/// Record an instruction. If it matches an existing gate, increment count.
-pub fn record_instruction(
-    instructions: &mut Vec<GateInstruction>,
-    text: &str,
-) -> serde_json::Value {
-    let lower = text.trim().to_lowercase();
-
-    for inst in instructions.iter_mut() {
-        if inst.status == "retired" {
-            continue;
-        }
-        if let Ok(re) = regex::Regex::new(&inst.pattern) {
-            if re.is_match(&lower) {
-                inst.count += 1;
-                let was_active = inst.status == "active";
-                inst.status = if inst.count >= inst.threshold {
-                    "active".to_string()
-                } else {
-                    "observing".to_string()
-                };
-                return serde_json::json!({
-                    "matched": true, "id": inst.count,
-                    "count": inst.count, "status": inst.status,
-                    "promoted": !was_active && inst.status == "active"
-                });
-            }
-        }
-    }
-
-    // No match — create new instruction
-    let pattern = extract_pattern(&lower);
-    let inst = GateInstruction {
-        instruction: text.to_string(),
-        pattern,
-        count: 1,
-        threshold: 3,
-        status: "observing".to_string(),
-    };
-    instructions.push(inst.clone());
-    serde_json::json!({"matched": false, "count": 1, "status": "observing"})
-}
+//! task_gate — repeated task gate: repeated instructions become standards
+//! (W6). FULL parity port of src/rune/task_gate.py (dichotomy wave, issue
+//! #2479 tranche).
+//!
+//! The kernel owns the verdict logic — pattern extraction (Python
+//! `\w+`/`len > 3`/first-5 lattice with the `re.escape` fallback),
+//! case-insensitive regex gate matching (first non-retired row wins),
+//! count/threshold/status transitions, the promotion decision, the
+//! active-gate context projection, the manual promotion payload and the
+//! hook echo. The sqlite IO (connect, row reads, UPDATE/INSERT) and the
+//! CLI stay in the Python facade (task_gate.py); the kernel is a pure
+//! function of the row docs the facade reads.
+//!
+//! Python semantics preserved verbatim (parity is pinned by the frozen
+//! oracle truth, frozen_tg_truth.json + the branch battery):
+//!   * `text.lower().strip()` feeds matching AND pattern extraction, but
+//!     the INSERTed instruction is the ORIGINAL text (parity case
+//!     new-instruction-verbatim-not-stripped).
+//!   * `len(w) > 3` is a CODEPOINT test, not bytes.
+//!   * the no-keyword fallback is Python 3.7+ `re.escape` of the first 50
+//!     CODEPOINTS of the stripped lower text (space IS escaped → `"\ "`,
+//!     which the matcher normalizes before handing to the regex crate).
+//!   * no-clock law: `now` is always an argument — the kernel never reads
+//!     the wall clock.
+//!
+//! FFI (doc-envelope law): every symbol takes one JSON doc and returns
+//! one JSON doc `{"ok": true, "value": ...}` / `{"ok": false, "err": ...}`.
+//! Strings are freed via capability_scoring's cs_free_string — this crate
+//! exports NO free symbol of its own (duplicate exports break the cdylib
+//! link when sibling crates share one process).
+use serde_json::{json, Value};
+use std::ffi::c_char;
+use std::fmt;
 
 #[cfg(test)]
-mod tests {
-    use super::*;
+mod branch_battery;
 
-    #[test]
-    fn test_first_record_creates_new() {
-        let mut instructions = vec![];
-        let r = record_instruction(&mut instructions, "always check the benchmark");
-        assert_eq!(r["matched"], false);
-        assert_eq!(instructions.len(), 1);
+// ── pattern extraction (name-preserving port of the oracle's inline logic) ──
+
+/// Python 3.7+ `re.escape`: only characters that can have special meaning
+/// in a regular expression are escaped. Space and the other whitespace
+/// characters ARE in that set (verbose-mode guards); letters/digits/underscore
+/// never are.
+fn py_re_escape(s: &str) -> String {
+    let special = |c: char| {
+        matches!(
+            c,
+            '(' | ')'
+                | '['
+                | ']'
+                | '{'
+                | '}'
+                | '?'
+                | '*'
+                | '+'
+                | '-'
+                | '|'
+                | '^'
+                | '$'
+                | '\\'
+                | '.'
+                | '&'
+                | '~'
+                | '#'
+                | ' '
+                | '\t'
+                | '\n'
+                | '\r'
+                | '\u{0b}'
+                | '\u{0c}'
+        )
+    };
+    let mut out = String::with_capacity(s.len() + 8);
+    for c in s.chars() {
+        if special(c) {
+            out.push('\\');
+        }
+        out.push(c);
     }
+    out
+}
 
-    #[test]
-    fn test_repeated_record_increments() {
-        let mut instructions = vec![];
-        record_instruction(&mut instructions, "always check the benchmark");
-        let r = record_instruction(
-            &mut instructions,
-            "always check the benchmark before building",
-        );
-        assert_eq!(r["matched"], true);
-        assert_eq!(r["count"], 2);
-    }
-
-    #[test]
-    fn test_promotion_at_threshold() {
-        let mut instructions = vec![];
-        let pattern = extract_pattern("always check the benchmark");
-        instructions.push(GateInstruction {
-            instruction: "always check the benchmark".into(),
-            pattern,
-            count: 2,
-            threshold: 3,
-            status: "observing".into(),
-        });
-        let r = record_instruction(&mut instructions, "always check the benchmark carefully");
-        assert_eq!(r["promoted"], true);
+/// First 50 CODEPOINTS of `s` (Python `s[:50]`), returned as a &str slice —
+/// byte-index slicing would split UTF-8.
+fn first_n_chars(s: &str, n: usize) -> &str {
+    match s.char_indices().nth(n) {
+        Some((i, _)) => &s[..i],
+        None => s,
     }
 }
+
+/// The gate pattern for a never-seen instruction: word-boundary alternation
+/// over the first five `\w+` keywords longer than three codepoints, falling
+/// back to the escaped 50-codepoint prefix of the (already lowered/stripped)
+/// text. Name preserved from the earlier parity stub.
+pub fn extract_pattern(lowered: &str) -> String {
+    let keywords: Vec<&str> = crate::word_tokens(lowered)
+        .filter(|w| w.chars().count() > 3)
+        .take(5)
+        .collect();
+    if keywords.is_empty() {
+        py_re_escape(first_n_chars(lowered, 50))
+    } else {
+        keywords
+            .iter()
+            .map(|w| format!(r"\b{}\b", py_re_escape(w)))
+            .collect::<Vec<_>>()
+            .join("|")
+    }
+}
+
+/// Python `re.findall(r"\w+", text)` — Unicode-aware word tokens, borrowed
+/// zero-copy from the input.
+fn word_tokens(text: &str) -> impl Iterator<Item = &str> {
+    text.split(|c: char| !c.is_alphanumeric() && c != '_')
+        .filter(|w| !w.is_empty())
+}
+
+// ── gate matching ────────────────────────────────────────────────────────────
+
+/// Python `re.search(pattern, text, re.IGNORECASE)`. Python's own re.escape
+/// escapes whitespace (`"\ "`), which the regex crate rejects — an escaped
+/// whitespace char denotes exactly that literal char, so unescape before
+/// compiling. A row whose pattern cannot compile matches nothing (the
+/// oracle's DB only ever holds oracle-generated patterns).
+fn py_regex_matches(pattern: &str, text: &str) -> bool {
+    let normalized = if pattern.contains("\\ ") || pattern.contains("\\\t") {
+        let mut s = String::with_capacity(pattern.len());
+        let mut chars = pattern.chars().peekable();
+        while let Some(c) = chars.next() {
+            if c == '\\' {
+                match chars.peek() {
+                    Some(' ') => {
+                        chars.next();
+                        s.push(' ');
+                        continue;
+                    }
+                    Some('\t') => {
+                        chars.next();
+                        s.push('\t');
+                        continue;
+                    }
+                    _ => {}
+                }
+            }
+            s.push(c);
+        }
+        s
+    } else {
+        pattern.to_string()
+    };
+    regex::RegexBuilder::new(&normalized)
+        .case_insensitive(true)
+        .build()
+        .map(|re| re.is_match(text))
+        .unwrap_or(false)
+}
+
+fn row_str<'a>(row: &'a Value, key: &str) -> &'a str {
+    row.get(key).and_then(Value::as_str).unwrap_or("")
+}
+
+fn row_i64(row: &Value, key: &str) -> i64 {
+    row.get(key).and_then(Value::as_i64).unwrap_or(0)
+}
+
+// ── public functions (name-preserving ports of the oracle) ──────────────────
+
+/// task_gate.record_instruction — match against every non-retired row in
+/// order; first match increments count and re-decides status; otherwise a
+/// new observing row (count 1, threshold 3) is produced for the facade to
+/// INSERT. Returns the kernel decision: the matched-branch update payload
+/// (including the `last_seen` stamp) or the full new-row doc.
+pub fn record_instruction(rows: &[Value], text: &str, now: &str) -> Value {
+    let lowered = text.to_lowercase();
+    let lowered = lowered.trim();
+
+    for row in rows {
+        if row_str(row, "status") == "retired" {
+            continue;
+        }
+        if py_regex_matches(row_str(row, "pattern"), lowered) {
+            let new_count = row_i64(row, "count") + 1;
+            let status = row_str(row, "status");
+            let new_status = if new_count >= row_i64(row, "threshold") {
+                "active"
+            } else {
+                "observing"
+            };
+            return json!({
+                "matched": true,
+                "id": row_i64(row, "id"),
+                "count": new_count,
+                "status": new_status,
+                "promoted": new_status == "active" && status != "active",
+                "last_seen": now,
+            });
+        }
+    }
+
+    json!({
+        "matched": false,
+        "new_row": {
+            "instruction": text,
+            "pattern": extract_pattern(lowered),
+            "count": 1,
+            "threshold": 3,
+            "status": "observing",
+            "created_at": now,
+            "last_seen": now,
+        },
+    })
+}
+
+/// task_gate.check_context — project the ACTIVE gates whose pattern matches
+/// the (lowered, NOT stripped) context, in row order. The active-status
+/// filter is part of the oracle's verdict (its SQL carries
+/// `WHERE status='active'`), so the kernel applies it to the full row set.
+pub fn check_context(rows: &[Value], context: &str) -> Value {
+    let lowered = context.to_lowercase();
+    let matches: Vec<Value> = rows
+        .iter()
+        .filter(|row| row_str(row, "status") == "active")
+        .filter(|row| py_regex_matches(row_str(row, "pattern"), &lowered))
+        .map(|row| {
+            json!({
+                "id": row_i64(row, "id"),
+                "instruction": row_str(row, "instruction"),
+                "count": row_i64(row, "count"),
+            })
+        })
+        .collect();
+    json!({ "matches": matches })
+}
+
+/// task_gate.promote_gate — the manual-promotion update payload; the facade
+/// applies the UPDATE and owns the `{"promoted": id}` echo.
+pub fn promote_gate(now: &str) -> Value {
+    json!({"status": "active", "promoted_at": now})
+}
+
+/// task_gate.handle_hook — the hook echo (data is accepted and ignored by
+/// the oracle, so the kernel never sees it).
+pub fn handle_hook(event: &str) -> Value {
+    json!({"status": "ok", "event": event, "handler": "task_gate"})
+}
+
+// ── FFI (doc-envelope law) ───────────────────────────────────────────────────
+
+/// Typed FFI failure lattice behind the doc-envelope Result — the
+/// caller-visible `err` string is unchanged (enum only shapes the Rust
+/// side; parity is pinned by the battery).
+enum FfiError {
+    NullPtr,
+    Utf8(std::str::Utf8Error),
+    Json(serde_json::Error),
+}
+
+impl fmt::Display for FfiError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            FfiError::NullPtr => write!(f, "null argument pointer"),
+            FfiError::Utf8(e) => write!(f, "{e}"),
+            FfiError::Json(e) => write!(f, "{e}"),
+        }
+    }
+}
+
+fn read_arg(ptr: *const c_char) -> Result<Value, FfiError> {
+    if ptr.is_null() {
+        return Err(FfiError::NullPtr);
+    }
+    let x = unsafe { std::ffi::CStr::from_ptr(ptr) }
+        .to_str()
+        .map_err(FfiError::Utf8)?;
+    serde_json::from_str(x).map_err(FfiError::Json)
+}
+
+fn envelope(r: Result<Value, FfiError>) -> *mut c_char {
+    let v = match r {
+        Ok(x) => json!({"ok": true, "value": x}),
+        Err(e) => json!({"ok": false, "err": e.to_string()}),
+    };
+    std::ffi::CString::new(serde_json::to_string(&v).unwrap_or_default())
+        .map(|c| c.into_raw())
+        .unwrap_or(std::ptr::null_mut())
+}
+
+fn rows_of(a: &Value) -> &[Value] {
+    a.get("instructions")
+        .and_then(Value::as_array)
+        .map(Vec::as_slice)
+        .unwrap_or(&[])
+}
+
+/// tg_record_instruction({instructions, text, now})
+///   → {matched, id/count/status/promoted/last_seen | new_row}
+#[no_mangle]
+pub extern "C" fn tg_record_instruction(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        record_instruction(
+            rows_of(&a),
+            a.get("text").and_then(Value::as_str).unwrap_or(""),
+            a.get("now").and_then(Value::as_str).unwrap_or(""),
+        )
+    }))
+}
+
+/// tg_check_context({instructions, context}) → {matches}
+#[no_mangle]
+pub extern "C" fn tg_check_context(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        check_context(
+            rows_of(&a),
+            a.get("context").and_then(Value::as_str).unwrap_or(""),
+        )
+    }))
+}
+
+/// tg_promote_gate({id, now}) → {status, promoted_at}
+#[no_mangle]
+pub extern "C" fn tg_promote_gate(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        promote_gate(a.get("now").and_then(Value::as_str).unwrap_or(""))
+    }))
+}
+
+/// tg_handle_hook({event}) → {status, event, handler}
+#[no_mangle]
+pub extern "C" fn tg_handle_hook(arg: *const c_char) -> *mut c_char {
+    envelope(read_arg(arg).map(|a| {
+        handle_hook(a.get("event").and_then(Value::as_str).unwrap_or(""))
+    }))
+}