feat(behavior): close PARTIAL gaps — dichotomy wave

Audit 6/14 → 14/14 FULL (missing fns _bh_core, connect,
_correction_matches_python, _habit_matches_python, _reinforce_rust,
_reinforce_python, list_items, behavior_stats). All ported
name-preserving into src/rust/behavior as REAL public fns, rows-in/
decisions-out — SQLite open/exec IO stays Python (no crate in this
repo links a sqlite driver; the memory_storage precedent):

* _bh_core() -> Option<BhCore> — the core probe, RUNE_BEHAVIOR_CORE=
  python pin honored (the in-process core has no load-failure branch).
* connect() -> ConnectSpec — path + WAL pragma + the 7 schema stmts
  VERBATIM as sqlite stores them on a fresh oracle connect.
* _correction_matches_python/_habit_matches_python — the inline
  Python-pin predicates over row structs.
* _reinforce_rust/_reinforce_python — both oracle halves as one
  fetch-row→math→write-back ReinforcePlan (result dict + UPDATE
  payload + activation-log row; ts an argument, no-clock law). The
  ep-shim's habitmath IS this crate's reinforce math — the battery
  replays both names and pins their coincidence.
* list_items — filters, ORDER BY confidence DESC (stable sort = the
  sqlite rowid-order tie law, pinned by a frozen tie case), LIMIT,
  item_type key law.
* behavior_stats — the five counters over row collections.

Row structs gain the stored columns (created_at/last_reinforced/
supersedes_id, decay_rate/last_activated) for verbatim row replay.
New FFI: bh_connect, bh_core_probe, bh_python_predicates,
bh_reinforce_plan, bh_list_items, bh_behavior_stats (doc-envelope
law, no free symbol).

Truth: gen_behavior_truth.py now pins the clock (RUNE_FROZEN_NOW →
freeze_truth._pin_clock over behavior.datetime) — every timestamp in
the corpus frozen, regeneration byte-identical x2; new sections:
pre-rows for all reinforce cases, 42+36-row predicate matrix, 5 full
list outputs, stats_full row dumps, connect schema (sqlite_master,
sqlite_% filtered), core_probe, tie pin. bh_pure_truth.json: 81-case
freeze_truth envelope corpus — freeze_truth --verify byte-identical.

Divergence documented (oracle behavior, not a bug): the ep kernel's
Num wire law renders integral confidences as JSON integers in RESULT
dicts ("confidence": 1) while sqlite REAL read-back renders 1.0 —
the plan applies the wire law to results, REAL floats to write-backs.

Battery: 17 tests — full pre→final row replay (timestamps verbatim),
schema verbatim, predicate matrix, list/stats verbatim, pure corpus.
clippy --all-targets -D warnings 0. Parity
scripts/check_behavior_parity.py: FROZEN 137/137 pure-fn replay
through the cdylib FFI + LIVE 14/14 facade-vs-pin. Python 7/7 both
pins (delegation live + RUNE_BEHAVIOR_CORE=python).

docs/dichotomy-rebase-honest-audit.md

@@ -2,14 +2,14 @@
 
 _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL only when 80%+ of its Python functions have name-preserving counterparts in the Go/Rust implementation — directories without the functions are facades and count as NOT TOUCHED._
 
-**FULL completion: 48%** (92/192 runtime modules at FULL)
+**FULL completion: 48%** (93/192 runtime modules at FULL)
 
-Function-level detail: 67% (1529/2279 runtime functions matched) — secondary; includes in-flight work
+Function-level detail: 67% (1537/2279 runtime functions matched) — secondary; includes in-flight work
 
 | Status | Modules | Lines | Meaning |
 |---|---|---|---|
-| FULL | 92 | 38822 | 80%+ of functions ported |
-| PARTIAL | 60 | 23960 | 20–80% — real logic exists, gaps named below |
+| FULL | 93 | 39359 | 80%+ of functions ported |
+| PARTIAL | 59 | 23423 | 20–80% — real logic exists, gaps named below |
 | NOT TOUCHED | 40 | 11513 | no name-preserving implementation |
 | Total runtime | 192 | 74295 | |
 
@@ -70,7 +70,6 @@ Function-level detail: 67% (1529/2279 runtime functions matched) — secondary;
 | claim_audit | 484 | 4/14 | _get_warm_neuron_store, reframe, _graph_has, _warm_has, verify, journal |
 | capability_broker | 798 | 10/19 | _native_tool_schemas, load_registry, _cards_are_stale, load_cards, generate_card, build_cards |
 | token_optimizer | 514 | 16/25 | get_optimizer, _tc_core, _to_core, to_dict, _to_state, _apply_state |
-| behavior | 537 | 6/14 | _bh_core, connect, _correction_matches_python, _habit_matches_python, _reinforce_rust, _reinforce_python |
 | builder_bridge | 326 | 3/11 | _construction_store, _active_construction_lines, _default_selector, _cached_memory, _get_selector, _retrieve_memory |
 | drift_core | 471 | 3/11 | _ms_core, _now_iso, promote_entry, compute_fama, compute_drift_velocity, compute_drift_direction |
 | time_awareness | 402 | 12/20 | _clock_args, _ta_core, get_time_awareness, to_dict, _to_state, _apply_state |
@@ -154,6 +153,7 @@ Function-level detail: 67% (1529/2279 runtime functions matched) — secondary;
 | graph_memory | 563 | 9/9 | RUST |
 | drift_ledger | 548 | 20/21 | RUST |
 | construction_pattern | 547 | 13/13 | RUST |
+| behavior | 537 | 14/14 | RUST |
 | failure_auto_filer | 527 | 24/24 | GO |
 | build_mode | 521 | 33/33 | GO |
 | growth_delta | 520 | 17/18 | GO |

scripts/check_behavior_parity.py

@@ -3,9 +3,15 @@
 oracle path, over the SAME case matrix the truth generator froze.
 Run: python3 scripts/check_behavior_parity.py  -> x/y lines, exit 1 on any
 divergence.
+
+FROZEN layer (gap-close): the new pure-fn surface (connect schema, core
+probe, python-pin predicates, reinforce plans, list_items,
+behavior_stats) replayed through the cdylib's ctypes FFI against the
+frozen oracle truth — live parity for the ported fns.
 """
 from __future__ import annotations
 
+import ctypes
 import json
 import os
 import sys
@@ -76,6 +82,109 @@ def both(name, fn):
     check(name, rust, py)
 
 
+# ── FROZEN layer: the gap-close fns through the cdylib FFI ────────────
+FROZEN_PASS = 0
+FROZEN_FAIL = 0
+
+
+def frozen_check(name, got, want):
+    global FROZEN_PASS, FROZEN_FAIL
+    if got == want:
+        FROZEN_PASS += 1
+    else:
+        FROZEN_FAIL += 1
+        DIVERGENCES.append((f"FROZEN {name}", got, want))
+
+
+_new_lib = ctypes.CDLL(str(bh_loader._LIB_PATH))
+for _sym in ("bh_connect", "bh_core_probe", "bh_python_predicates",
+             "bh_reinforce_plan", "bh_list_items", "bh_behavior_stats"):
+    _fn = getattr(_new_lib, _sym)
+    _fn.argtypes = [ctypes.c_void_p]
+    _fn.restype = ctypes.c_void_p
+
+
+def ffi(sym, payload):
+    data = json.dumps(payload).encode()
+    buf = ctypes.create_string_buffer(data)
+    ptr = getattr(_new_lib, sym)(ctypes.cast(buf, ctypes.c_void_p))
+    if not ptr:
+        raise RuntimeError(f"{sym} returned null")
+    out = ctypes.string_at(ptr).decode()
+    bh_loader._free(ptr)
+    doc = json.loads(out)
+    if not doc.get("ok"):
+        raise RuntimeError(doc.get("err", f"{sym} failed"))
+    return doc["value"]
+
+
+os.environ["RUNE_BEHAVIOR_CORE"] = "python"
+_truth = json.loads((Path(__file__).resolve().parents[1]
+                     / "src" / "rust" / "behavior" / "src" / "bh_truth.json").read_text())
+
+# connect: schema + pragma through the FFI (the frozen truth carries no
+# path — no absolute paths in frozen truth; the path law is asserted
+# separately: memory-dir relative, behavior.db suffix)
+_connect_got = ffi("bh_connect", {})
+frozen_check("connect-schema",
+             {"journal_mode": _connect_got["journal_mode"], "schema": _connect_got["schema"]},
+             {"journal_mode": _truth["cases"]["connect"]["journal_mode"],
+              "schema": _truth["cases"]["connect"]["schema"]})
+if not _connect_got["path"].endswith("/behavior.db"):
+    FROZEN_FAIL += 1
+    DIVERGENCES.append(("FROZEN connect-path", _connect_got["path"], "*/behavior.db"))
+else:
+    FROZEN_PASS += 1
+# core probe under the pin (frozen core: null)
+frozen_check("core-probe-pin", ffi("bh_core_probe", {}), {"available": False})
+
+# python-pin predicate matrix
+for c in _truth["cases"]["predicates"]["corrections"]:
+    got = ffi("bh_python_predicates", {"kind": "correction", "row": c["row"],
+                                       "context": c["context"], "context_low": c["context_low"]})
+    frozen_check(f"pred-corr-{c['row']['id']}-{c['context'][:12]}", got, c["matched"])
+for h in _truth["cases"]["predicates"]["habits"]:
+    got = ffi("bh_python_predicates", {"kind": "habit", "row": h["row"], "context": h["context"]})
+    frozen_check(f"pred-habit-{h['row']['id']}-{h['context'][:12]}", got, h["matched"])
+
+# reinforce plans: full pre->final threading through BOTH paths
+for key in ("reinforce_habit", "reinforce_correction"):
+    for case in _truth["cases"][key]:
+        name = case["name"]
+        if case["pre_row"] is None:
+            item_id = 99991 if name == "habit-not-found" else 99992
+            payload = {"item_id": item_id, "outcome": case["outcome"],
+                       "item_type": case["item_type"], "row": None, "ts": "ts"}
+            frozen_check(f"reinf-{name}", ffi("bh_reinforce_plan", payload)["result"],
+                         case["results"][0])
+            continue
+        ts = case["pre_row"]["created_at"]
+        cur = dict(case["pre_row"])
+        for i in range(case["n"]):
+            payload = {"item_id": case["pre_row"]["id"], "outcome": case["outcome"],
+                       "item_type": case["item_type"], "row": cur, "ts": ts, "path": "rust"}
+            plan = ffi("bh_reinforce_plan", payload)
+            twin = ffi("bh_reinforce_plan", {**payload, "path": "python"})
+            frozen_check(f"reinf-{name}-step{i}-halves", plan, twin)
+            frozen_check(f"reinf-{name}-step{i}", plan["result"], case["results"][i])
+            cur.update(plan["writeback"])
+        frozen_check(f"reinf-{name}-final", cur, case["final_row"])
+
+# list_items + behavior_stats over the frozen row collections
+_sf = _truth["cases"]["stats_full"]
+for case in _truth["cases"]["list_full"]:
+    got = ffi("bh_list_items", {"corrections": _sf["corrections"], "habits": _sf["habits"],
+                                "item_type": case["item_type"], "limit": case["limit"]})
+    frozen_check(f"list-{case['name']}", got, case["result"])
+frozen_check("stats", ffi("bh_behavior_stats", {"corrections": _sf["corrections"],
+                                                "habits": _sf["habits"],
+                                                "activations": _sf["activations"]}),
+             _sf["stats"])
+os.environ["RUNE_BEHAVIOR_CORE"] = "auto"
+
+# ── LIVE layer: the delegating facade vs the python pin ───────────────
+
+
 # ── add shapes ────────────────────────────────────────────────────────
 def add_shapes():
     out = [bh.add_correction("say X", "say Y"),
@@ -174,9 +283,10 @@ def list_stats_shapes():
 
 both("list-stats", list_stats_shapes)
 
+print(f"FROZEN {FROZEN_PASS}/{FROZEN_PASS + FROZEN_FAIL} pure-fn replay through cdylib FFI")
 print(f"{PASS}/{PASS + FAIL} live parity (rust facade vs python pin)")
 for name, rust, py in DIVERGENCES:
     print(f"DIVERGENCE {name}:")
     print(f"  rust: {json.dumps(rust, sort_keys=True)[:400]}")
     print(f"  py:   {json.dumps(py, sort_keys=True)[:400]}")
-sys.exit(1 if FAIL else 0)
+sys.exit(1 if (FAIL or FROZEN_FAIL) else 0)

scripts/gen_behavior_truth.py

@@ -1,27 +1,40 @@
 #!/usr/bin/env python3
 """Freeze behavior oracle truth for the Rust port battery
-(dichotomy wave #2479 tranche, batch: behavior).
+(dichotomy wave #2479 tranche, batch: behavior; gap-close pass).
 
-behavior.py's decision surface (9 public fns):
+behavior.py's decision surface (14 fns):
   connect, add_correction, add_habit, match_context, reinforce,
-  list_items, behavior_stats, main (CLI — not frozen; argparse/IO),
-  PLUGIN (data — embedded verbatim in the crate).
+  _reinforce_rust, _reinforce_python, _correction_matches_python,
+  _habit_matches_python, _bh_core, list_items, behavior_stats,
+  main (CLI — not frozen; argparse/IO), PLUGIN (data — embedded
+  verbatim in the crate).
 
 The decision math (reinforce counting/confidence/promotion/state) is
 served IN-PROCESS by the Rust evidence_provenance cdylib through
 ep_loader — the oracle IS the strangler shim. The frozen truth pins
-RUNE_BEHAVIOR_CORE=python (no-op today; the POISONED-TRUTH LAW: a
-future delegating facade must never silently record kernel output as
-oracle truth).
+RUNE_BEHAVIOR_CORE=python (the POISONED-TRUTH LAW: a future delegating
+facade must never silently record kernel output as oracle truth).
 
 Frozen: add_correction/add_habit return shapes, match_context branch
-matrix (pattern/domain/confidence/match_type/invalid-regex), reinforce
-branch matrix (habit/correction x success/failure/neutral x
-promotion/archival/not-found), list_items filters+limits, stats
-counts. SQLite rows replayed as dicts (row_factory=Row -> dict(c)).
+matrix (pattern/domain/confidence/match_type/invalid-regex), the
+python-pin predicate matrix (every row × context through
+_correction_matches_python/_habit_matches_python), reinforce branch
+matrix (habit/correction x success/failure/neutral x
+promotion/archival/not-found, WITH pre-rows), list_items filters+
+limits+FULL row outputs, stats counts + the row collections they count,
+connect's schema (sqlite_master DDL, creation order), the _bh_core pin
+probe. SQLite rows replayed as dicts (row_factory=Row -> dict(c)).
 
-Output: /tmp/behavior_truth.json — replayed VERBATIM by
-src/rust/behavior/src/branch_battery.rs.
+Clock law: RUNE_FROZEN_NOW pins behavior.datetime (the freeze_truth
+harness clock) — every created_at/last_reinforced/last_activated in the
+corpus is frozen, so regeneration is byte-identical and the battery
+replays timestamps VERBATIM.
+
+Outputs:
+  /tmp/behavior_truth.json            — replayed VERBATIM by
+        src/rust/behavior/src/branch_battery.rs
+  src/rust/behavior/bh_pure_truth.json — freeze_truth-envelope corpus
+        (pure-fn cases; byte-verified via freeze_truth.py --verify)
 """
 from __future__ import annotations
 
@@ -33,14 +46,24 @@ from pathlib import Path
 
 os.environ["RUNE_BEHAVIOR_CORE"] = "python"  # freeze the ORACLE, never the kernel
 
+FROZEN_NOW = "2026-10-13T00:00:00"
+os.environ["RUNE_FROZEN_NOW"] = FROZEN_NOW
+
 _tmp = tempfile.mkdtemp(prefix="bh-truth-")
 os.environ["RUNE_HOME"] = _tmp
 os.environ["RUNE_MEMORY_DIR"] = os.path.join(_tmp, "memory")
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import freeze_truth  # noqa: E402  (shared harness: clock pin + envelope)
 
 from rune import behavior as bh  # noqa: E402
 
+freeze_truth._pin_clock(bh)  # behavior.datetime -> frozen clock
+
+CRATE = Path(__file__).resolve().parents[1] / "src" / "rust" / "behavior"
+
 
 def row(table, rid):
     db = bh.connect()
@@ -114,19 +137,47 @@ cases["match"].append(match_case("habit-regex-hit", "run the unit tests"))
 cases["match"].append(match_case("low-conf-filtered", "zzz-no-match-ever quiet"))
 cases["match"].append(match_case("bad-regex-skipped", "([unclosed anything"))
 
+# ── python-pin predicate matrix (gap-close surface) ───────────────────
+# Every correction/habit row present at match time × every match
+# context, through the INLINE oracle predicates — the verbatim replay
+# corpus for _correction_matches_python / _habit_matches_python.
+MATCH_CONTEXTS = [
+    "deploy to shell now", "deploy the editor", "unrelated chatter here",
+    "run the unit tests", "zzz-no-match-ever quiet", "([unclosed anything",
+]
+_db = bh.connect()
+_corr_rows = [dict(r) for r in _db.execute("SELECT * FROM corrections").fetchall()]
+_habit_rows = [dict(r) for r in _db.execute("SELECT * FROM habits").fetchall()]
+_db.close()
+cases["predicates"] = {
+    "corrections": [
+        {"context": ctx, "row": dict(c), "context_low": ctx.lower(),
+         "matched": bh._correction_matches_python(dict(c), ctx, ctx.lower())}
+        for ctx in MATCH_CONTEXTS for c in _corr_rows
+    ],
+    "habits": [
+        {"context": ctx, "row": dict(h),
+         "matched": bh._habit_matches_python(dict(h), ctx)}
+        for ctx in MATCH_CONTEXTS for h in _habit_rows
+    ],
+}
+
 # ── reinforce matrix ──────────────────────────────────────────────────
 def reinf_case(name, item_type, outcome, seed_fn, pre=None, n=1):
     rid = seed_fn()["added"]
+    table = "habits" if item_type == "habit" else "corrections"
     if pre:
         db = bh.connect()
         for k, v in pre.items():
-            db.execute(f"UPDATE {'habits' if item_type == 'habit' else 'corrections'} SET {k}=? WHERE id=?", (v, rid))
+            db.execute(f"UPDATE {table} SET {k}=? WHERE id=?", (v, rid))
         db.commit()
         db.close()
+    pre_row = row(table, rid)
     outs = [bh.reinforce(rid, outcome, item_type=item_type) for _ in range(n)]
     return {"name": name, "item_type": item_type, "outcome": outcome,
-            "pre": pre or {}, "n": n, "results": outs,
-            "final_row": row("habits" if item_type == "habit" else "corrections", rid)}
+            "pre": pre or {}, "n": n, "pre_row": pre_row,
+            "results": outs,
+            "final_row": row(table, rid)}
 
 
 cases["reinforce_habit"].append(reinf_case(
@@ -159,12 +210,12 @@ cases["reinforce_correction"].append(reinf_case(
     lambda: bh.add_correction("x5", "y5"), n=1))
 cases["reinforce_habit"].append(
     {"name": "habit-not-found", "item_type": "habit", "outcome": "success",
-     "pre": {}, "n": 1,
+     "pre": {}, "n": 1, "pre_row": None,
      "results": [bh.reinforce(99991, "success", item_type="habit")],
      "final_row": None})
 cases["reinforce_correction"].append(
     {"name": "correction-not-found", "item_type": "correction",
-     "outcome": "success", "pre": {}, "n": 1,
+     "outcome": "success", "pre": {}, "n": 1, "pre_row": None,
      "results": [bh.reinforce(99992, "success", item_type="correction")],
      "final_row": None})
 
@@ -183,14 +234,77 @@ cases["list"].append({"name": "limit-1", "n_corr": len(got.get("corrections", []
 # ── stats ─────────────────────────────────────────────────────────────
 cases["stats"].append({"name": "counts", "expected": bh.behavior_stats()})
 
+# ── gap-close surfaces: tie pin, full rows, list/stats replay ────────
+# Tie law: two lowest-id corrections share the top confidence — SQLite
+# returns rowid order for equal ORDER BY keys; the port's stable sort
+# must reproduce it (pinned here, not assumed).
+db = bh.connect()
+db.execute("UPDATE corrections SET confidence=0.85 WHERE id=(SELECT MIN(id) FROM corrections)")
+db.execute("UPDATE corrections SET confidence=0.85 WHERE id=(SELECT MIN(id) FROM corrections WHERE id > (SELECT MIN(id) FROM corrections))")
+db.commit()
+db.close()
+
+_dump_db = bh.connect()
+_all_corrections = [dict(r) for r in _dump_db.execute("SELECT * FROM corrections ORDER BY id").fetchall()]
+_all_habits = [dict(r) for r in _dump_db.execute("SELECT * FROM habits ORDER BY id").fetchall()]
+_n_activations = _dump_db.execute("SELECT COUNT(*) FROM habit_activations").fetchone()[0]
+_dump_db.close()
+
+cases["list_full"] = []
+for t, lim in [("all", 20), ("correction", 20), ("habit", 20), ("all", 1), ("all", 3)]:
+    got = bh.list_items(item_type=t, limit=lim)
+    cases["list_full"].append({
+        "name": f"{t}-limit{lim}", "item_type": t, "limit": lim,
+        "result": json.loads(json.dumps(got))})
+cases["stats_full"] = {
+    "corrections": json.loads(json.dumps(_all_corrections)),
+    "habits": json.loads(json.dumps(_all_habits)),
+    "activations": _n_activations,
+    "stats": bh.behavior_stats(),
+}
+
+# ── connect (schema as sqlite stores it) + the _bh_core pin probe ────
+db = bh.connect()
+cases["connect"] = {
+    "journal_mode": db.execute("PRAGMA journal_mode").fetchone()[0],
+    "schema": [
+        {"kind": r["type"], "name": r["name"], "sql": r["sql"]}
+        for r in db.execute(
+            "SELECT type, name, sql FROM sqlite_master "
+            "WHERE sql IS NOT NULL AND name NOT LIKE 'sqlite_%'").fetchall()],
+}
+db.close()
+cases["core_probe"] = {"runes_behavior_core_env": os.environ["RUNE_BEHAVIOR_CORE"],
+                       "core": bh._bh_core()}
+
 truth = {"frozen": "behavior decision surface (reinforce math via ep strangler shim)",
+         "frozen_now": FROZEN_NOW,
          "oracle": "src/rune/behavior.py",
          "cases": cases,
          "plugin": bh.PLUGIN}
 
 out = Path("/tmp/behavior_truth.json")
 out.write_text(json.dumps(truth, indent=1, default=str) + "\n")
+
+# ── freeze_truth-envelope corpus: the PURE fn surface ────────────────
+# (predicates over row dicts + reinforce not-found envelopes + the
+# _bh_core pin probe — byte-verified via freeze_truth.py --verify)
+pure_cases = []
+for ctx in MATCH_CONTEXTS:
+    for c in _corr_rows:
+        pure_cases.append(("_correction_matches_python", (dict(c), ctx, ctx.lower()), {}))
+    for h in _habit_rows:
+        pure_cases.append(("_habit_matches_python", (dict(h), ctx), {}))
+pure_cases.append(("reinforce", (99991, "success"), {"item_type": "habit"}))
+pure_cases.append(("reinforce", (99992, "success"), {"item_type": "correction"}))
+pure_cases.append(("_bh_core", (), {}))
+pure_truth = freeze_truth.freeze("rune.behavior", pure_cases, oracle_module="rune.behavior")
+freeze_truth.write_envelope(FROZEN_NOW, pure_truth, CRATE / "bh_pure_truth.json")
+
 n = (len(cases["add_correction"]) + len(cases["add_habit"]) + len(cases["match"])
      + len(cases["reinforce_habit"]) + len(cases["reinforce_correction"])
      + len(cases["list"]) + len(cases["stats"]))
-print(f"froze {n} cases -> {out}")
+print(f"froze {n} cases + {len(cases['predicates']['corrections'])}+"
+      f"{len(cases['predicates']['habits'])} predicate rows + "
+      f"{len(cases['list_full'])} full lists -> {out}")
+print(f"froze {len(pure_truth)} pure cases -> {CRATE / 'bh_pure_truth.json'}")

src/rust/behavior/src/branch_battery.rs

@@ -3,14 +3,25 @@
 //! scripts/gen_behavior_truth.py against src/rune/behavior.py, the
 //! strangler shim serving ep_loader decision math) and is replayed
 //! VERBATIM: add/insert shapes, the match_context branch matrix, the
-//! reinforce matrix (habit/correction x outcome x promotion/archival),
-//! list filters, and stats counts.
+//! python-pin predicate matrix, the reinforce matrix (habit/correction
+//! x outcome x promotion/archival, full pre→final row replay), the
+//! connect schema, list filters + FULL row outputs, and stats counts.
+//! bh_pure_truth.json (freeze_truth envelope) replays through the same
+//! public fns — the in-crate twin of `freeze_truth.py --verify`.
 
-use super::{correction_matches, correction_reinforce, correction_visible, habit_matches,
-            habit_reinforce, habit_visible, parse_scope_tags, plugin, HabitRow};
+use super::{behavior_stats, connect, list_items, plugin,
+            _bh_core, _correction_matches_python, _habit_matches_python,
+            _reinforce_python, _reinforce_rust,
+            correction_matches, correction_reinforce, correction_visible, habit_matches,
+            habit_reinforce, habit_visible, parse_scope_tags, CorrectionRow, HabitRow};
 use serde_json::Value;
 
 const TRUTH: &str = include_str!("bh_truth.json");
+const PURE_TRUTH: &str = include_str!("../bh_pure_truth.json");
+
+/// Env-touching tests (RUNE_BEHAVIOR_CORE / RUNE_MEMORY_DIR) serialize
+/// on one lock — std::env is process-global.
+static ENV_LOCK: std::sync::Mutex<()> = std::sync::Mutex::new(());
 
 fn truth() -> Value {
     serde_json::from_str(TRUTH).expect("bh_truth.json parses")
@@ -353,3 +364,232 @@ fn test_correction_update_dict_shape() {
     assert_eq!(upd["confidence"].as_f64(), Some(0.2));
     assert_eq!(upd["evidence_count"].as_f64(), Some(1.0), "failure keeps count");
 }
+
+// ── gap-close surfaces ───────────────────────────────────────────────────
+
+fn corr_row_full(row: &Value) -> CorrectionRow {
+    serde_json::from_value(row.clone()).expect("correction row deserializes")
+}
+
+/// connect: the schema DDL replayed verbatim (creation order, sqlite's
+/// stored text), the WAL pragma, and the path law.
+#[test]
+fn test_connect_schema() {
+    let t = truth();
+    let frozen = &t["cases"]["connect"];
+    let spec = connect();
+    assert_eq!(spec.journal_mode, frozen["journal_mode"].as_str().unwrap());
+    let frozen_schema = frozen["schema"].as_array().unwrap();
+    assert_eq!(spec.schema.len(), frozen_schema.len(), "schema stmt count");
+    for (stmt, want) in spec.schema.iter().zip(frozen_schema) {
+        assert_eq!(stmt.kind, want["kind"].as_str().unwrap(), "{} kind", stmt.name);
+        assert_eq!(stmt.name, want["name"].as_str().unwrap(), "stmt name");
+        assert_eq!(stmt.sql, want["sql"].as_str().unwrap(), "{} DDL verbatim", stmt.name);
+    }
+    let _guard = ENV_LOCK.lock().unwrap();
+    let key = "RUNE_MEMORY_DIR";
+    let saved = std::env::var(key).ok();
+    std::env::set_var(key, "/tmp/bh-battery-mem");
+    assert_eq!(super::db_path(), "/tmp/bh-battery-mem/behavior.db");
+    // the spec snapshots the path at connect() time
+    assert_eq!(connect().path, "/tmp/bh-battery-mem/behavior.db");
+    match saved {
+        Some(v) => std::env::set_var(key, v),
+        None => std::env::remove_var(key),
+    }
+}
+
+/// _bh_core: the RUNE_BEHAVIOR_CORE=python pin honored (the frozen
+/// probe recorded None under the pin); every other value resolves.
+#[test]
+fn test_core_probe_pin() {
+    let t = truth();
+    let frozen = &t["cases"]["core_probe"];
+    assert_eq!(frozen["runes_behavior_core_env"].as_str(), Some("python"));
+    assert!(frozen["core"].is_null(), "oracle under pin → None");
+    let _guard = ENV_LOCK.lock().unwrap();
+    let key = "RUNE_BEHAVIOR_CORE";
+    let saved = std::env::var(key).ok();
+    std::env::set_var(key, "python");
+    assert!(_bh_core().is_none(), "pin → fallback");
+    std::env::remove_var(key);
+    assert!(_bh_core().is_some(), "auto → core resolved");
+    std::env::set_var(key, "auto");
+    assert!(_bh_core().is_some());
+    match saved {
+        Some(v) => std::env::set_var(key, v),
+        None => std::env::remove_var(key),
+    }
+}
+
+/// The python-pin predicate matrix: every frozen (row, context) pair
+/// replays through the inline-predicate ports.
+#[test]
+fn test_python_predicates_matrix() {
+    let t = truth();
+    let p = &t["cases"]["predicates"];
+    let n_corr = p["corrections"].as_array().unwrap().len();
+    let n_habit = p["habits"].as_array().unwrap().len();
+    assert!(n_corr >= 40 && n_habit >= 30, "matrix is full ({n_corr}+{n_habit})");
+    for c in p["corrections"].as_array().unwrap() {
+        let row = corr_row_full(&c["row"]);
+        let ctx = c["context"].as_str().unwrap();
+        let ctx_low = c["context_low"].as_str().unwrap();
+        assert_eq!(
+            _correction_matches_python(&row, ctx, ctx_low),
+            c["matched"].as_bool().unwrap(),
+            "correction {} vs {:?}",
+            row.id, ctx
+        );
+    }
+    for h in p["habits"].as_array().unwrap() {
+        let row = habit_row(&h["row"]);
+        let ctx = h["context"].as_str().unwrap();
+        assert_eq!(
+            _habit_matches_python(&row, ctx),
+            h["matched"].as_bool().unwrap(),
+            "habit {} vs {:?}",
+            row.id, ctx
+        );
+    }
+}
+
+/// The reinforce matrix as full pre→final row replays through
+/// `_reinforce_rust` AND `_reinforce_python` (both oracle halves): every
+/// step's return dict, the write-back threading, and the final row
+/// (timestamps included — the frozen clock pins them).
+#[test]
+fn test_reinforce_plan_replay() {
+    let t = truth();
+    for key in ["reinforce_habit", "reinforce_correction"] {
+        for case in t["cases"][key].as_array().unwrap() {
+            let name = case["name"].as_str().unwrap();
+            let outcome = case["outcome"].as_str().unwrap();
+            let item_type = case["item_type"].as_str().unwrap();
+            let n = case["n"].as_i64().unwrap();
+            if case["pre_row"].is_null() {
+                let id = if name == "habit-not-found" { 99991 } else { 99992 };
+                let plan = _reinforce_rust(id, outcome, item_type, None, "ts");
+                assert_eq!(plan.result, case["results"][0], "{name}: not-found envelope");
+                assert!(plan.writeback.is_null() && plan.activation.is_none(), "{name}");
+                continue;
+            }
+            let ts = case["pre_row"]["created_at"].as_str().unwrap();
+            let mut cur = case["pre_row"].clone();
+            for i in 0..n as usize {
+                let plan = _reinforce_rust(case["pre_row"]["id"].as_i64().unwrap(), outcome, item_type, Some(&cur), ts);
+                let twin = _reinforce_python(case["pre_row"]["id"].as_i64().unwrap(), outcome, item_type, Some(&cur), ts);
+                assert_eq!(plan.result, twin.result, "{name}: step {i} halves coincide");
+                assert_eq!(plan.writeback, twin.writeback, "{name}: step {i} writebacks coincide");
+                assert_eq!(plan.activation, twin.activation, "{name}: step {i} activation rows coincide");
+                assert_eq!(plan.result, case["results"][i], "{name}: step {i} result");
+                if item_type == "habit" {
+                    let act = plan.activation.as_ref().unwrap();
+                    assert_eq!(act["activated_at"].as_str(), Some(ts), "{name}: ts pinned");
+                    assert_eq!(act["context"].as_str(), Some(""), "{name}: logged context ''");
+                }
+                // thread the write-back (the oracle's UPDATE)
+                for (k, v) in plan.writeback.as_object().unwrap() {
+                    cur[k.as_str()] = v.clone();
+                }
+            }
+            assert_eq!(cur, case["final_row"], "{name}: final row (verbatim)");
+        }
+    }
+}
+
+/// list_items over the frozen row collections: every full output
+/// (filters, ORDER BY confidence DESC with the tie law, LIMIT, keys)
+/// replays verbatim.
+#[test]
+fn test_list_items_replay() {
+    let t = truth();
+    let sf = &t["cases"]["stats_full"];
+    let corrections: Vec<CorrectionRow> =
+        serde_json::from_value(sf["corrections"].clone()).unwrap();
+    let habits: Vec<HabitRow> = serde_json::from_value(sf["habits"].clone()).unwrap();
+    for case in t["cases"]["list_full"].as_array().unwrap() {
+        let got = list_items(
+            &corrections,
+            &habits,
+            case["item_type"].as_str().unwrap(),
+            case["limit"].as_i64().unwrap(),
+        );
+        assert_eq!(got, case["result"], "{}: full output", case["name"]);
+    }
+    // tie law: the two 0.85 corrections appear in rowid order behind the
+    // single 1.0 one.
+    let all = list_items(&corrections, &habits, "all", 20);
+    let ids: Vec<i64> = all["corrections"].as_array().unwrap()
+        .iter().map(|c| c["id"].as_i64().unwrap()).collect();
+    let pos = |id: i64| ids.iter().position(|x| *x == id).unwrap();
+    assert!(pos(11) < pos(1) && pos(1) < pos(2), "confidence order then rowid ties: {ids:?}");
+}
+
+/// behavior_stats over the frozen row collections, both captured stats
+/// (pre- and post-tie-pin — counts are unaffected by confidence edits).
+#[test]
+fn test_behavior_stats_replay() {
+    let t = truth();
+    let sf = &t["cases"]["stats_full"];
+    let corrections: Vec<CorrectionRow> =
+        serde_json::from_value(sf["corrections"].clone()).unwrap();
+    let habits: Vec<HabitRow> = serde_json::from_value(sf["habits"].clone()).unwrap();
+    let activations = sf["activations"].as_i64().unwrap();
+    let got = behavior_stats(&corrections, &habits, activations);
+    assert_eq!(got, sf["stats"], "stats_full counts");
+    assert_eq!(got, t["cases"]["stats"][0]["expected"], "matrix-time counts");
+}
+
+/// bh_pure_truth.json (freeze_truth envelope): the pure-fn corpus the
+/// harness byte-verifies against the live oracle, replayed through the
+/// same public fns in-crate.
+#[test]
+fn test_pure_truth_corpus() {
+    let pure: Value = serde_json::from_str(PURE_TRUTH).expect("bh_pure_truth.json parses");
+    let cases = pure["cases"].as_array().unwrap();
+    assert!(cases.len() >= 80, "corpus is full ({} cases)", cases.len());
+    let mut pinned = 0;
+    for c in cases {
+        let fn_name = c["fn"].as_str().unwrap();
+        let args = c["args"].as_array().unwrap();
+        match fn_name {
+            "_correction_matches_python" => {
+                let row = corr_row_full(&args[0]);
+                assert_eq!(
+                    _correction_matches_python(&row, args[1].as_str().unwrap(), args[2].as_str().unwrap()),
+                    c["expected"].as_bool().unwrap()
+                );
+            }
+            "_habit_matches_python" => {
+                let row = habit_row(&args[0]);
+                assert_eq!(
+                    _habit_matches_python(&row, args[1].as_str().unwrap()),
+                    c["expected"].as_bool().unwrap()
+                );
+            }
+            "reinforce" => {
+                let item_type = c["kwargs"]["item_type"].as_str().unwrap();
+                let plan = _reinforce_rust(
+                    args[0].as_i64().unwrap(), args[1].as_str().unwrap(), item_type, None, "ts");
+                assert_eq!(plan.result, c["expected"]);
+            }
+            "_bh_core" => {
+                // frozen under the harness pin (RUNE_BEHAVIOR_CORE=python)
+                let _guard = ENV_LOCK.lock().unwrap();
+                let key = "RUNE_BEHAVIOR_CORE";
+                let saved = std::env::var(key).ok();
+                std::env::set_var(key, "python");
+                assert!(_bh_core().is_none());
+                assert!(c["expected"].is_null());
+                match saved {
+                    Some(v) => std::env::set_var(key, v),
+                    None => std::env::remove_var(key),
+                }
+                pinned += 1;
+            }
+            other => panic!("unknown fn in pure corpus: {other}"),
+        }
+    }
+    assert_eq!(pinned, 1, "exactly one _bh_core pin case");
+}

src/rust/behavior/src/lib.rs

@@ -1,10 +1,10 @@
 //! behavior — behavioral patterns: scoped corrections and habit formation.
 //!
 //! Dichotomy-wave port (#2479 tranche) of `src/rune/behavior.py`'s FULL
-//! decision surface. The oracle is a strangler shim: SQLite I/O + regex
-//! matching live in Python while the reinforce counting/confidence math
-//! runs in-process through the `evidence_provenance` cdylib. This crate
-//! reunites the WHOLE surface in Rust:
+//! decision surface, gap-closed to 14/14. The oracle is a strangler shim:
+//! SQLite I/O lives in Python while the reinforce counting/confidence
+//! math runs in-process through the `evidence_provenance` cdylib. This
+//! crate reunites the WHOLE surface in Rust:
 //!
 //!   * `habit_reinforce` / `correction_reinforce` — the decision math
 //!     (name-preserving port of evidence_provenance's habitmath, which is
@@ -13,20 +13,38 @@
 //!     regex, domain substring, confidence gate, invalid-regex skip),
 //!   * `correction_visible` / `habit_visible` — the SQL WHERE clauses as
 //!     pure predicates (state + confidence gates),
+//!   * `_correction_matches_python` / `_habit_matches_python` — the
+//!     inline Python-pin predicates the oracle falls back to,
+//!   * `_reinforce_rust` / `_reinforce_python` — the two oracle halves of
+//!     reinforce (kernel delegation + ep-shim fallback) as one
+//!     fetch-row→math→write-back plan over row data,
+//!   * `_bh_core` — the core probe (RUNE_BEHAVIOR_CORE=python pin honored),
+//!   * `connect` — the open/create decision surface as data (path, WAL
+//!     pragma, the schema DDL verbatim as sqlite stores it),
+//!   * `list_items` / `behavior_stats` — the list filters/ordering and
+//!     the stats counts over row collections,
 //!   * `PLUGIN` — the module plugin descriptor (frozen data),
 //!   * `db_path` — the memory-dir path resolution (RUNE_MEMORY_DIR override
 //!     → else rune_home()/memory, mirroring memory_env).
 //!
 //! IO stays Python: SQLite open/exec (system libsqlite3 has no write
-//! path in the memory_storage precedent — this crate links NO sqlite and
-//! owns zero SQL text), wall-clock `_now`, argparse CLI, DB_PATH constant.
-//! The facade passes ROWS (fetched Python-side) and receives decisions.
+//! path in the memory_storage precedent — this crate links NO sqlite
+//! driver; `connect` carries the schema as DATA and the facade applies
+//! it), wall-clock `_now` (every timestamp is an argument), argparse
+//! CLI. The facade passes ROWS (fetched Python-side) and receives
+//! decisions.
 //!
 //! FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
 //!   * bh_habit_reinforce({row, outcome}) -> update dict
 //!   * bh_correction_reinforce({row, outcome}) -> update dict
 //!   * bh_match_row({kind, row, context}) -> {"visible", "matched"}
 //!   * bh_plugin() -> the PLUGIN descriptor
+//!   * bh_connect() -> {path, journal_mode, schema}
+//!   * bh_core_probe() -> {"available": bool}
+//!   * bh_python_predicates({kind, row, context, context_low}) -> bool
+//!   * bh_reinforce_plan({item_id, outcome, item_type, row, ts, path}) -> plan
+//!   * bh_list_items({corrections, habits, item_type, limit}) -> result
+//!   * bh_behavior_stats({corrections, habits, activations}) -> stats
 //!
 //! NOTE: no free-symbol export (capability_scoring's cs_free_string serves
 //! the whole process — duplicate symbols break the cdylib link).
@@ -59,6 +77,15 @@ pub struct CorrectionRow {
     pub confidence: f64,
     #[serde(default)]
     pub state: String,
+    // The stored-row columns list_items/stats replay verbatim (the
+    // battery's frozen rows carry them; defaults keep partial dicts —
+    // the reinforce FFI payloads — deserializable).
+    #[serde(default)]
+    pub created_at: Option<String>,
+    #[serde(default)]
+    pub last_reinforced: Option<String>,
+    #[serde(default)]
+    pub supersedes_id: Option<i64>,
 }
 
 fn default_evidence() -> i64 {
@@ -91,14 +118,24 @@ pub struct HabitRow {
     pub min_activations: i64,
     #[serde(default = "default_confidence")]
     pub confidence: f64,
+    #[serde(default = "default_decay")]
+    pub decay_rate: f64,
     #[serde(default)]
     pub state: String,
+    #[serde(default)]
+    pub created_at: Option<String>,
+    #[serde(default)]
+    pub last_activated: Option<String>,
 }
 
 fn default_min_act() -> i64 {
     3
 }
 
+fn default_decay() -> f64 {
+    0.95
+}
+
 // ── reinforce decision math ─────────────────────────────────────────────
 // (Name-preserving port of evidence_provenance habitmath — the formulas
 // this module's shim delegates to. Python semantics preserved exactly:
@@ -273,6 +310,312 @@ pub fn parse_scope_tags(raw: &str) -> serde_json::Value {
     serde_json::from_str(raw).unwrap_or(serde_json::json!({}))
 }
 
+// ── core probe + connect (the gap-close surface) ─────────────────────────
+
+/// The resolved in-process behavior core. Twin of the oracle's loader
+/// MODULE handle (`_bh_core` returns the loader module, never the raw
+/// CDLL — BhCore is that handle's shape in the reunited crate).
+#[derive(Debug, Clone, Copy)]
+pub struct BhCore;
+
+/// The Rust behavior-core probe (name-preserving port of `_bh_core`).
+/// RUNE_BEHAVIOR_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 _bh_core() -> Option<BhCore> {
+    if std::env::var("RUNE_BEHAVIOR_CORE").unwrap_or_default() == "python" {
+        None
+    } else {
+        Some(BhCore)
+    }
+}
+
+/// One schema statement exactly as sqlite stores it on a fresh oracle
+/// connect (creation order = the executescript order; the IF NOT EXISTS
+/// guard is the oracle's apply-time concern and is not part of the
+/// stored text).
+#[derive(Debug, Clone, Copy)]
+pub struct SchemaStmt {
+    pub kind: &'static str,
+    pub name: &'static str,
+    pub sql: &'static str,
+}
+
+const SCHEMA: [SchemaStmt; 7] = [
+    SchemaStmt { kind: "table", name: "corrections", sql: r#"CREATE TABLE corrections (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            original_behavior TEXT,
+            corrected_behavior TEXT NOT NULL,
+            scope_tags TEXT DEFAULT '{}',
+            scope_pattern TEXT DEFAULT '',
+            scope_domain TEXT DEFAULT '',
+            evidence_count INTEGER DEFAULT 1,
+            confidence REAL DEFAULT 0.5,
+            state TEXT DEFAULT 'active',
+            created_at TEXT NOT NULL,
+            last_reinforced TEXT,
+            supersedes_id INTEGER
+        )"# },
+    SchemaStmt { kind: "table", name: "habits", sql: r#"CREATE TABLE habits (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            name TEXT DEFAULT '',
+            trigger_pattern TEXT NOT NULL,
+            context_tags TEXT DEFAULT '[]',
+            action_template TEXT NOT NULL,
+            activation_count INTEGER DEFAULT 0,
+            success_count INTEGER DEFAULT 0,
+            failure_count INTEGER DEFAULT 0,
+            min_activations INTEGER DEFAULT 3,
+            confidence REAL DEFAULT 0.5,
+            decay_rate REAL DEFAULT 0.95,
+            state TEXT DEFAULT 'candidate',
+            created_at TEXT NOT NULL,
+            last_activated TEXT
+        )"# },
+    SchemaStmt { kind: "table", name: "habit_activations", sql: r#"CREATE TABLE habit_activations (
+            id INTEGER PRIMARY KEY AUTOINCREMENT,
+            habit_id INTEGER REFERENCES habits(id),
+            context TEXT NOT NULL,
+            outcome TEXT DEFAULT 'neutral',
+            feedback TEXT DEFAULT '',
+            activated_at TEXT NOT NULL
+        )"# },
+    SchemaStmt { kind: "index", name: "idx_corrections_state", sql: r#"CREATE INDEX idx_corrections_state ON corrections(state)"# },
+    SchemaStmt { kind: "index", name: "idx_corrections_domain", sql: r#"CREATE INDEX idx_corrections_domain ON corrections(scope_domain)"# },
+    SchemaStmt { kind: "index", name: "idx_habits_state", sql: r#"CREATE INDEX idx_habits_state ON habits(state)"# },
+    SchemaStmt { kind: "index", name: "idx_habits_trigger", sql: r#"CREATE INDEX idx_habits_trigger ON habits(trigger_pattern)"# },
+];
+
+/// connect()'s deterministic content as data: where the database lives,
+/// the journal pragma value the oracle sets, and the schema the
+/// executescript creates. The sqlite open/mkdir IO stays Python-side.
+#[derive(Debug, Clone)]
+pub struct ConnectSpec {
+    pub path: String,
+    pub journal_mode: &'static str,
+    pub schema: [SchemaStmt; 7],
+}
+
+/// Open/create the behavior database (name-preserving port of
+/// `connect`): the decision surface — path resolution, WAL journal
+/// mode, the full schema — returned as data for the facade to apply.
+pub fn connect() -> ConnectSpec {
+    ConnectSpec { path: db_path(), journal_mode: "wal", schema: SCHEMA }
+}
+
+// ── the Python-pin predicates (the oracle's inline fallbacks) ────────────
+
+/// The inline per-row correction predicate (name-preserving port of
+/// `_correction_matches_python`): the scope pattern must re.search the
+/// context IGNORECASE (invalid regex → row skipped) and the domain must
+/// occur in context_low (`context.lower()` — precomputed by the oracle
+/// caller; passing anything else is a caller bug).
+pub fn _correction_matches_python(row: &CorrectionRow, context: &str, context_low: &str) -> bool {
+    if !row.scope_pattern.is_empty() {
+        match Regex::new(&format!("(?i){}", row.scope_pattern)) {
+            Ok(re) => {
+                if !re.is_match(context) {
+                    return false;
+                }
+            }
+            Err(_) => return false,
+        }
+    }
+    if !row.scope_domain.is_empty() && !context_low.contains(&row.scope_domain.to_lowercase()) {
+        return false;
+    }
+    true
+}
+
+/// The inline per-row habit predicate (name-preserving port of
+/// `_habit_matches_python`): re.search(trigger, context, IGNORECASE),
+/// invalid regex → row skipped.
+pub fn _habit_matches_python(row: &HabitRow, context: &str) -> bool {
+    match Regex::new(&format!("(?i){}", row.trigger_pattern)) {
+        Ok(re) => re.is_match(context),
+        Err(_) => false,
+    }
+}
+
+// ── reinforce: fetch-row → math → write-back, as a plan over rows ────────
+
+/// One reinforce step's decision output: the facade's return dict, the
+/// UPDATE payload (`ts` pinned by the caller — no-clock law), and (the
+/// habit path) the activation-log INSERT row.
+#[derive(Debug, Clone, Serialize)]
+pub struct ReinforcePlan {
+    pub result: serde_json::Value,
+    pub writeback: serde_json::Value,
+    pub activation: Option<serde_json::Value>,
+}
+
+/// The ep kernel's Num wire law: the oracle's reinforce RESULTS come
+/// through the evidence_provenance cdylib, which renders integral f64s
+/// as JSON integers (its `Num` variant — counts/enums law) while every
+/// other float keeps its decimal point. Write-back payloads do NOT use
+/// this law — they land in REAL columns and read back as floats.
+fn wire_num(x: f64) -> serde_json::Value {
+    if x.is_finite() && x.fract() == 0.0 {
+        serde_json::json!(x as i64)
+    } else {
+        serde_json::json!(x)
+    }
+}
+
+/// The reinforce write-back plan common to both oracle halves. Counts
+/// serialize as integers (sqlite INTEGER-column read-back parity).
+fn reinforce_plan(
+    item_id: i64,
+    outcome: &str,
+    item_type: &str,
+    row: Option<&serde_json::Value>,
+    ts: &str,
+) -> ReinforcePlan {
+    let none = |kind: &str| ReinforcePlan {
+        result: serde_json::json!({"error": format!("{kind} {item_id} not found")}),
+        writeback: serde_json::Value::Null,
+        activation: None,
+    };
+    let row = match row {
+        Some(r) => r,
+        None => return none(if item_type == "habit" { "Habit" } else { "Correction" }),
+    };
+    if item_type == "habit" {
+        let habit: HabitRow = match serde_json::from_value(row.clone()) {
+            Ok(h) => h,
+            Err(_) => return none("Habit"),
+        };
+        let upd = habit_reinforce(&habit, outcome);
+        let (count, success, failure) = (
+            upd["activation_count"].as_f64().unwrap_or_default() as i64,
+            upd["success_count"].as_f64().unwrap_or_default() as i64,
+            upd["failure_count"].as_f64().unwrap_or_default() as i64,
+        );
+        let confidence = upd["confidence"].as_f64().unwrap_or_default();
+        let state = upd["state"].as_str().unwrap_or_default().to_string();
+        ReinforcePlan {
+            result: serde_json::json!({
+                "reinforced": item_id, "confidence": wire_num(confidence),
+                "state": state,
+            }),
+            writeback: serde_json::json!({
+                "activation_count": count, "success_count": success,
+                "failure_count": failure, "confidence": confidence,
+                "state": state, "last_activated": ts,
+            }),
+            activation: Some(serde_json::json!({
+                "habit_id": item_id, "context": "", "outcome": outcome,
+                "activated_at": ts,
+            })),
+        }
+    } else {
+        let (confidence, new_count, new_state) = correction_reinforce(
+            num(row, "confidence", 0.5),
+            num(row, "evidence_count", 1.0) as i64,
+            outcome,
+        );
+        ReinforcePlan {
+            result: serde_json::json!({
+                "reinforced": item_id, "confidence": wire_num(confidence),
+                "state": new_state.as_str(),
+            }),
+            writeback: serde_json::json!({
+                "confidence": confidence, "evidence_count": new_count,
+                "state": new_state.as_str(), "last_reinforced": ts,
+            }),
+            activation: None,
+        }
+    }
+}
+
+/// The bh-core delegation half of reinforce (name-preserving port of
+/// `_reinforce_rust`): the fetched row and the pinned timestamp go in,
+/// the return dict + UPDATE/INSERT payloads come out. Raises-through
+/// becomes data: a missing row yields the not-found envelope.
+pub fn _reinforce_rust(
+    item_id: i64,
+    outcome: &str,
+    item_type: &str,
+    row: Option<&serde_json::Value>,
+    ts: &str,
+) -> ReinforcePlan {
+    reinforce_plan(item_id, outcome, item_type, row, ts)
+}
+
+/// The evidence_provenance shim half (name-preserving port of
+/// `_reinforce_python`). The oracle routes this path through the ep
+/// cdylib's habitmath — the kernels this crate's `habit_reinforce` /
+/// `correction_reinforce` are the ports of — so the shim half and the
+/// delegation half are one implementation here; the battery replays
+/// BOTH names over the frozen truth to pin their coincidence.
+pub fn _reinforce_python(
+    item_id: i64,
+    outcome: &str,
+    item_type: &str,
+    row: Option<&serde_json::Value>,
+    ts: &str,
+) -> ReinforcePlan {
+    reinforce_plan(item_id, outcome, item_type, row, ts)
+}
+
+// ── list_items + behavior_stats over row collections ─────────────────────
+
+/// List corrections and/or habits (name-preserving port of
+/// `list_items`): corrections WHERE state='active', habits WHERE state
+/// IN ('active','candidate'), both ORDER BY confidence DESC LIMIT n.
+/// Ties keep fetch order — sqlite yields rowid order for equal keys,
+/// which a stable sort reproduces. Result keys follow item_type
+/// ('all' → both, 'correction'/'habit' → that one only).
+pub fn list_items(
+    corrections: &[CorrectionRow],
+    habits: &[HabitRow],
+    item_type: &str,
+    limit: i64,
+) -> serde_json::Value {
+    let limit = limit.max(0) as usize;
+    let mut out = serde_json::Map::new();
+    if item_type == "all" || item_type == "correction" {
+        let mut rows: Vec<&CorrectionRow> =
+            corrections.iter().filter(|c| c.state == "active").collect();
+        rows.sort_by(|a, b| b.confidence.total_cmp(&a.confidence));
+        let rows: Vec<&CorrectionRow> = rows.into_iter().take(limit).collect();
+        out.insert(
+            "corrections".into(),
+            serde_json::to_value(rows).unwrap_or(serde_json::Value::Null),
+        );
+    }
+    if item_type == "all" || item_type == "habit" {
+        let mut rows: Vec<&HabitRow> = habits
+            .iter()
+            .filter(|h| h.state == "active" || h.state == "candidate")
+            .collect();
+        rows.sort_by(|a, b| b.confidence.total_cmp(&a.confidence));
+        let rows: Vec<&HabitRow> = rows.into_iter().take(limit).collect();
+        out.insert(
+            "habits".into(),
+            serde_json::to_value(rows).unwrap_or(serde_json::Value::Null),
+        );
+    }
+    serde_json::Value::Object(out)
+}
+
+/// Return behavioral pattern statistics (name-preserving port of
+/// `behavior_stats`): the five counters over the row collections; the
+/// activation count is passed in (its rows aren't decision input).
+pub fn behavior_stats(corrections: &[CorrectionRow], habits: &[HabitRow], activations: i64) -> serde_json::Value {
+    serde_json::json!({
+        "corrections_active":
+            corrections.iter().filter(|c| c.state == "active").count() as i64,
+        "corrections_archived":
+            corrections.iter().filter(|c| c.state == "archived").count() as i64,
+        "habits_active":
+            habits.iter().filter(|h| h.state == "active").count() as i64,
+        "habits_candidate":
+            habits.iter().filter(|h| h.state == "candidate").count() as i64,
+        "total_activations": activations,
+    })
+}
+
 // ── path + plugin data ──────────────────────────────────────────────────
 
 /// behavior.db path resolution (mirrors memory_env: RUNE_MEMORY_DIR
@@ -404,6 +747,105 @@ pub extern "C" fn bh_plugin() -> *mut std::os::raw::c_char {
     envelope(Ok(plugin()))
 }
 
+/// bh_connect() -> {path, journal_mode, schema: [{kind, name, sql}]}.
+#[no_mangle]
+pub extern "C" fn bh_connect() -> *mut std::os::raw::c_char {
+    let spec = connect();
+    let schema: Vec<serde_json::Value> = spec
+        .schema
+        .iter()
+        .map(|s| serde_json::json!({"kind": s.kind, "name": s.name, "sql": s.sql}))
+        .collect();
+    envelope(Ok(serde_json::json!({
+        "path": spec.path,
+        "journal_mode": spec.journal_mode,
+        "schema": schema,
+    })))
+}
+
+/// bh_core_probe() -> {"available": bool} — the `_bh_core` pin honored.
+#[no_mangle]
+pub extern "C" fn bh_core_probe() -> *mut std::os::raw::c_char {
+    envelope(Ok(serde_json::json!({"available": _bh_core().is_some()})))
+}
+
+/// bh_python_predicates({kind, row, context, context_low?}) -> bool.
+/// kind: "correction" (context_low defaults to context.lower()) | "habit".
+#[no_mangle]
+pub extern "C" fn bh_python_predicates(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let kind = str_field(&a, "kind");
+        let row_v = a.get("row").cloned().unwrap_or(serde_json::json!({}));
+        let context = str_field(&a, "context");
+        if kind == "habit" {
+            let row: HabitRow = serde_json::from_value(row_v).map_err(|e| e.to_string())?;
+            Ok(serde_json::json!(_habit_matches_python(&row, &context)))
+        } else {
+            let context_low = match a.get("context_low") {
+                Some(serde_json::Value::String(s)) => s.clone(),
+                _ => context.to_lowercase(),
+            };
+            let row: CorrectionRow = serde_json::from_value(row_v).map_err(|e| e.to_string())?;
+            Ok(serde_json::json!(_correction_matches_python(&row, &context, &context_low)))
+        }
+    }))
+}
+
+/// bh_reinforce_plan({item_id, outcome, item_type, row?, ts, path?}) ->
+/// the ReinforcePlan (path "python" routes `_reinforce_python`, else
+/// `_reinforce_rust`).
+#[no_mangle]
+pub extern "C" fn bh_reinforce_plan(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let item_id = num(&a, "item_id", 0.0) as i64;
+        let outcome = str_field(&a, "outcome");
+        let item_type = str_field(&a, "item_type");
+        let ts = str_field(&a, "ts");
+        let row = a.get("row").filter(|v| !v.is_null()).cloned();
+        let plan = if str_field(&a, "path") == "python" {
+            _reinforce_python(item_id, &outcome, &item_type, row.as_ref(), &ts)
+        } else {
+            _reinforce_rust(item_id, &outcome, &item_type, row.as_ref(), &ts)
+        };
+        serde_json::to_value(plan).map_err(|e| e.to_string())
+    }))
+}
+
+/// bh_list_items({corrections, habits, item_type, limit}) -> the result.
+#[no_mangle]
+pub extern "C" fn bh_list_items(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let corrections: Vec<CorrectionRow> = serde_json::from_value(
+            a.get("corrections").cloned().unwrap_or(serde_json::json!([])),
+        )
+        .map_err(|e| e.to_string())?;
+        let habits: Vec<HabitRow> = serde_json::from_value(
+            a.get("habits").cloned().unwrap_or(serde_json::json!([])),
+        )
+        .map_err(|e| e.to_string())?;
+        let item_type = str_field(&a, "item_type");
+        let limit = num(&a, "limit", 20.0) as i64;
+        Ok(list_items(&corrections, &habits, &item_type, limit))
+    }))
+}
+
+/// bh_behavior_stats({corrections, habits, activations}) -> the stats.
+#[no_mangle]
+pub extern "C" fn bh_behavior_stats(arg: *const std::os::raw::c_char) -> *mut std::os::raw::c_char {
+    envelope(read_arg(arg).and_then(|a| {
+        let corrections: Vec<CorrectionRow> = serde_json::from_value(
+            a.get("corrections").cloned().unwrap_or(serde_json::json!([])),
+        )
+        .map_err(|e| e.to_string())?;
+        let habits: Vec<HabitRow> = serde_json::from_value(
+            a.get("habits").cloned().unwrap_or(serde_json::json!([])),
+        )
+        .map_err(|e| e.to_string())?;
+        let activations = num(&a, "activations", 0.0) as i64;
+        Ok(behavior_stats(&corrections, &habits, activations))
+    }))
+}
+
 #[cfg(test)]
 #[path = "branch_battery.rs"]
 mod branch_battery;

verification/behavior-cargo-test.exit

@@ -0,0 +1 @@
+0

verification/behavior-clippy.exit

@@ -0,0 +1 @@
+0

verification/behavior-freeze-verify.exit

@@ -0,0 +1 @@
+0

verification/behavior-live-parity.exit

@@ -0,0 +1 @@
+0

verification/behavior-pytest-live.exit

@@ -0,0 +1 @@
+0

verification/behavior-pytest-python.exit

@@ -0,0 +1 @@
+0