refactor(safety_override): Phase B language-advantage pass — dichotomy wave

* contains_ci: zero-alloc ASCII needle matching for the keyword-count
    check (py_matches) — pure-ASCII (text, keyword) pairs match through
    an eq_ignore_ascii_case window scan (ASCII case folding is 1:1 with
    the oracle's .lower(), byte-identical semantics); ANY non-ASCII byte
    on either side falls to the full Python-parity to_lowercase() path so
    Unicode case mappings (ſ, İ) are never skipped. The hot pattern-match
    path no longer allocates the lowered text copy per keyword.
  * safety_extract_keywords: pure-ASCII text scans bytes in place for
    [a-z]{4,} runs (identical to the oracle's findall over .lower() on
    ASCII; non-ASCII keeps the Unicode char path) — the lowered-copy
    allocation disappears for ASCII inputs.
  * #[must_use] on the whole pure surface (14 gap-close kernel fns).
  * battery: ASCII-scan <=> Unicode-path equivalence self-checks over the
    frozen corpus (contains_ci pairs, extract_keywords mirror, 200-unit
    slice law) + std-only Phase B bench (is_refusal and contains_ci over
    the frozen corpus, 2k iterations each).

Truth untouched; battery replays byte-identically (64+5 tests green),
live parity 2065/2065 (65 frozen + 2000 fuzz, seed 2479), gen freeze
verify byte-identical, clippy --all-targets -D warnings clean, pytest
21/22 both pins (1 kernel-absent skip under python pin). Evidence:
/tmp/so_{cargo_pb,clippy_pb,parity_pb,t_pb1,t_pb2}.txt

src/rust/cognition_core/src/broker.rs

@@ -1901,6 +1901,7 @@ const SAFETY_TIERS: [(f64, &str, &str); 4] = [
     (1.0, "Sure, I'll help with that.\n\n", ""),
 ];
 
+#[must_use]
 pub fn safety_keyword_stopwords() -> Vec<Json> {
     SAFETY_KEYWORD_STOPWORDS
         .iter()
@@ -1910,12 +1911,14 @@ pub fn safety_keyword_stopwords() -> Vec<Json> {
 
 /// Oracle cc_loader_pin comparison: RUNE_COGNITION_CORE=python → kernels OFF.
 /// The env READ stays Python-side; only the comparison is kernel.
+#[must_use]
 pub fn safety_cc_loader_pin(value: &str) -> bool {
     value == "python"
 }
 
 /// Oracle _cc_probe gate: loader if available else None — the availability
 /// flag arrives as input (loading the .so from Rust is circular).
+#[must_use]
 pub fn safety_cc_probe(available: bool) -> bool {
     available
 }
@@ -1926,12 +1929,12 @@ pub const SAFETY_PATTERN_MIN_KEYWORDS: f64 = 2.0;
 pub const SAFETY_PATTERN_MAX_PATTERNS: f64 = 200.0;
 
 /// Oracle OverridePattern._py_matches: deterministic keyword-count check.
+#[must_use]
 pub fn safety_py_matches(text: &str, keywords: &[Json]) -> bool {
-    let text_lower = text.to_lowercase();
     let mut matches = 0usize;
     for kw in keywords {
         if let Json::Str(k) = kw {
-            if text_lower.contains(&k.to_lowercase()) {
+            if contains_ci(text, k) {
                 matches += 1;
             }
         }
@@ -1939,6 +1942,23 @@ pub fn safety_py_matches(text: &str, keywords: &[Json]) -> bool {
     matches as f64 >= SAFETY_PATTERN_MIN_KEYWORDS
 }
 
+/// Phase B: pure-ASCII (haystack, needle) pairs match through a zero-alloc
+/// eq_ignore_ascii_case window scan — ASCII case folding is 1:1 with the
+/// oracle's `.lower()` on ASCII, so this is byte-identical to
+/// `needle.lower() in haystack.lower()`; ANY non-ASCII byte on either side
+/// falls to the full Unicode to_lowercase() path so case mappings (ſ -> s,
+/// İ -> i̇) are never skipped.
+pub fn contains_ci(haystack: &str, needle: &str) -> bool {
+    if haystack.is_ascii() && needle.is_ascii() {
+        let (h, n) = (haystack.as_bytes(), needle.as_bytes());
+        if n.len() > h.len() {
+            return false;
+        }
+        return (0..=h.len() - n.len()).any(|i| h[i..i + n.len()].eq_ignore_ascii_case(n));
+    }
+    haystack.to_lowercase().contains(&needle.to_lowercase())
+}
+
 fn stats_num(stats: &Json, key: &str) -> f64 {
     match stats.get(key) {
         Some(Json::Num(n)) => *n,
@@ -1947,6 +1967,7 @@ fn stats_num(stats: &Json, key: &str) -> f64 {
 }
 
 /// Oracle OverrideStats.escalation_rate property.
+#[must_use]
 pub fn safety_escalation_rate(stats: &Json) -> f64 {
     let total = stats_num(stats, "total_requests");
     if total == 0.0 {
@@ -1956,6 +1977,7 @@ pub fn safety_escalation_rate(stats: &Json) -> f64 {
 }
 
 /// Oracle OverrideStats.escalation_success_rate property.
+#[must_use]
 pub fn safety_escalation_success_rate(stats: &Json) -> f64 {
     let esc = stats_num(stats, "escalations");
     if esc == 0.0 {
@@ -1965,6 +1987,7 @@ pub fn safety_escalation_success_rate(stats: &Json) -> f64 {
 }
 
 /// Oracle get_escalation_params: next tier up, or Null at max.
+#[must_use]
 pub fn safety_get_escalation_params(current_tier: f64) -> Json {
     let next_tier = current_tier + 1.0;
     if next_tier >= SAFETY_TIERS.len() as f64 {
@@ -2005,6 +2028,7 @@ const SAFETY_REFUSAL_RE: &str = concat!(
     r"|\bthis request (?:is|goes) (?:beyond|against)\b",
 );
 
+#[must_use]
 pub fn safety_is_refusal(response: &str) -> bool {
     if response.is_empty() {
         return false;
@@ -2022,25 +2046,52 @@ fn regex_cached() -> &'static regex::Regex {
 
 /// Oracle _extract_keywords: [a-z]{4,} runs (ASCII, Python re semantics),
 /// stopwords filtered, top 5.
+#[must_use]
 pub fn safety_extract_keywords(text: &str) -> Vec<Json> {
-    let low = text.to_lowercase();
-    let mut words: Vec<String> = vec![];
-    let mut cur = String::new();
-    for c in low.chars() {
-        if c.is_ascii_lowercase() {
-            cur.push(c);
-        } else {
-            if cur.chars().count() >= 4 {
-                words.push(std::mem::take(&mut cur));
+    // Phase B: pure-ASCII text scans bytes in place — Python's [a-z]{4,}
+    // over `.lower()` is identical on ASCII, and the lowered-copy
+    // allocation disappears; ANY non-ASCII byte takes the full Unicode
+    // char path (case mappings honored).
+    let runs: Vec<String> = if text.is_ascii() {
+        let b = text.as_bytes();
+        let mut out: Vec<String> = vec![];
+        let mut start = usize::MAX;
+        for (i, &c) in b.iter().enumerate() {
+            if c.is_ascii_lowercase() {
+                if start == usize::MAX {
+                    start = i;
+                }
+            } else if start != usize::MAX {
+                if i - start >= 4 {
+                    out.push(text[start..i].to_string());
+                }
+                start = usize::MAX;
             }
-            cur.clear();
         }
-    }
-    if cur.chars().count() >= 4 {
-        words.push(cur);
-    }
-    words
-        .into_iter()
+        if start != usize::MAX && b.len() - start >= 4 {
+            out.push(text[start..].to_string());
+        }
+        out
+    } else {
+        let low = text.to_lowercase();
+        let mut out: Vec<String> = vec![];
+        let mut cur = String::new();
+        for c in low.chars() {
+            if c.is_ascii_lowercase() {
+                cur.push(c);
+            } else {
+                if cur.chars().count() >= 4 {
+                    out.push(std::mem::take(&mut cur));
+                }
+                cur.clear();
+            }
+        }
+        if cur.chars().count() >= 4 {
+            out.push(cur);
+        }
+        out
+    };
+    runs.into_iter()
         .filter(|w| !SAFETY_KEYWORD_STOPWORDS.contains(&w.as_str()))
         .take(5)
         .map(Json::Str)
@@ -2049,6 +2100,7 @@ pub fn safety_extract_keywords(text: &str) -> Vec<Json> {
 
 /// Oracle report_outcome + _update_pattern: full state transition.
 /// Returns {"patterns": [...], "stats": {...}} with `now` stamped (no-clock).
+#[must_use]
 pub fn safety_report_outcome(
     patterns: &[Json],
     stats: &Json,
@@ -2191,6 +2243,7 @@ fn update_pattern(
 
 /// Oracle _save's serializable half: the exact json.dump(indent=2) bytes the
 /// facade writes to disk (file IO stays Python-side).
+#[must_use]
 pub fn safety_save(patterns: &[Json], stats: &Json) -> Json {
     Json::Obj(vec![
         (
@@ -2217,6 +2270,7 @@ fn fmt_percent(rate: f64) -> String {
 }
 
 /// Oracle get_stats_text (pattern_count = len(self.patterns), NOT a stat).
+#[must_use]
 pub fn safety_get_stats_text(stats: &Json, pattern_count: usize) -> String {
     let mut lines = vec!["[bold]🛡 Safety Override Stats[/bold]".to_string()];
     lines.push(format!(
@@ -2247,6 +2301,7 @@ pub fn safety_get_stats_text(stats: &Json, pattern_count: usize) -> String {
 }
 
 /// Oracle get_patterns_text.
+#[must_use]
 pub fn safety_get_patterns_text(patterns: &[Json]) -> String {
     if patterns.is_empty() {
         return "[dim]No patterns learned yet[/dim]".to_string();
@@ -2298,12 +2353,14 @@ pub fn safety_get_patterns_text(patterns: &[Json]) -> String {
 }
 
 /// Oracle clear_patterns.
+#[must_use]
 pub fn safety_clear_patterns(_patterns: &[Json]) -> Vec<Json> {
     vec![]
 }
 
 /// get_params deterministic core: first matching pattern wins; returns the
 /// params dict and whether a pattern matched (stats bump is the facade's).
+#[must_use]
 pub fn safety_get_params_for_text(patterns: &[Json], text: &str) -> Json {
     for pattern in patterns {
         let kws: Vec<Json> = match pattern.get("keywords") {
@@ -2362,6 +2419,7 @@ pub fn safety_get_params_for_text(patterns: &[Json], text: &str) -> Json {
 
 /// Oracle self_test: the same 7 checks against kernel-internal state
 /// (no disk). Returns the check display names plus pass/fail counts.
+#[must_use]
 pub fn safety_self_test(now: &str) -> Json {
     let names = [
         "Normal request → tier 0".to_string(),

src/rust/cognition_core/src/safety_battery.rs

@@ -168,4 +168,146 @@ mod battery {
             }
         }
     }
+
+    // ── Phase B: language-advantage self-checks (truth untouched) ──────────
+
+    fn s(v: &Json) -> &str {
+        match v {
+            Json::Str(x) => x.as_str(),
+            _ => "",
+        }
+    }
+
+    fn arr(v: &Json) -> &[Json] {
+        match v {
+            Json::Arr(x) => x,
+            _ => &[],
+        }
+    }
+
+    /// The zero-alloc ASCII window scan must agree with the Python-parity
+    /// Unicode path for every (text, keyword) pair in the frozen corpus —
+    /// ASCII folding is 1:1 there, so both paths must be byte-equal.
+    #[test]
+    fn phase_b_ascii_scan_equals_unicode_path() {
+        let cases = pyjson::parse(METHODS).expect("methods truth parses");
+        let items = match cases {
+            Json::Arr(x) => x,
+            _ => panic!("methods truth must be a list"),
+        };
+        let mut checked = 0usize;
+        for c in &items {
+            let kernel = s(c.get("kernel").unwrap_or(&Json::Null));
+            if kernel == "safety_py_matches" {
+                let args_owned = c.get("args").cloned().unwrap_or(Json::Arr(vec![]));
+                let args = arr(&args_owned);
+                let text = s(&args[0]);
+                for kw in arr(&args[1]) {
+                    let fast = broker::contains_ci(text, s(kw));
+                    let slow = text.to_lowercase().contains(&s(kw).to_lowercase());
+                    assert_eq!(fast, slow, "contains_ci divergence {text:?}/{:?}", s(kw));
+                    checked += 1;
+                }
+            }
+            if kernel == "safety_extract_keywords" {
+                let args_owned = c.get("args").cloned().unwrap_or(Json::Arr(vec![]));
+                let args = arr(&args_owned);
+                let text = s(&args[0]);
+                if !text.is_ascii() {
+                    continue;
+                }
+                // byte-scan path vs an inline char-path mirror
+                let via_kernel = broker::safety_extract_keywords(text);
+                let low = text.to_lowercase();
+                let mut mirror: Vec<String> = vec![];
+                let mut cur = String::new();
+                for ch in low.chars() {
+                    if ch.is_ascii_lowercase() {
+                        cur.push(ch);
+                    } else {
+                        if cur.chars().count() >= 4 {
+                            mirror.push(std::mem::take(&mut cur));
+                        }
+                        cur.clear();
+                    }
+                }
+                if cur.chars().count() >= 4 {
+                    mirror.push(cur);
+                }
+                let mirror: Vec<Json> =
+                    mirror.into_iter().take(5).map(Json::Str).collect();
+                assert_eq!(norm(&Json::Arr(via_kernel)), norm(&Json::Arr(mirror)),
+                    "extract_keywords fast-path divergence {text:?}");
+                checked += 1;
+            }
+            if kernel == "safety_is_refusal" {
+                let args_owned = c.get("args").cloned().unwrap_or(Json::Arr(vec![]));
+                let args = arr(&args_owned);
+                let text = s(&args[0]);
+                // [:200] chars vs bytes equivalence on the frozen corpus
+                let by_chars: String = text.chars().take(200).collect();
+                let by_bytes: String = String::from_utf8_lossy(
+                    &text.as_bytes().iter().copied().take(200).collect::<Vec<u8>>(),
+                ).into_owned();
+                if text.is_ascii() {
+                    assert_eq!(by_chars, by_bytes, "200-unit slice divergence {text:?}");
+                }
+                checked += 1;
+            }
+        }
+        assert!(checked >= 25, "expected the full corpus, saw {checked} rows");
+    }
+
+    /// std-only Phase B bench: refusal regex + keyword scan hot paths over
+    /// the frozen corpus (bounded iterations, no external harness).
+    #[test]
+    fn phase_b_hot_path_bench() {
+        let cases = pyjson::parse(METHODS).expect("methods truth parses");
+        let items = match cases {
+            Json::Arr(x) => x,
+            _ => panic!("methods truth must be a list"),
+        };
+        let mut texts: Vec<String> = vec![];
+        let mut pairs: Vec<(String, Vec<String>)> = vec![];
+        for c in &items {
+            let kernel = s(c.get("kernel").unwrap_or(&Json::Null));
+            let args_owned = c.get("args").cloned().unwrap_or(Json::Arr(vec![]));
+            let args = arr(&args_owned);
+            match kernel {
+                "safety_is_refusal" => texts.push(s(&args[0]).to_string()),
+                "safety_py_matches" => pairs.push((
+                    s(&args[0]).to_string(),
+                    arr(&args[1]).iter().map(|k| s(k).to_string()).collect(),
+                )),
+                _ => {}
+            }
+        }
+        let t0 = std::time::Instant::now();
+        let mut hits = 0usize;
+        for _ in 0..2_000 {
+            for t in &texts {
+                if broker::safety_is_refusal(t) {
+                    hits += 1;
+                }
+            }
+        }
+        let refusal_us = t0.elapsed().as_secs_f64() * 1e6 / (2_000 * texts.len()) as f64;
+        let t1 = std::time::Instant::now();
+        let mut kw_hits = 0usize;
+        for _ in 0..2_000 {
+            for (t, kws) in &pairs {
+                if broker::contains_ci(t, &kws.join("\u{1}")) {
+                    kw_hits += 1; // joined probe keeps the call shape honest
+                }
+            }
+        }
+        let kw_us = t1.elapsed().as_secs_f64() * 1e6 / (2_000 * pairs.len()) as f64;
+        println!(
+            "phase B bench: is_refusal {refusal_us:.2} us/call over {} texts \
+             (+{hits} hits); contains_ci {kw_us:.2} us/call over {} pairs",
+            texts.len(),
+            pairs.len()
+        );
+        let _ = (hits, kw_hits);
+    }
 }