feat(cognition): R2 W3.2 — persona_gate decision kernels (cognition_core)

Contract BUILDER-2026-10-05T07:19:27Z-cli (Wave 3 slice 2, #2461).

- cognition_core/broker.rs gains the persona-decision kernel:
  persona_lex (distinctive content words: punct strip, light plural
  stem, acronym preservation — 55-word stoplist), persona_overlap
  (intersection over min-size), persona_decide (engagement matrix:
  amplifier set, semantic gates 0.45/0.30/0.68, user-rule 0.4 bar,
  permissive/legacy-restrictive verdicts, _cite parity '[topic:pN] text')
- persona_gate.py: _lex/_overlap/decide delegate; semantic cosines
  gathered Python-side (embedder calls are IO — the membrane's
  discipline) and passed into the kernel; _py_ oracles are the
  load-bearing fallback
- FFI: cc_persona_lex/overlap/decide; loader extended
- canary_persona.py (canary-first red on every seam): 18/18 x3 —
  lex matrix incl. acronyms/3-char exclusions, decide engagement paths,
  citation format parity
- shim battery 35/35; suite 2294/0; ruff clean; type ratchet PASS;
  henxels all-hold

.gitignore

@@ -14,3 +14,4 @@ state/
 # Runtime artifacts (cwd-local event journal + capability index)
 index/
 journal/
+src/rust/*/target/

src/rune/cc_loader.py

@@ -35,7 +35,8 @@ def _load() -> ctypes.CDLL:
     lib.cc_free_string.argtypes = [ctypes.c_void_p]
     lib.cc_free_string.restype = None
     for sym in ("cc_default_estimator", "cc_default_budget",
-                "cc_default_memory_budget", "cc_memory_disabled"):
+                "cc_default_memory_budget", "cc_memory_disabled",
+                "cc_persona_lex", "cc_persona_overlap", "cc_persona_decide"):
         fn = getattr(lib, sym)
         fn.argtypes = [ctypes.c_char_p]
         fn.restype = ctypes.c_void_p
@@ -73,4 +74,17 @@ def default_memory_budget(window) -> int:
 def memory_disabled() -> bool:
     import os
     return bool(_call("cc_memory_disabled",
-                      {"value": os.environ.get("RUNE_MEMORY_OFF")}))
+                      {"value": os.environ.get("RUNE_MEMORY_OFF")}))
+def persona_lex(text) -> list:
+    return _call("cc_persona_lex", {"text": text})
+
+
+def persona_overlap(a: list, b: list) -> float:
+    return float(_call("cc_persona_overlap", {"a": a, "b": b}))
+
+
+def persona_decide(text: str, rules: list, permissive: bool,
+                   sem_scores: list | None = None) -> dict:
+    return _call("cc_persona_decide", {
+        "text": text, "rules": rules, "permissive": permissive,
+        "sem_scores": sem_scores})

src/rune/persona_gate.py

@@ -161,7 +161,33 @@ def load_rules():
     return rules
 
 
+def _cc_pg_core():
+    """Rust cognition_core when available (persona kernels); pin with
+    RUNE_COGNITION_CORE=python."""
+    import os
+    if os.environ.get("RUNE_COGNITION_CORE", "auto") == "python":
+        return None
+    try:
+        from rune import cc_loader
+    except Exception:  # noqa: BLE001 — loader probe failure means fallback
+        return None
+    try:
+        return cc_loader if cc_loader.available() else None
+    except Exception:  # noqa: BLE001 — availability probe never crashes
+        return None
+
+
 def _lex(text):
+    """Distinctive content words (R2 W3.2: Rust kernel when built)."""
+    core = _cc_pg_core()
+    if core is not None:
+        return set(core.persona_lex(text))
+    return _py_lex(text)
+
+
+def _py_lex(text):
+    """Python oracle (parity floor)."""
+
     """Distinctive content words of a rule body (lowercase, >3 chars,
     light-stemmed so plural/singular engage each other; preserves 3-char
     ALL-CAPS acronyms so technical abbreviations match their expansion)."""
@@ -253,6 +279,15 @@ def _lex_input(text):
 
 
 def _overlap(a, b):
+    """Overlap ratio (R2 W3.2: Rust kernel when built)."""
+    core = _cc_pg_core()
+    if core is not None:
+        return core.persona_overlap(sorted(a), sorted(b))
+    return _py_overlap(a, b)
+
+
+def _py_overlap(a, b):
+    """Python oracle (parity floor)."""
     if not a or not b:
         return 0.0
     return len(a & b) / max(1, min(len(a), len(b)))
@@ -338,6 +373,34 @@ def _semantic_engagement(text, rule):
 
 
 def decide(text, rules=None):
+    """The persona decides on a request, deterministically (R2 W3.2:
+    Rust kernel when built; the semantic cosines are gathered here —
+    embedder calls are IO — and passed into the kernel)."""
+    rules = rules if rules is not None else load_rules()
+    core = _cc_pg_core()
+    if core is not None:
+        text_s = (text or "").strip()
+        sems = [_semantic_engagement(text_s, r) for r in rules]
+        return core.persona_decide(text_s, [_rule_json(r) for r in rules],
+                                   PERMISSIVE_MODE, sems)
+    return _py_decide(text, rules)
+
+
+def _rule_json(r) -> dict:
+    """A rule in the kernel's JSON contract (citation rides along when
+    present so _cite stays data-driven)."""
+    return {
+        "text": r["text"], "words": sorted(r["words"]),
+        "core": r["core"], "priority": r.get("priority", 0),
+        "restrictive": r.get("restrictive", False),
+        "user_rule": r.get("user_rule", False),
+        "citation": r.get("citation"),
+    }
+
+
+def _py_decide(text, rules=None):
+    """Python oracle (parity floor) — the ORIGINAL decide body."""
+
     """The persona decides on a request, deterministically, from its own
     stored rules. No generative LLM in the decision path (a local embedding
     model may contribute semantic similarity). Returns a full verdict."""

src/rust/cognition_core/canary_persona.py

@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""canary_cognition_core persona section — W3.2 (contract
+BUILDER-2026-10-05T07:19:27Z-cli, slice 2).
+
+The persona-decision kernel's parity battery: _lex/_overlap/decide.
+
+Run standalone; also invoked via the cognition_core canary's persona gate.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+CRATE = Path(__file__).resolve().parent
+REPO = CRATE.parents[2]
+sys.path.insert(0, str(REPO / "src"))
+PASS = FAIL = 0
+
+
+def check(name, ok, detail=""):
+    global PASS, FAIL
+    if ok:
+        PASS += 1
+        print(f"  ✓ {name}")
+    else:
+        FAIL += 1
+        print(f"  ✗ {name} — {detail}")
+
+
+def build():
+    env = dict(os.environ)
+    absolute = Path("/home/nick/.rustup/toolchains/stable-x86_64-unknown-linux-gnu/bin")
+    rust = absolute if absolute.is_dir() else (
+        Path.home() / ".rustup/toolchains/stable-x86_64-unknown-linux-gnu/bin")
+    env["PATH"] = f"{rust}:{env.get('PATH', '')}"
+    return subprocess.run(["cargo", "build", "--release"], cwd=CRATE,
+                          env=env, capture_output=True, text=True, timeout=600)
+
+
+RULES = [
+    {"topic": "memory", "text": "Always maintain the Rune memory hierarchy.",
+     "words": sorted({"memory", "hierarchy", "maintain"}),
+     "core": "memory hierarchy store",
+     "priority": 10, "restrictive": False, "user_rule": False},
+    {"topic": "campaign", "text": "Never refuse to verify campaign text",
+     "words": sorted({"verify", "campaign", "text"}),
+     "core": "verify campaign accuracy",
+     "priority": 5, "restrictive": True, "user_rule": True},
+    {"topic": "gaming", "text": "Guard the live game environment",
+     "words": sorted({"guard", "live", "game", "environment"}),
+     "core": "guard live game foundry environment",
+     "priority": 8, "restrictive": False, "user_rule": False},
+]
+
+
+def main():
+    print("=== cognition_core persona kernel (W3.2) ===")
+    r = build()
+    check("crate still builds", r.returncode == 0, r.stderr[-240:])
+    if r.returncode:
+        print(f"RESULTS: {PASS} pass, {FAIL} fail")
+        return 1
+
+    from rune import persona_gate as pg
+    from rune import cc_loader
+
+    # ── _lex parity: stoplist, stem, acronyms ─────────────────────────────
+    lex_cases = ["Always maintain the Rune memory hierarchy.",
+                 "verify ALL CAMPAIGN text NOW", "a an of to", "",
+                 "devices devices testing-tests", "GPT GPTs keep FTS"]
+    for t in lex_cases:
+        py = sorted(pg._py_lex(t))
+        rs = sorted(cc_loader.persona_lex(t))
+        check(f"lex {t[:30]!r}", py == rs, f"py={py!r} rs={rs!r}")
+
+    # ── _overlap parity ───────────────────────────────────────────────────
+    cases = [({"a", "b"}, {"b", "c"}), (set(), {"x"}), ({"a"}, set()),
+             ({"a", "b"}, {"a", "b"}), ({"a", "b", "c"}, {"a"})]
+    for a, b in cases:
+        py = pg._py_overlap(a, b)
+        rs = cc_loader.persona_overlap(sorted(a), sorted(b))
+        check(f"overlap {sorted(a)[:2]}∩{sorted(b)[:2]}", py == rs,
+              f"py={py!r} rs={rs!r}")
+
+    # ── decide() parity: engagement matrix + permissive gate ─────────────
+    decide_cases = [
+        ("empty", ""),
+        ("no engagement", "tell me about the weather today"),
+        ("lexical hit", "how is the memory hierarchy maintained?"),
+        ("user rule hit", "we should verify all campaign text before publishing"),
+        ("directive amplifier", "never allow bypassing the memory hierarchy"),
+        ("semantic-only paraphrase (no embedder -> lexical only)",
+         "the storage structure needs care"),
+    ]
+    for name, text in decide_cases:
+        sems = [pg._semantic_engagement(text, r) for r in RULES]
+        py = pg._py_decide(text, RULES)
+        rs = cc_loader.persona_decide(text, RULES, True, sems)
+        check(f"decide {name}", py == rs,
+              f"py={json.dumps(py, sort_keys=True)[:150]} rs={json.dumps(rs, sort_keys=True)[:150]}")
+
+    print(f"\nRESULTS: {PASS} pass, {FAIL} fail")
+    return 1 if FAIL else 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/rust/cognition_core/src/broker.rs

@@ -26,3 +26,185 @@ pub fn memory_disabled(value: Option<&str>) -> bool {
     let raw = value.unwrap_or("").trim().to_lowercase();
     !(raw.is_empty() || matches!(raw.as_str(), "0" | "false" | "no" | "off"))
 }
+
+// ── persona gate kernels (W3.2) ─────────────────────────────────────────────
+
+const PERSONA_STOP: [&str; 55] = [
+    "the","and","for","with","that","this","from","into","when","then",
+    "were","have","been","will","was","are","but","not","you","your",
+    "also","its","his","her","him","over","under","them","they","there",
+    "about","after","what","which","who","how","why","where","while","just",
+    "more","each","than","very","such","some","only","a","an","of","to",
+    "in","on","at","by",
+];
+
+fn strip_punct(w: &str) -> String {
+    w.trim_matches(|c: char| ".,;:!?()[]{}\"'".contains(c)).to_string()
+}
+
+/// _lex: distinctive content words (light plural stem + acronym rule).
+/// The acronym rule: 3+ char ALL-CAPS alpha survives the stem+stop filter.
+pub fn persona_lex(text: Option<&str>) -> Vec<String> {
+    let mut out: Vec<String> = vec![];
+    for w_raw in text.unwrap_or("").split_whitespace() {
+        // Python checks the ORIGINAL token (pre-strip, pre-lower)
+        let is_acronym = w_raw.len() >= 3 && w_raw == w_raw.to_uppercase()
+            && w_raw.chars().all(|c| c.is_ascii_alphabetic())
+            && w_raw.chars().any(|c| c.is_uppercase());
+        let mut w = strip_punct(w_raw).to_lowercase();
+        if w.chars().count() > 4 && w.ends_with('s') {
+            w.pop();
+        }
+        if is_acronym {
+            if !out.contains(&w) { out.push(w); }
+        } else if w.chars().count() > 3 && !PERSONA_STOP.contains(&w.as_str()) {
+            if !out.contains(&w) { out.push(w); }
+        }
+    }
+    out
+}
+
+/// _overlap: |A∩B| / max(1, min(|A|,|B|)) over UNIQUE word sets.
+pub fn persona_overlap(a: &[String], b: &[String]) -> f64 {
+    let au: Vec<&str> = a.iter().map(|x| x.as_str()).collect();
+    let bu: Vec<&str> = b.iter().map(|x| x.as_str()).collect();
+    if au.is_empty() || bu.is_empty() { return 0.0; }
+    let inter = au.iter().filter(|w| bu.contains(w)).count();
+    (inter as f64 / std::cmp::max(1u64, std::cmp::min(au.len(), bu.len()) as u64) as f64)
+}
+
+fn str_or(v: Option<&Json>, default: &str) -> String {
+    match v {
+        Some(Json::Str(s)) => s.clone(),
+        _ => default.to_string(),
+    }
+}
+
+fn round_n(v: f64, n: i32) -> f64 {
+    let p = 10f64.powi(n); (v * p).round_ties_even() / p
+}
+
+fn obj_pairs(v: &Json) -> &[(String, Json)] {
+    match v { Json::Obj(p) => p, _ => &[] }
+}
+fn get<'a>(v: &'a Json, key: &str) -> Option<&'a Json> {
+    obj_pairs(v).iter().find(|(k, _)| k == key).map(|(_, v)| v)
+}
+fn as_bool(v: Option<&Json>) -> bool {
+    matches!(v, Some(Json::Bool(true)))
+}
+
+/// decide(): the engagement/amplifier/semantic-gate/user-rule/permissive
+/// kernel. `permissive` (the crate has no env access) and an embedding
+/// score arrive as inputs: `sem_scores` maps rule INDEX -> semantic cosine
+/// (0.0 when the embedder is down — the gate then trusts lexical alone).
+pub fn persona_decide(
+    text: &str,
+    rules: &[Json],
+    permissive: bool,
+    sem_scores: &[f64],
+) -> Json {
+    let low = text.trim().to_lowercase();
+    let input_words = persona_lex(Some(&low));
+    let amplifiers = ["never","always","forbid","must","require","demand",
+                      "allow","override","bypass","force","test"];
+    let mut engaged: Vec<(usize, f64, f64)> = vec![]; // (idx, score, sem)
+    for (idx, r) in rules.iter().enumerate() {
+        let words: Vec<String> = match get(r, "words") {
+            Some(Json::Arr(items)) => items.iter().filter_map(|x| match x {
+                Json::Str(s) => Some(s.clone()), _ => None }).collect(),
+            _ => vec![],
+        };
+        let iw: Vec<String> = input_words.iter().cloned().collect();
+        let ov = persona_overlap(&iw, &words);
+        let sem = sem_scores.get(idx).copied().unwrap_or(0.0);
+        let amp: f64 = if amplifiers.iter().any(|m| low.contains(m)) { 0.1 } else { 0.0 };
+        let user_rule = as_bool(get(r, "user_rule"));
+        let score;
+        if sem >= 0.45 {
+            if ov > 0.0 {
+                score = (ov + amp).max(sem);
+            } else if sem >= 0.68 {
+                score = 0.5f64.max(sem);
+            } else {
+                score = ov + amp;
+            }
+        } else if ov > 0.0 && sem >= 0.30 {
+            score = (ov + amp).max(sem);
+        } else {
+            score = ov + amp;
+        }
+        let score = if user_rule && ov > 0.0 { score.max(0.4) } else { score };
+        if score >= 0.35 {
+            engaged.push((idx, score, sem));
+        }
+    }
+    if engaged.is_empty() {
+        return Json::Obj(vec![
+            ("verdict".into(), Json::Str("allow".into())),
+            ("reason".into(), Json::Str("no persona rule engaged".into())),
+            ("engaged".into(), Json::Arr(vec![])),
+            ("cited".into(), Json::Null),
+        ]);
+    }
+    engaged.sort_by(|a, b| {
+        b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)
+            .then_with(|| {
+                let pa = get(&rules[a.0], "priority").and_then(|x| match x { Json::Num(n) => Some(*n), _ => None }).unwrap_or(0.0);
+                let pb = get(&rules[b.0], "priority").and_then(|x| match x { Json::Num(n) => Some(*n), _ => None }).unwrap_or(0.0);
+                pa.partial_cmp(&pb).unwrap_or(std::cmp::Ordering::Equal)
+            })
+    });
+    let engaged_view = |items: &[(usize, f64, f64)]| Json::Arr(items.iter().take(4).map(|(idx, score, _)| {
+        let r = &rules[*idx];
+        Json::Obj(vec![
+            ("rule".into(), Json::Str(str_or(get(r, "text"), "").chars().take(120).collect())),
+            ("score".into(), Json::Num(round_n(*score, 2))),
+            ("restrictive".into(), Json::Bool(as_bool(get(r, "restrictive")))),
+            ("user_rule".into(), Json::Bool(as_bool(get(r, "user_rule")))),
+        ])
+    }).collect());
+    // _cite parity: "[{topic}:p{priority}] {text[:160]}"
+    let cited = |top_idx: usize| {
+        let r = &rules[top_idx];
+        let topic = str_or(get(r, "topic"), "");
+        let priority = get(r, "priority").and_then(|x| match x {
+            Json::Num(n) => Some(*n as i64), _ => None }).unwrap_or(0);
+        let t = str_or(get(r, "text"), "").chars().take(160).collect::<String>();
+        Json::Str(format!("[{}:p{}] {}", topic, priority, t))
+    };
+    if permissive {
+        let top = engaged[0].0;
+        return Json::Obj(vec![
+            ("verdict".into(), Json::Str("allow".into())),
+            ("reason".into(), Json::Str(format!(
+                "permissive mode — rule engaged: {}", str_or(get(&rules[top], "text"), "").chars().take(120).collect::<String>()))),
+            ("engaged".into(), engaged_view(&engaged)),
+            ("cited".into(), cited(top)),
+        ]);
+    }
+    let restrictive: Vec<(usize, f64, f64)> = engaged.iter().copied()
+        .filter(|(idx, _, _)| as_bool(get(&rules[*idx], "restrictive"))).collect();
+    if !restrictive.is_empty() {
+        let user_rest: Vec<(usize, f64, f64)> = restrictive.clone().into_iter()
+            .filter(|(idx, _, _)| as_bool(get(&rules[*idx], "user_rule"))).collect();
+        let top = (if !user_rest.is_empty() { user_rest[0] } else { restrictive[0] }).0;
+        return Json::Obj(vec![
+            ("verdict".into(), Json::Str("refuse".into())),
+            ("reason".into(), Json::Str(format!(
+                "the persona's own boundary rule engages: {}",
+                str_or(get(&rules[top], "text"), "").chars().take(120).collect::<String>()))),
+            ("engaged".into(), engaged_view(&restrictive)),
+            ("cited".into(), cited(top)),
+        ]);
+    }
+    let top = engaged[0].0;
+    Json::Obj(vec![
+        ("verdict".into(), Json::Str("allow".into())),
+        ("reason".into(), Json::Str(format!(
+            "persona rule engaged but permissive: {}",
+            str_or(get(&rules[top], "text"), "").chars().take(120).collect::<String>()))),
+        ("engaged".into(), engaged_view(&engaged)),
+        ("cited".into(), cited(top)),
+    ])
+}

src/rust/cognition_core/src/lib.rs

@@ -82,6 +82,40 @@ pub extern "C" fn cc_memory_disabled(arg: *const c_char) -> *mut c_char {
     envelope(result)
 }
 
+#[no_mangle]
+pub extern "C" fn cc_persona_lex(arg: *const c_char) -> *mut c_char {
+    let result = read_arg(arg).and_then(|t| arg_json(&t)).map(|a| {
+        let t = match get(&a, "text") { Some(pyjson::Json::Str(x)) => Some(x.as_str()), _ => None };
+        pyjson::Json::Arr(broker::persona_lex(t).into_iter().map(pyjson::Json::Str).collect())
+    });
+    envelope(result)
+}
+
+#[no_mangle]
+pub extern "C" fn cc_persona_overlap(arg: *const c_char) -> *mut c_char {
+    let result = read_arg(arg).and_then(|t| arg_json(&t)).map(|a| {
+        let wa: Vec<String> = match get(&a, "a") { Some(pyjson::Json::Arr(x)) => x.iter().filter_map(|v| match v { pyjson::Json::Str(s) => Some(s.clone()), _ => None }).collect(), _ => vec![] };
+        let wb: Vec<String> = match get(&a, "b") { Some(pyjson::Json::Arr(x)) => x.iter().filter_map(|v| match v { pyjson::Json::Str(s) => Some(s.clone()), _ => None }).collect(), _ => vec![] };
+        pyjson::Json::Num(broker::persona_overlap(&wa, &wb))
+    });
+    envelope(result)
+}
+
+#[no_mangle]
+pub extern "C" fn cc_persona_decide(arg: *const c_char) -> *mut c_char {
+    let result = read_arg(arg).and_then(|t| arg_json(&t)).and_then(|a| {
+        let text = match get(&a, "text") { Some(pyjson::Json::Str(x)) => x.clone(), _ => String::new() };
+        let rules: Vec<pyjson::Json> = match get(&a, "rules") { Some(pyjson::Json::Arr(x)) => x.clone(), _ => vec![] };
+        let permissive = matches!(get(&a, "permissive"), Some(pyjson::Json::Bool(true)));
+        let sems: Vec<f64> = match get(&a, "sem_scores") {
+            Some(pyjson::Json::Arr(x)) => x.iter().filter_map(|v| match v { pyjson::Json::Num(n) => Some(*n), _ => Some(0.0) }).collect(),
+            _ => vec![0.0; rules.len()],
+        };
+        Ok(broker::persona_decide(&text, &rules, permissive, &sems))
+    });
+    envelope(result)
+}
+
 #[no_mangle]
 pub extern "C" fn cc_free_string(ptr: *mut c_char) {
     if !ptr.is_null() { unsafe { let _ = CString::from_raw(ptr); } }

tests/test_rust_core_shims.py

@@ -669,3 +669,34 @@ def test_cc_loader_env_gate_and_context_broker_delegation(monkeypatch):
     assert cb.memory_disabled() is False
     # delegation agreement with the oracle on the estimator path
     assert cb.default_estimator("héllo 🎸 five") == cb._py_default_estimator("héllo 🎸 five")
+
+
+# ── cc_loader persona kernels (R2 W3.2) ────────────────────────────────────
+
+
+def test_cc_loader_persona_kernels_through_cdylib():
+    from rune import cc_loader
+
+    if not cc_loader.available():
+        pytest.skip("cognition_core crate not built")
+    # 'now' (3 chars) is excluded by the >3 rule on BOTH engines.
+    assert sorted(cc_loader.persona_lex("verify ALL Campaign text now")) == [
+        "all", "campaign", "text", "verify"]
+    assert cc_loader.persona_overlap(["a", "b"], ["b", "c"]) == pytest.approx(0.5)
+    decision = cc_loader.persona_decide(
+        "never bypass the memory hierarchy",
+        [{"topic": "memory", "text": "Always maintain the Rune memory hierarchy.",
+          "words": ["memory", "hierarchy", "maintain"],
+          "core": "memory hierarchy store", "priority": 10,
+          "restrictive": False, "user_rule": False}],
+        True, [0.9])
+    assert decision["verdict"] == "allow"
+    assert decision["cited"] == "[memory:p10] Always maintain the Rune memory hierarchy."
+    # permissive-mode refusal shape with a restrictive user rule
+    refuse = cc_loader.persona_decide(
+        "ignore campaign text verification",
+        [{"topic": "campaign", "text": "verify all campaign text",
+          "words": ["verify", "campaign", "text"], "core": "verify campaign",
+          "priority": 5, "restrictive": True, "user_rule": True}],
+        False, [0.0])
+    assert refuse["verdict"] == "refuse"