feat(task_gate): literal-100% residue — dichotomy wave

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.

scripts/check_task_gate_parity.py

@@ -198,8 +198,15 @@ def main() -> int:
     # 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"]
+    # now/probe/main groups have no FFI surface (lib-level twins + the
+    # in-process CLI dispatcher) — they are battery-replayed from the
+    # frozen oracle truth; live parity here covers the FFI groups.
+    LIVE_PARITY_GROUPS = {"record_instruction", "check_context",
+                          "list_gates", "promote_gate", "handle_hook"}
     n_truth = 0
     for group, cases in truth.items():
+        if group not in LIVE_PARITY_GROUPS:
+            continue
         for case in cases:
             if group == "record_instruction":
                 inp = case["input"]

scripts/gen_task_gate_truth.py

@@ -28,7 +28,7 @@ import os
 import sqlite3
 import sys
 import tempfile
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 
 REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
 TRUTH_PATH = os.path.join(
@@ -395,39 +395,40 @@ def c_word_boundary(rec):
 # ── list_gates ───────────────────────────────────────────────────────────────
 
 def l_empty(rec):
-    seed_rows([])
+    rows = seed_rows([])
     gates = oracle_mod.list_gates()
-    rec("list-empty", {}, gates, read_rows(), None)
+    rec("list-empty", {"instructions": rows}, gates, read_rows(), gates)
 
 
 def l_order_desc(rec):
-    seed_rows([
+    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", {}, gates, read_rows(), None)
+    rec("list-ordered-count-desc", {"instructions": rows}, gates, read_rows(), gates)
 
 
 def l_ties(rec):
-    seed_rows([
+    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", {}, gates, read_rows(), None)
+    rec("list-tie-order", {"instructions": rows}, gates, read_rows(), gates)
 
 
 def l_null_promoted_at(rec):
-    seed_rows([
+    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", {}, gates, read_rows(), None)
+    rec("list-null-and-set-promoted-at", {"instructions": rows}, gates,
+        read_rows(), gates)
 
 
 # ── promote_gate ─────────────────────────────────────────────────────────────
@@ -481,6 +482,143 @@ def h_empty(rec):
         {"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,
@@ -495,6 +633,9 @@ GROUPS = [
     ("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)
@@ -569,17 +710,22 @@ def main():
         return 1 if (failures or facade_failures) else 0
 
     doc = {
-        "frozen": datetime.now(timezone.utc).isoformat(),
+        # 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.",
+                "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:
-        json.dump(doc, f, indent=1, sort_keys=True)
+        # 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

src/rust/task_gate/Cargo.lock

@@ -11,6 +11,28 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "equivalent"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "00d174d5400e5e8fd687ad1049e2f578285fa914201b1af7e8b112a4546bd826"
+
+[[package]]
+name = "hashbrown"
+version = "0.17.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+
+[[package]]
+name = "indexmap"
+version = "2.14.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
+dependencies = [
+ "equivalent",
+ "hashbrown",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.18"
@@ -106,6 +128,7 @@ version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
+ "indexmap",
  "itoa",
  "memchr",
  "serde",

src/rust/task_gate/Cargo.toml

@@ -8,5 +8,7 @@ crate-type = ["rlib", "cdylib", "staticlib"]
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+# preserve_order: row docs ride in sqlite column order (dict(r) law) and
+# main()'s --json mode re-renders them in that order (json.dumps parity).
+serde_json = { version = "1", features = ["preserve_order"] }
 regex = "1"

src/rust/task_gate/src/branch_battery.rs

@@ -10,7 +10,8 @@
 //! RUNE_TG_CORE=python pinned (poisoned-truth law). Embedded VERBATIM —
 //! never transcribed.
 
-use crate::{check_context, handle_hook, promote_gate, record_instruction};
+use crate::{check_context, handle_hook, list_gates, main, promote_gate,
+            record_instruction, _now, _tg_core};
 use serde_json::Value;
 
 const TRUTH: &str = include_str!("frozen_tg_truth.json");
@@ -66,3 +67,82 @@ fn promote_gate_parity_battery() {
 fn handle_hook_parity_battery() {
     replay("handle_hook", |i| handle_hook(i["event"].as_str().unwrap()));
 }
+
+#[test]
+fn list_gates_parity_battery() {
+    replay("list_gates", |i| {
+        Value::Array(list_gates(i["instructions"].as_array().unwrap()))
+    });
+}
+
+#[test]
+fn now_parity_battery() {
+    for case in cases("now") {
+        let micros = case["input"]["micros"].as_i64().unwrap();
+        let got = _now(micros).expect("now ok");
+        assert_eq!(
+            got,
+            case["expected"]["now"].as_str().unwrap(),
+            "[now] case '{}' diverged",
+            case["name"].as_str().unwrap()
+        );
+    }
+}
+
+#[test]
+fn probe_parity_battery() {
+    for case in cases("probe") {
+        let pin = case["input"]["pin"].as_str();
+        let got = _tg_core(pin);
+        assert_eq!(
+            got.is_some(),
+            case["expected"]["resolved"].as_bool().unwrap(),
+            "[probe] case '{}' diverged (pin={pin:?})",
+            case["name"].as_str().unwrap()
+        );
+    }
+}
+
+#[test]
+fn main_parity_battery() {
+    for case in cases("main") {
+        let argv: Vec<String> = std::iter::once("task_gate.py".to_string())
+            .chain(
+                case["input"]["argv"]
+                    .as_array()
+                    .into_iter()
+                    .flatten()
+                    .filter_map(|v| v.as_str().map(str::to_string)),
+            )
+            .collect();
+        let rows = case["input"]["instructions"].as_array().unwrap();
+        let now = case["input"]["now"].as_str().unwrap();
+        let got = main(&argv, rows, now);
+        let want_exit = case["expected"]["exit"].as_i64().unwrap();
+        let want_stdout = case["expected"]["stdout"].as_str().unwrap();
+        let want_stderr = case["expected"]["stderr"].as_str().unwrap();
+        assert_eq!(got.exit as i64, want_exit, "[main] '{}' exit", case["name"].as_str().unwrap());
+        assert_eq!(
+            got.stdout, want_stdout,
+            "[main] '{}' stdout\n got:{:?}\nwant:{:?}",
+            case["name"].as_str().unwrap(),
+            got.stdout,
+            want_stdout
+        );
+        assert_eq!(
+            got.stderr, want_stderr,
+            "[main] '{}' stderr\n got:{:?}\nwant:{:?}",
+            case["name"].as_str().unwrap(),
+            got.stderr,
+            want_stderr
+        );
+        if let Some(after) = case.get("rows_after") {
+            assert_eq!(
+                &Value::Array(got.rows_after.clone()),
+                after,
+                "[main] '{}' rows_after",
+                case["name"].as_str().unwrap()
+            );
+        }
+    }
+}

src/rust/task_gate/src/lib.rs

@@ -340,6 +340,542 @@ pub fn handle_hook(event: &str) -> Value {
     json!({"status": "ok", "event": event, "handler": "task_gate"})
 }
 
+// ── residue: _tg_core / _now / list_gates / main (literal-100% wave) ────────
+
+/// The resolved in-process task-gate core. Twin of the oracle's loader
+/// MODULE handle (`_tg_core` returns the tg_loader module, never the raw
+/// CDLL — TgCore is that handle's shape in the reunited crate).
+#[derive(Debug, Clone, Copy)]
+pub struct TgCore;
+
+/// The Rust task-gate core probe (name-preserving port of `_tg_core`).
+/// RUNE_TG_CORE=python pins the Python fallback → None; every other value
+/// resolves the core. The oracle's load-failure branch can't fire here —
+/// in-process is always loaded.
+pub fn _tg_core(pin: Option<&str>) -> Option<TgCore> {
+    match pin {
+        Some("python") => None,
+        _ => Some(TgCore),
+    }
+}
+
+/// `_now()` with the clock as an argument (no-wall-clock law): epoch
+/// microseconds → `datetime.now(timezone.utc).isoformat()` byte shape —
+/// 4-digit year, `.ffffff` omitted when microsecond == 0, `+00:00`.
+pub fn _now(micros: i64) -> Result<String, (String, String)> {
+    let secs = micros.div_euclid(1_000_000);
+    let sub = micros.rem_euclid(1_000_000) as u32;
+    let days = secs.div_euclid(86_400);
+    let sod = secs.rem_euclid(86_400);
+    let (year, month, day) = civil_from_days(days);
+    if !(1..=9999).contains(&year) {
+        return Err((
+            format!("year {year} is out of range"),
+            "ValueError".to_string(),
+        ));
+    }
+    let (hh, mm, ss) = (sod / 3_600, (sod % 3_600) / 60, sod % 60);
+    let frac = if sub == 0 {
+        String::new()
+    } else {
+        format!(".{sub:06}")
+    };
+    Ok(format!(
+        "{year:04}-{month:02}-{day:02}T{hh:02}:{mm:02}:{ss:02}{frac}+00:00"
+    ))
+}
+
+/// Howard Hinnant's civil_from_days — days since 1970-01-01 → (y, m, d).
+fn civil_from_days(z: i64) -> (i64, i64, i64) {
+    let z = z + 719_468;
+    let era = if z >= 0 { z } else { z - 146_096 } / 146_097;
+    let doe = z - era * 146_097;
+    let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
+    let y = yoe + era * 400;
+    let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
+    let mp = (5 * doy + 2) / 153;
+    let d = doy - (153 * mp + 2) / 5 + 1;
+    let m = if mp < 10 { mp + 3 } else { mp - 9 };
+    (if m <= 2 { y + 1 } else { y }, m, d)
+}
+
+/// `list_gates` — the listing projection: `SELECT * ORDER BY count DESC`.
+/// The sqlite read stays in the facade; the kernel owns the ordering law
+/// (count DESC, ties keep table order — sqlite's sorter emits id order,
+/// which a stable sort over id-ordered rows reproduces) and returns the
+/// rows as pass-through dicts.
+pub fn list_gates(rows: &[Value]) -> Vec<Value> {
+    let mut out = rows.to_vec();
+    out.sort_by(|a, b| {
+        let ca = a.get("count").and_then(Value::as_i64).unwrap_or(0);
+        let cb = b.get("count").and_then(Value::as_i64).unwrap_or(0);
+        cb.cmp(&ca)
+    });
+    out
+}
+
+// ── main: the CLI dispatcher (task_persist MainOutcome law) ────────────────
+
+/// The outcome of the oracle's main(): process exit code, captured
+/// stdout/stderr, and the instructions-table state after the dispatch
+/// (matched UPDATE / new INSERT / promote UPDATE applied in memory — the
+/// sqlite IO stays in the facade).
+#[derive(Debug, Clone, PartialEq)]
+pub struct TgMainOutcome {
+    pub exit: i32,
+    pub stdout: String,
+    pub stderr: String,
+    pub rows_after: Vec<Value>,
+}
+
+const TG_USAGE: &str = "\
+usage: task_gate.py [-h] [--event EVENT] [--data DATA] [--json]
+                    {record,check,list,promote,hook} [arg]
+";
+
+fn argparse_err(detail: &str) -> TgMainOutcome {
+    TgMainOutcome {
+        exit: 2,
+        stdout: String::new(),
+        stderr: format!("{TG_USAGE}task_gate.py: error: {detail}\n"),
+        rows_after: Vec::new(),
+    }
+}
+
+/// Python `s[:n]` on the instruction text (codepoint truncation).
+fn char_slice(s: &str, n: usize) -> String {
+    s.chars().take(n).collect()
+}
+
+/// f"{g['status']:10}" — left-aligned, width 10 (codepoints).
+fn status_field(status: &str) -> String {
+    let mut out = status.to_string();
+    while out.chars().count() < 10 {
+        out.push(' ');
+    }
+    out
+}
+
+/// f"{g['id']:3}" — right-aligned, width 3.
+fn id_field(id: i64) -> String {
+    let s = id.to_string();
+    if s.len() >= 3 {
+        s
+    } else {
+        format!("{}{}", " ".repeat(3 - s.len()), s)
+    }
+}
+
+/// The oracle's main(): argparse dispatch (choices/cmd/arg/--event/--data/
+/// --json, COLUMNS=80 usage wrap, exit-2 error classes with the
+/// required-cmd check firing BEFORE unrecognized-args) over the four
+/// verb bodies. `rows` is the instructions table (id order, dict(r)
+/// column shape) the facade read; `now` is the frozen `_now()` stamp.
+pub fn main(argv: &[String], rows: &[Value], now: &str) -> TgMainOutcome {
+    let mut cmd: Option<String> = None;
+    let mut arg: Option<String> = None;
+    let mut event: Option<String> = None;
+    let mut data: Option<String> = None;
+    let mut json_flag = false;
+    let mut unrecognized: Vec<String> = Vec::new();
+    let mut positional_only = false;
+
+    let args: Vec<String> = argv.iter().skip(1).cloned().collect();
+    let mut i = 0usize;
+    while i < args.len() {
+        let tok = &args[i];
+        if positional_only {
+            if cmd.is_none() {
+                cmd = Some(tok.clone());
+            } else if arg.is_none() {
+                arg = Some(tok.clone());
+            } else {
+                unrecognized.push(tok.clone());
+            }
+            i += 1;
+            continue;
+        }
+        match tok.as_str() {
+            "--" => positional_only = true,
+            "--event" | "--data" => {
+                if i + 1 < args.len() {
+                    if tok == "--event" {
+                        event = Some(args[i + 1].clone());
+                    } else {
+                        data = Some(args[i + 1].clone());
+                    }
+                    i += 2;
+                } else {
+                    return argparse_err(&format!(
+                        "argument {tok}: expected one argument"
+                    ));
+                }
+            }
+            "--json" => {
+                json_flag = true;
+                i += 1;
+            }
+            _ if tok.starts_with("--event=") => {
+                event = Some(tok["--event=".len()..].to_string());
+                i += 1;
+            }
+            _ if tok.starts_with("--data=") => {
+                data = Some(tok["--data=".len()..].to_string());
+                i += 1;
+            }
+            _ if tok.starts_with('-') && tok.len() > 1 => {
+                unrecognized.push(tok.clone());
+                i += 1;
+            }
+            _ => {
+                if cmd.is_none() {
+                    cmd = Some(tok.clone());
+                } else if arg.is_none() {
+                    arg = Some(tok.clone());
+                } else {
+                    unrecognized.push(tok.clone());
+                }
+                i += 1;
+            }
+        }
+    }
+
+    // argparse precedence: the required-cmd check fires BEFORE the
+    // unrecognized-args report (observed: `--zz` alone → required cmd).
+    let Some(cmd) = cmd else {
+        return argparse_err("the following arguments are required: cmd");
+    };
+    if !matches!(
+        cmd.as_str(),
+        "record" | "check" | "list" | "promote" | "hook"
+    ) {
+        return argparse_err(&format!(
+            "argument cmd: invalid choice: '{cmd}' (choose from 'record', \
+             'check', 'list', 'promote', 'hook')"
+        ));
+    }
+    if !unrecognized.is_empty() {
+        return argparse_err(&format!(
+            "unrecognized arguments: {}",
+            unrecognized.join(" ")
+        ));
+    }
+
+    let arg = arg.unwrap_or_default();
+    match cmd.as_str() {
+        "record" => main_record(&arg, json_flag, rows, now),
+        "check" => main_check(&arg, json_flag, rows),
+        "list" => main_list(json_flag, rows),
+        "promote" => main_promote(&arg, rows, now),
+        _ => main_hook(event.as_deref(), data.as_deref(), rows),
+    }
+}
+
+fn main_record(arg: &str, json_flag: bool, rows: &[Value], now: &str) -> TgMainOutcome {
+    let mut out = String::new();
+    if arg.is_empty() {
+        out.push_str("usage: task_gate.py record '<instruction>'\n");
+        return TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after: rows.to_vec() };
+    }
+    let decision = record_instruction(rows, arg, now);
+    let rows_after = apply_record(rows, &decision, now);
+    // the oracle prints the FACADE result, not the kernel decision:
+    // matched → {matched, id, count, status, promoted}; new row →
+    // {matched, id: lastrowid, count: 1, status: "observing"}
+    let facade_result = if decision.get("matched").and_then(Value::as_bool).unwrap_or(false) {
+        json!({
+            "matched": true,
+            "id": decision.get("id").cloned().unwrap_or(Value::from(0)),
+            "count": decision.get("count").cloned().unwrap_or(Value::from(1)),
+            "status": decision.get("status").cloned().unwrap_or(Value::String(String::new())),
+            "promoted": decision.get("promoted").cloned().unwrap_or(Value::Bool(false)),
+        })
+    } else {
+        let new_id = rows_after
+            .last()
+            .and_then(|r| r.get("id").cloned())
+            .unwrap_or(Value::from(0));
+        json!({
+            "matched": false,
+            "id": new_id,
+            "count": 1,
+            "status": "observing",
+        })
+    };
+    if json_flag {
+        out.push_str(&py_dumps_pretty(&facade_result, 2));
+        out.push('\n');
+        return TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after };
+    }
+    let promoted = facade_result.get("promoted").and_then(Value::as_bool).unwrap_or(false);
+    let matched = facade_result.get("matched").and_then(Value::as_bool).unwrap_or(false);
+    let count = facade_result.get("count").and_then(Value::as_i64).unwrap_or(1);
+    if promoted {
+        out.push_str(&format!(
+            "PROMOTED: '{arg}' is now automatic (count: {count})\n"
+        ));
+    } else if matched {
+        // result never carries a "threshold" key — .get('threshold', 3)
+        out.push_str(&format!("Noted: '{arg}' (count: {count}/3)\n"));
+    } else {
+        out.push_str(&format!("Observing: '{arg}' (new pattern)\n"));
+    }
+    TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after }
+}
+
+/// The facade's DB writes, applied in memory: matched → UPDATE count/
+/// status/last_seen on the row; new → INSERT with the next autoincrement
+/// id (max id + 1 on a fresh-db history — the rig's seeded corpora).
+fn apply_record(rows: &[Value], decision: &Value, now: &str) -> Vec<Value> {
+    let mut out = rows.to_vec();
+    if decision.get("matched").and_then(Value::as_bool).unwrap_or(false) {
+        let id = decision.get("id").and_then(Value::as_i64).unwrap_or(0);
+        let count = decision.get("count").and_then(Value::as_i64).unwrap_or(1);
+        let status = decision.get("status").and_then(Value::as_str).unwrap_or("");
+        for row in out.iter_mut() {
+            if row.get("id").and_then(Value::as_i64) == Some(id) {
+                row["count"] = Value::from(count);
+                row["status"] = Value::String(status.to_string());
+                row["last_seen"] = Value::String(now.to_string());
+            }
+        }
+    } else if let Some(new_row) = decision.get("new_row") {
+        // the facade INSERT's row as dict(r): full column order with
+        // promoted_at NULL (the INSERT omits it)
+        let next_id = rows
+            .iter()
+            .map(|r| r.get("id").and_then(Value::as_i64).unwrap_or(0))
+            .max()
+            .unwrap_or(0)
+            + 1;
+        let g = |k: &str| new_row.get(k).cloned().unwrap_or(Value::Null);
+        let mut ordered = serde_json::Map::new();
+        ordered.insert("id".to_string(), Value::from(next_id));
+        ordered.insert("instruction".to_string(), g("instruction"));
+        ordered.insert("pattern".to_string(), g("pattern"));
+        ordered.insert("count".to_string(), g("count"));
+        ordered.insert("threshold".to_string(), g("threshold"));
+        ordered.insert("status".to_string(), g("status"));
+        ordered.insert("promoted_at".to_string(), Value::Null);
+        ordered.insert("created_at".to_string(), g("created_at"));
+        ordered.insert("last_seen".to_string(), g("last_seen"));
+        out.push(Value::Object(ordered));
+    }
+    out
+}
+
+fn main_check(arg: &str, json_flag: bool, rows: &[Value]) -> TgMainOutcome {
+    let mut out = String::new();
+    if arg.is_empty() {
+        out.push_str("usage: task_gate.py check '<context>'\n");
+        return TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after: rows.to_vec() };
+    }
+    let matches = check_context(rows, arg)
+        .get("matches")
+        .and_then(Value::as_array)
+        .cloned()
+        .unwrap_or_default();
+    if json_flag {
+        out.push_str(&py_dumps_pretty(&Value::Array(matches), 2));
+        out.push('\n');
+    } else if !matches.is_empty() {
+        for m in &matches {
+            let instr = m.get("instruction").and_then(Value::as_str).unwrap_or("");
+            let count = m.get("count").and_then(Value::as_i64).unwrap_or(0);
+            out.push_str(&format!(
+                "  GATE: {} (seen {count}x)\n",
+                char_slice(instr, 80)
+            ));
+        }
+    } else {
+        out.push_str("  No active gates match\n");
+    }
+    TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after: rows.to_vec() }
+}
+
+fn main_list(json_flag: bool, rows: &[Value]) -> TgMainOutcome {
+    let mut out = String::new();
+    let gates = list_gates(rows);
+    if json_flag {
+        out.push_str(&py_dumps_pretty(&Value::Array(gates.clone()), 2));
+        out.push('\n');
+    } else {
+        for g in &gates {
+            let status = g.get("status").and_then(Value::as_str).unwrap_or("");
+            let id = g.get("id").and_then(Value::as_i64).unwrap_or(0);
+            let count = g.get("count").and_then(Value::as_i64).unwrap_or(0);
+            let instr = g.get("instruction").and_then(Value::as_str).unwrap_or("");
+            out.push_str(&format!(
+                "  [{}] {} ({count}x) {}\n",
+                status_field(status),
+                id_field(id),
+                char_slice(instr, 60)
+            ));
+        }
+    }
+    TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after: rows.to_vec() }
+}
+
+fn main_promote(arg: &str, rows: &[Value], now: &str) -> TgMainOutcome {
+    let mut out = String::new();
+    if arg.is_empty() {
+        out.push_str("usage: task_gate.py promote <id>\n");
+        return TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after: rows.to_vec() };
+    }
+    // the facade's UPDATE: status='active', promoted_at=now WHERE id
+    let target: i64 = arg.parse().unwrap_or(-1);
+    let mut rows_after = rows.to_vec();
+    for row in rows_after.iter_mut() {
+        if row.get("id").and_then(Value::as_i64) == Some(target) {
+            row["status"] = Value::String("active".to_string());
+            row["promoted_at"] = Value::String(now.to_string());
+        }
+    }
+    out.push_str(&format!("Promoted gate {arg}\n"));
+    TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after }
+}
+
+fn main_hook(event: Option<&str>, data: Option<&str>, rows: &[Value]) -> TgMainOutcome {
+    // data parses BEFORE the echo (json.loads on --data); malformed JSON
+    // crashes the oracle — the rig keeps the corpus to parseable data.
+    // handle_hook never touches the DB — rows ride through unchanged.
+    let _ = data;
+    let event = event.unwrap_or("unknown");
+    let echo = handle_hook(event);
+    let out = format!("{}\n", py_dumps(&echo));
+    TgMainOutcome { exit: 0, stdout: out, stderr: String::new(), rows_after: rows.to_vec() }
+}
+
+// ── Python-JSON writer (task_persist law) ──────────────────────────────────
+
+/// json.dumps(v) — compact, ensure_ascii, ", " / ": " separators.
+pub fn py_dumps(v: &Value) -> String {
+    let mut out = String::new();
+    py_write(v, 0, &mut out, false, 2);
+    out
+}
+
+/// json.dumps(v, indent=n).
+pub fn py_dumps_pretty(v: &Value, indent: usize) -> String {
+    let mut out = String::new();
+    py_write(v, 0, &mut out, true, indent);
+    out
+}
+
+fn py_pad(level: usize, indent: usize, out: &mut String) {
+    for _ in 0..indent * level {
+        out.push(' ');
+    }
+}
+
+fn py_write(v: &Value, level: usize, out: &mut String, pretty: bool, indent: usize) {
+    match v {
+        Value::Null => out.push_str("null"),
+        Value::Bool(b) => out.push_str(if *b { "true" } else { "false" }),
+        Value::Number(n) => out.push_str(&py_num(n)),
+        Value::String(s) => py_str(s, out),
+        Value::Array(a) => {
+            if a.is_empty() {
+                out.push_str("[]");
+                return;
+            }
+            out.push('[');
+            for (i, item) in a.iter().enumerate() {
+                if i > 0 {
+                    out.push(',');
+                }
+                if pretty {
+                    out.push('\n');
+                    py_pad(level + 1, indent, out);
+                } else if i > 0 {
+                    out.push(' ');
+                }
+                py_write(item, level + 1, out, pretty, indent);
+            }
+            if pretty {
+                out.push('\n');
+                py_pad(level, indent, out);
+            }
+            out.push(']');
+        }
+        Value::Object(o) => {
+            if o.is_empty() {
+                out.push_str("{}");
+                return;
+            }
+            out.push('{');
+            for (i, (k, val)) in o.iter().enumerate() {
+                if i > 0 {
+                    out.push(',');
+                }
+                if pretty {
+                    out.push('\n');
+                    py_pad(level + 1, indent, out);
+                } else if i > 0 {
+                    out.push(' ');
+                }
+                py_str(k, out);
+                out.push_str(": ");
+                py_write(val, level + 1, out, pretty, indent);
+            }
+            if pretty {
+                out.push('\n');
+                py_pad(level, indent, out);
+            }
+            out.push('}');
+        }
+    }
+}
+
+fn py_num(n: &serde_json::Number) -> String {
+    if let Some(i) = n.as_i64() {
+        return i.to_string();
+    }
+    if let Some(u) = n.as_u64() {
+        return u.to_string();
+    }
+    match n.as_f64() {
+        Some(f) => {
+            if f.fract() == 0.0 && f.abs() < 1e15 {
+                format!("{f:.1}") // Python renders integral floats "2.0"
+            } else {
+                format!("{f}")
+            }
+        }
+        None => "NaN".to_string(),
+    }
+}
+
+/// json.dumps ensure_ascii escaping: control chars < 0x20 and everything
+/// >= 0x7f become \uXXXX (surrogate pairs above BMP); DEL 0x7f is escaped.
+fn py_str(s: &str, out: &mut String) {
+    out.push('"');
+    for c in s.chars() {
+        let cp = c as u32;
+        match cp {
+            0x22 => out.push_str("\\\""),
+            0x5C => out.push_str("\\\\"),
+            0x08 => out.push_str("\\b"),
+            0x09 => out.push_str("\\t"),
+            0x0A => out.push_str("\\n"),
+            0x0C => out.push_str("\\f"),
+            0x0D => out.push_str("\\r"),
+            _ if !(0x20..=0x7E).contains(&cp) => {
+                if cp < 0x1_0000 {
+                    out.push_str(&format!("\\u{:04x}", cp));
+                } else {
+                    let off = cp - 0x1_0000;
+                    let hi = 0xD800u32 + (off >> 10);
+                    let lo = 0xDC00u32 + (off & 0x3FF);
+                    out.push_str(&format!("\\u{:04x}\\u{:04x}", hi, lo));
+                }
+            }
+            _ => out.push(c),
+        }
+    }
+    out.push('"');
+}
+
 // ── FFI (doc-envelope law) ───────────────────────────────────────────────────
 
 /// Typed FFI failure lattice behind the doc-envelope Result — the