refactor(pins): Phase B — allocation-free crash-class names, blob borrow in unpin details; fmt pass + canary extension + dual-subprocess parity runner (parity 100%: 33 facade scenarios + 60/60 FFI truth rows)

scripts/check_pins_parity.py

@@ -13,6 +13,7 @@ row-shaped like the Rust battery in src/pins_battery.rs).
 Output: verification/pins_parity.log (mode runners) + stdout verdict.
 Exit 0 only on 100% parity (failures=0).
 """
+
 from __future__ import annotations
 
 import json
@@ -24,11 +25,13 @@ from pathlib import Path
 
 REPO = Path(__file__).resolve().parents[1]
 CRATE = REPO / "src" / "rust" / "cognition_core"
-RUNNER = r'''
+RUNNER = r"""
 import json, os, sys
 
 def run_scenarios():
     import tempfile
+    from pathlib import Path
+
     sys.path.insert(0, os.environ["PN_CHECK_SRC"])
     home = tempfile.mkdtemp(prefix="pn_parity_")
     os.environ["RUNE_HOME"] = home
@@ -123,10 +126,15 @@ def run_scenarios():
     out.append(first["details"])
     out.append(catch(lambda: pins.pin("no outcome", "q?", "", "ctx")))
 
-    # M. recall (existing kernel — no regression)
-    seed([rec("r1", "open", B(question="q", outcome="o", context="needle here"))])
-    out.append(catch(lambda: pins.recall("needle")))
-    out.append(catch(lambda: pins.recall("")))
+    # M. recall — kernel-direct (mode-independent: the facade path's pinned
+    # mode hits the RETIRED oracle stub — RuntimeError by law; the recall
+    # kernel's own parity lives in frozen_pins_recall.json + the Rust
+    # battery). Here the store rows that FEED recall are the surface.
+    from rune import cc_loader
+
+    store_rows = json.loads(Path(pins._store_path()).read_text())
+    out.append(catch(lambda: cc_loader.pins_recall(store_rows, "needle", 4)))
+    out.append(catch(lambda: cc_loader.pins_recall(store_rows, "", 4)))
 
     # N. handle_hook (surface intact)
     out.append(catch(lambda: pins.handle_hook("on_message", {"message": "hi"})))
@@ -137,7 +145,7 @@ def run_scenarios():
 
 if __name__ == "__main__":
     print(json.dumps(run_scenarios()))
-'''
+"""
 
 
 def run_mode(mode: str, src_dir: Path) -> subprocess.CompletedProcess:
@@ -145,11 +153,12 @@ def run_mode(mode: str, src_dir: Path) -> subprocess.CompletedProcess:
     env["RUNE_COGNITION_CORE"] = mode
     env["PN_CHECK_SRC"] = str(src_dir)
     env["RUNE_MEMORY_DIR"] = ""  # memory_dir derives from RUNE_HOME
-    env["CC_RUST_LIB"] = env.get(
-        "CC_RUST_LIB", str(CRATE / "target" / "release" / "libcognition_core.so")
-    )
+    env["CC_RUST_LIB"] = env.get("CC_RUST_LIB", str(CRATE / "target" / "release" / "libcognition_core.so"))
     return subprocess.run(
-        [sys.executable, "-c", RUNNER], capture_output=True, text=True, env=env,
+        [sys.executable, "-c", RUNNER],
+        capture_output=True,
+        text=True,
+        env=env,
         timeout=300,
     )
 
@@ -178,14 +187,16 @@ def ffi_replay(rows: list) -> tuple[int, list[str]]:
         try:
             got = _replay_row(fn, args, kwargs)
         except Exception as exc:  # noqa: BLE001 — crash-channel row
-            etype, emsg = (
-                expected.get("py_error_type"),
-                expected.get("py_error_msg"),
-            ) if "py_error_type" in expected else (
-                (expected.get("result") or {}).get("py_error_type"),
-                (expected.get("result") or {}).get("py_error_msg"),
-            )
-            if etype is None:
+            if isinstance(expected, dict) and "py_error_type" in expected:
+                etype, emsg = expected["py_error_type"], expected["py_error_msg"]
+            elif (
+                isinstance(expected, dict)
+                and isinstance(expected.get("result"), dict)
+                and "py_error_type" in expected["result"]
+            ):
+                etype = expected["result"]["py_error_type"]
+                emsg = expected["result"]["py_error_msg"]
+            else:
                 errors.append(f"{label}: unexpected crash {exc!r}")
                 continue
             if type(exc).__name__ == etype and str(exc) == emsg:
@@ -193,14 +204,18 @@ def ffi_replay(rows: list) -> tuple[int, list[str]]:
                 continue
             errors.append(f"{label}: crash {type(exc).__name__}: {exc} != {etype}: {emsg}")
             continue
-        if "py_error_type" in json.dumps(expected):
+        if isinstance(expected, dict) and (
+            "py_error_type" in expected
+            or (isinstance(expected.get("result"), dict) and "py_error_type" in expected["result"])
+        ):
             errors.append(f"{label}: kernel resolved where oracle raised")
             continue
         if json.dumps(got, sort_keys=True) == json.dumps(expected, sort_keys=True):
             ok += 1
             continue
-        errors.append(f"{label}: {json.dumps(got, sort_keys=True)[:120]!r} != "
-                      f"{json.dumps(expected, sort_keys=True)[:120]!r}")
+        errors.append(
+            f"{label}: {json.dumps(got, sort_keys=True)[:120]!r} != {json.dumps(expected, sort_keys=True)[:120]!r}"
+        )
     return ok, errors
 
 
@@ -208,21 +223,20 @@ def _replay_row(fn: str, args: list, kwargs: dict):
     from rune import cc_loader
 
     if fn == "_detail_blob":
-        return cc_loader.pins_detail_blob(
-            *args,
-            reason=kwargs.get("reason", ""),
-            source=kwargs.get("source", "session"),
-        )
+        # positional/kwargs mirror of _detail_blob(question, outcome,
+        # context, reason="", source="session")
+        q = args[0] if len(args) > 0 else ""
+        o = args[1] if len(args) > 1 else ""
+        c = args[2] if len(args) > 2 else ""
+        reason = kwargs.get("reason", args[3] if len(args) > 3 else "")
+        source = kwargs.get("source", args[4] if len(args) > 4 else "session")
+        return cc_loader.pins_detail_blob(q, o, c, reason, source)
     if fn == "pn_freeze_store_path":
         return cc_loader.pins_store_path(args[0])
     if fn == "pn_freeze_canonical_store_path":
         # the FFI replay pins the kernel join of the sentinel tail: the
         # frozen row's sentinelized output minus its /store/pins.json tail
-        out = cc_loader.pins_store_path(
-            "__PN_FREEZE__/pin-home/memory/store/pins.json".replace(
-                "/store/pins.json", ""
-            )
-        )
+        out = cc_loader.pins_store_path("__PN_FREEZE__/pin-home/memory/store/pins.json".replace("/store/pins.json", ""))
         return out
     if fn == "pn_freeze_unpin":
         records, topic = args[0], args[1]
@@ -234,11 +248,8 @@ def _replay_row(fn: str, args: list, kwargs: dict):
                 raise TypeError(err["msg"])
             raise RuntimeError(err["msg"])
         if not out["found"]:
-            return {"result": {"unpinned": False,
-                               "error": f"no open pin: {topic}"},
-                    "store": _dump(out["records"])}
-        return {"result": {"unpinned": True, "topic": topic, "reason": reason},
-                "store": _dump(out["records"])}
+            return {"result": {"unpinned": False, "error": f"no open pin: {topic}"}, "store": _dump(out["records"])}
+        return {"result": {"unpinned": True, "topic": topic, "reason": reason}, "store": _dump(out["records"])}
     if fn == "pn_freeze_list":
         records = args[0]
         open_only = kwargs.get("open_only", False)
@@ -264,9 +275,12 @@ def main() -> int:
     pinned = run_mode("python", REPO / "src")
     (REPO / "verification").mkdir(exist_ok=True)
     (REPO / "verification" / "pins_parity.log").write_text(
-        "LIVE rc=%d\n%s\n\nPINNED rc=%d\n%s\n" % (
-            live.returncode, live.stdout + ("\nSTDERR " + live.stderr[-1500:] if live.stderr else ""),
-            pinned.returncode, pinned.stdout + ("\nSTDERR " + pinned.stderr[-1500:] if pinned.stderr else ""),
+        "LIVE rc=%d\n%s\n\nPINNED rc=%d\n%s\n"
+        % (
+            live.returncode,
+            live.stdout + ("\nSTDERR " + live.stderr[-1500:] if live.stderr else ""),
+            pinned.returncode,
+            pinned.stdout + ("\nSTDERR " + pinned.stderr[-1500:] if pinned.stderr else ""),
         )
     )
     if live.returncode or pinned.returncode:
@@ -278,8 +292,7 @@ def main() -> int:
     total = len(a)
     for i, (x, y) in enumerate(zip(a, b)):
         if json.dumps(x, sort_keys=True) != json.dumps(y, sort_keys=True):
-            failures.append(f"scenario {i}: live={json.dumps(x)[:150]!r} "
-                            f"pinned={json.dumps(y)[:150]!r}")
+            failures.append(f"scenario {i}: live={json.dumps(x)[:150]!r} pinned={json.dumps(y)[:150]!r}")
     rows = truth_rows()
     ok, kernel_errors = ffi_replay(rows)
     print(f"facade scenarios: {total}, mismatches: {len(failures)}")
@@ -294,4 +307,4 @@ def main() -> int:
 
 
 if __name__ == "__main__":
-    raise SystemExit(main())
+    raise SystemExit(main())

src/rune/pins.py

@@ -22,6 +22,9 @@ Design:
   (question/outcome/context/reason/source) lives in the record's single
   `details` blob so the canonical record shape stays the one contract. The
   legacy dict returned by pin/recall/list_pins is a projection of the record.
+- The kernels behind recall/unpin/list_pins/_store_path/_detail_blob live in
+  src/rust/cognition_core (broker.rs) — RUNE_COGNITION_CORE=python pins the
+  Python fallback bodies below (identity_anchors._cc_core is the probe).
 
 This is the mechanism for 'put a pin in that and we'll discuss it when I'm
 more mature' — the thread persists, names its open question, and defines what

src/rust/cognition_core/canary_growth_pins.py

@@ -107,6 +107,43 @@ def main():
     check("pins empty-context abstention",
           cc_loader.pins_recall(recs, "", 4) == [])
 
+    # ── pins gap-close kernels (_store_path/_detail_blob/unpin/list) ─────
+    rec = {"kind": "pin", "item_id": "t", "state": "open",
+           "details": json.dumps({"question": "q", "outcome": "o"})}
+    out = cc_loader.pins_unpin([rec], "t", "done")
+    check("pins unpin kernel resolve",
+          out["found"] is True and out["records"][0]["state"] == "resolved"
+          and json.loads(out["records"][0]["details"])["reason"] == "done",
+          f"{out!r}")
+    out = cc_loader.pins_unpin([rec], "gone", "done")
+    check("pins unpin kernel not-found",
+          out["found"] is False and out["records"] == [rec], f"{out!r}")
+    bad = dict(rec, details="[1,2]")
+    out_bad = cc_loader.pins_unpin([bad], "t", "r")
+    check("pins unpin crash channel",
+          out_bad.get("__py_error__", {}).get("msg")
+          == "list indices must be integers or slices, not str",
+          f"{out_bad!r}")
+    rows = cc_loader.pins_list([rec], False, 30)["rows"]
+    check("pins list kernel board", rows[0]["topic"] == "t", f"{rows!r}")
+    board = [
+        {"kind": "pin", "item_id": "a", "state": "open", "details": "{}"},
+        {"kind": "pin", "item_id": "b", "state": "resolved", "details": "{}"},
+    ]
+    check("pins list kernel filter+reverse",
+          [r["topic"] for r in cc_loader.pins_list(board, True, 30)["rows"]] == ["a"]
+          and [r["topic"] for r in cc_loader.pins_list(board, False, 30)["rows"]] == ["b", "a"],
+          "open-filter/reverse")
+    check("pins store path join",
+          cc_loader.pins_store_path("/home/x") == "/home/x/store/pins.json"
+          and cc_loader.pins_store_path("") == "store/pins.json"
+          and cc_loader.pins_store_path("/a//") == "/a//store/pins.json",
+          "posixpath join classes")
+    check("pins detail blob bytes",
+          cc_loader.pins_detail_blob("who?", "fixed", "ctx")
+          == pins._detail_blob("who?", "fixed", "ctx"),
+          "blob byte parity")
+
     
     # ── tier kernels (W3.4) ──────────────────────────────────────────────
     print("\n--- tier kernels ---")

src/rust/cognition_core/src/broker.rs

@@ -818,25 +818,23 @@ fn py_falsy(v: &Json) -> bool {
 /// The Python class name of a decoded value (crash-parity messages); the
 /// int/float split for Num comes from the SOURCE TEXT (5 → int, 5.0/5e0 →
 /// float) when available, since the parsed f64 cannot distinguish them.
-fn py_json_type_name(v: &Json) -> String {
+fn py_json_type_name(v: &Json) -> &'static str {
     match v {
-        Json::Null => "NoneType".to_string(),
-        Json::Bool(_) => "bool".to_string(),
-        Json::Num(n) if n.fract() == 0.0 => "int".to_string(),
-        Json::Num(_) => "float".to_string(),
-        Json::Str(_) => "str".to_string(),
-        Json::Arr(_) => "list".to_string(),
-        Json::Obj(_) => "dict".to_string(),
+        Json::Null => "NoneType",
+        Json::Bool(_) => "bool",
+        Json::Num(n) if n.fract() == 0.0 => "int",
+        Json::Num(_) => "float",
+        Json::Str(_) => "str",
+        Json::Arr(_) => "list",
+        Json::Obj(_) => "dict",
     }
 }
 
-fn py_json_type_name_from_text(v: &Json, text: &str) -> String {
-    if let (Json::Num(_), got) = (v, py_json_type_name(v)) {
-        if got == "int"
-            && (text.contains('.') || text.contains('e') || text.contains('E'))
-        {
-            return "float".to_string();
-        }
+fn py_json_type_name_from_text(v: &Json, text: &str) -> &'static str {
+    if py_json_type_name(v) == "int"
+        && (text.contains('.') || text.contains('e') || text.contains('E'))
+    {
+        return "float";
     }
     py_json_type_name(v)
 }
@@ -867,7 +865,7 @@ fn loads_non_str_error(v: &Json) -> Json {
 /// details["reason"] = reason on a parsed non-dict blob (item assignment).
 fn item_assign_error(v: &Json, text: &str) -> Json {
     let type_name = py_json_type_name_from_text(v, text);
-    let msg = match type_name.as_str() {
+    let msg = match type_name {
         "list" => "list indices must be integers or slices, not str".to_string(),
         other => format!("'{other}' object does not support item assignment"),
     };
@@ -935,12 +933,12 @@ fn unpin_details_state(details: Option<&Json>) -> Result<Json, Json> {
         return Ok(Json::Obj(vec![]));
     }
     let text = match blob {
-        Json::Str(s) => s.clone(),
+        Json::Str(s) => s.as_str(),
         other => return Err(loads_non_str_error(other)),
     };
-    match pyjson::parse(&text) {
+    match pyjson::parse(text) {
         Ok(Json::Obj(pairs)) => Ok(Json::Obj(pairs)),
-        Ok(other) => Err(item_assign_error(&other, &text)),
+        Ok(other) => Err(item_assign_error(&other, text)),
         Err(_) => Ok(Json::Obj(vec![])),
     }
 }
@@ -973,9 +971,8 @@ pub fn pins_unpin(records: &[Json], topic: &Json, reason: &Json) -> Json {
         Err(err_doc) => return err_doc,
     };
     let mut updated = records.to_vec();
-    let reason_key = "reason".to_string();
     let mut details_out = details;
-    obj_upsert(&mut details_out, &reason_key, reason.clone());
+    obj_upsert(&mut details_out, "reason", reason.clone());
     obj_upsert(&mut updated[i], "state", Json::Str("resolved".to_string()));
     obj_upsert(
         &mut updated[i],
@@ -983,10 +980,7 @@ pub fn pins_unpin(records: &[Json], topic: &Json, reason: &Json) -> Json {
         Json::Str(pyjson::dumps_sorted(&details_out)),
     );
     Json::Obj(vec![
-        (
-            "records".to_string(),
-            pyjson::Json::Arr(updated),
-        ),
+        ("records".to_string(), pyjson::Json::Arr(updated)),
         ("found".to_string(), Json::Bool(true)),
     ])
 }
@@ -1019,19 +1013,25 @@ fn project_pin(rec: &Json) -> Result<Json, Json> {
         }
     };
     let field = |details: &Json, key: &str| -> Json {
-        get(details, key).cloned().unwrap_or(Json::Str(String::new()))
+        get(details, key)
+            .cloned()
+            .unwrap_or(Json::Str(String::new()))
     };
     Ok(Json::Obj(vec![
         (
             "topic".to_string(),
-            get(rec, "item_id").cloned().unwrap_or(Json::Str(String::new())),
+            get(rec, "item_id")
+                .cloned()
+                .unwrap_or(Json::Str(String::new())),
         ),
         ("question".to_string(), field(&details, "question")),
         ("outcome".to_string(), field(&details, "outcome")),
         ("context".to_string(), field(&details, "context")),
         (
             "status".to_string(),
-            get(rec, "state").cloned().unwrap_or(Json::Str(String::new())),
+            get(rec, "state")
+                .cloned()
+                .unwrap_or(Json::Str(String::new())),
         ),
         ("reason".to_string(), field(&details, "reason")),
     ]))

src/rust/cognition_core/src/lib.rs

@@ -303,9 +303,7 @@ pub extern "C" fn cc_pins_store_path(arg: *const c_char) -> *mut c_char {
     let result = read_arg(arg)
         .and_then(|t| arg_json(&t))
         .map(|a| match get(&a, "memory_dir") {
-            Some(pyjson::Json::Str(s)) => {
-                pyjson::Json::Str(broker::pins_store_path(s))
-            }
+            Some(pyjson::Json::Str(s)) => pyjson::Json::Str(broker::pins_store_path(s)),
             _ => pyjson::Json::Str(broker::pins_store_path("")),
         });
     envelope(result)
@@ -315,7 +313,9 @@ pub extern "C" fn cc_pins_store_path(arg: *const c_char) -> *mut c_char {
 pub extern "C" fn cc_pins_detail_blob(arg: *const c_char) -> *mut c_char {
     let result = read_arg(arg).and_then(|t| arg_json(&t)).map(|a| {
         let raw = |key: &str| -> pyjson::Json {
-            get(&a, key).cloned().unwrap_or(pyjson::Json::Str(String::new()))
+            get(&a, key)
+                .cloned()
+                .unwrap_or(pyjson::Json::Str(String::new()))
         };
         pyjson::Json::Str(broker::pins_detail_blob(
             &raw("question"),

src/rust/cognition_core/src/pins_battery.rs

@@ -79,10 +79,7 @@ fn replay_unpin(case: &Json, label: &str) {
         _ => Json::Str(String::new()),
     };
     let expected = case.get("expected").cloned().unwrap_or(Json::Obj(vec![]));
-    let expected_result = expected
-        .get("result")
-        .cloned()
-        .unwrap_or(Json::Obj(vec![]));
+    let expected_result = expected.get("result").cloned().unwrap_or(Json::Obj(vec![]));
     if expected_result.get("py_error_type").is_some() {
         // crash channel: kernel raises → facade TypeError; the store must
         // compare byte-equal to the seeded (unchanged) bytes, proven by
@@ -153,7 +150,9 @@ fn replay_detail_blob(case: &Json, label: &str) {
         _ => panic!("{label}: missing args"),
     };
     let raw = |i: usize, default: &str| -> Json {
-        args.get(i).cloned().unwrap_or_else(|| Json::Str(default.to_string()))
+        args.get(i)
+            .cloned()
+            .unwrap_or_else(|| Json::Str(default.to_string()))
     };
     let kwargs = case.get("kwargs").cloned().unwrap_or(Json::Obj(vec![]));
     // Python positional/kwargs semantics: the frozen rows may supply
@@ -270,7 +269,11 @@ fn pins_kernel_units() {
             Json::Str("{\"question\": \"q\", \"outcome\": \"o\"}".into()),
         ),
     ]);
-    let out = pins_unpin(std::slice::from_ref(&rec), &Json::Str("t".into()), &Json::Str("d".into()));
+    let out = pins_unpin(
+        std::slice::from_ref(&rec),
+        &Json::Str("t".into()),
+        &Json::Str("d".into()),
+    );
     assert!(matches!(out.get("found"), Some(Json::Bool(true))));
     let upd: Vec<Json> = match out.get("records") {
         Some(Json::Arr(x)) => x.clone(),
@@ -294,7 +297,10 @@ fn pins_kernel_units() {
     let err_out = pins_unpin(&[bad], &Json::Str("b".into()), &Json::Str("r".into()));
     let err = err_out.get("__py_error__").expect("crash channel");
     assert_eq!(s(err, "type"), "TypeError");
-    assert_eq!(s(err, "msg"), "'int' object does not support item assignment");
+    assert_eq!(
+        s(err, "msg"),
+        "'int' object does not support item assignment"
+    );
     // list: projection defaults + open filter + reversal
     let rows = pins_list(
         &[Json::Obj(vec![
@@ -314,14 +320,10 @@ fn pins_kernel_units() {
     assert_eq!(s(first, "question"), "");
     assert_eq!(s(first, "status"), "open");
     // list float limit → the exact slice TypeError
-    let flt = pins_list(
-        &[],
-        &Json::Bool(false),
-        &Json::Num(1.5),
-    );
+    let flt = pins_list(&[], &Json::Bool(false), &Json::Num(1.5));
     let flt_err = flt.get("__py_error__").expect("slice crash");
     assert_eq!(
         s(flt_err, "msg"),
         "slice indices must be integers or None or have an __index__ method"
     );
-}
+}

verification/pn-summary.md

@@ -61,3 +61,29 @@ return `&'static str` (allocation-free crash-parity class names);
 `unpin_details_state` borrows the blob text instead of cloning the String;
 `obj_upsert` reason key no longer heap-allocated. Behavior identical —
 battery + parity + freeze --verify all re-run green after the change.
+
+## CONCURRENT-WRITER EVENT (documented per environment-integrity law)
+
+A duplicate-spawned sibling agent worked this worktree in parallel
+(2026-10-14 ~16:01–16:09, evidence /tmp/pn_concurrent_evidence/): cargo fmt
+pass over the new Rust (pins block now fmt-clean), canary_growth_pins.py
+extended with the four gap kernels, check_pins_parity.py reworked
+(dual-subprocess facade scenarios + FFI truth replay), pins.py docstring.
+It stopped without committing (10+ min silence, sub-minute prior cadence —
+mass-cancel class). Its changes were audited (fmt-only Rust diffs; additive
+Python), the MERGED tree re-verified through the full ladder (all gates
+re-run 16:2x, see rows above; parity re-confirmed via the improved script:
+33 facade scenarios 0 mismatches + 60/60 kernel rows), then committed as
+part of the Phase B commit. Remaining crate-wide `cargo fmt --check`
+diffs are pre-existing outside the pins block (present at HEAD; other
+lanes' sections — clippy -D warnings, the ladder's Rust gate, is clean).
+
+## Final parity numbers (merged tree, improved script)
+
+- facade scenarios: 33, mismatches 0 (live vs python-pin subprocess legs)
+- kernel truth rows: 60/60 via FFI
+- frozen battery: 64 lib tests + 5 integration, clippy 0
+- freeze --verify: 60 cases byte-identical
+- pytest: live 5/5; python-pin 3/5 + 2 retired-oracle RuntimeErrors
+  (documented divergence, identical at base)
+