feat(pins): close PARTIAL gaps — dichotomy wave (7/11 -> 11/11 RUST; 60-row frozen truth via pn_freeze_seams rig, battery replay verbatim, live parity 190/190 both pins)

docs/dichotomy-rebase-honest-audit.md

@@ -2,16 +2,16 @@
 
 _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: 64%** (122/192 runtime modules at FULL)
+**FULL completion: 64%** (123/192 runtime modules at FULL)
 
-Function-level detail: 74% (1698/2288 runtime functions matched) — secondary; includes in-flight work
+Function-level detail: 74% (1702/2288 runtime functions matched) — secondary; includes in-flight work
 
 | Status | Modules | Lines | Meaning |
 |---|---|---|---|
-| FULL | 122 | 49925 | 80%+ of functions ported |
-| PARTIAL | 31 | 14257 | 20–80% — real logic exists, gaps named below |
+| FULL | 123 | 50245 | 80%+ of functions ported |
+| PARTIAL | 30 | 13975 | 20–80% — real logic exists, gaps named below |
 | NOT TOUCHED | 39 | 11471 | no name-preserving implementation |
-| Total runtime | 192 | 75653 | |
+| Total runtime | 192 | 75691 | |
 
 ## NOT TOUCHED — the port queue, in priority order (largest first)
 
@@ -70,7 +70,6 @@ Function-level detail: 74% (1698/2288 runtime functions matched) — secondary;
 | voicebox | 164 | 5/11 | _run, start_recording, stop_recording, _transcribe_and_type, on_press, on_release |
 | persona_definition | 210 | 2/7 | _by_topic, compile_full, emit_constitution, emit_identity, emit_persona |
 | drift_observability | 399 | 5/9 | _do_core, main, _load_alerts, _save_alerts |
-| pins | 282 | 7/11 | _store_path, _detail_blob, unpin, list_pins |
 | prompt_sync | 487 | 11/15 | _now, _rust_available, get_store, main |
 | task_gate | 301 | 5/9 | _tg_core, _now, list_gates, main |
 | connection_pool | 226 | 7/10 | get_pool, _kernel_state, _get_connection_core |
@@ -164,6 +163,7 @@ Function-level detail: 74% (1698/2288 runtime functions matched) — secondary;
 | progressive_escalation | 340 | 9/9 | RUST |
 | builder_bridge | 326 | 11/11 | GO |
 | drift_compression | 323 | 7/7 | RUST |
+| pins | 320 | 11/11 | RUST |
 | outcome_ledger | 318 | 7/8 | RUST |
 | construction_projection | 317 | 4/4 | RUST |
 | message_queue | 316 | 13/13 | GO |

scripts/check_pins_parity.py

@@ -0,0 +1,297 @@
+#!/usr/bin/env python3
+"""check_pins_parity.py — live parity for the pins gap-close (dichotomy wave).
+
+Runs the SAME scenario corpus through (a) the LIVE delegation (cc kernel
+authoritative, RUNE_COGNITION_CORE=auto) and (b) the oracle-pinned floor
+(RUNE_COGNITION_CORE=python) in SEPARATE clean subprocesses, byte-comparing
+every observable: return envelopes, post-call store file contents, and the
+exception class+message of the crash-parity rows. Plus: every frozen
+gap-truth row (src/rust/cognition_core/src/frozen_pins_truth.json) replayed
+through the FFI against the pinned-oracle expectations (kernel-vs-oracle,
+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
+import os
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+CRATE = REPO / "src" / "rust" / "cognition_core"
+RUNNER = r'''
+import json, os, sys
+
+def run_scenarios():
+    import tempfile
+    sys.path.insert(0, os.environ["PN_CHECK_SRC"])
+    home = tempfile.mkdtemp(prefix="pn_parity_")
+    os.environ["RUNE_HOME"] = home
+    from rune import pins
+    out = []
+
+    def seed(records):
+        os.makedirs(pins._store_path().parent, exist_ok=True)
+        pins._save(records)
+
+    def rec(iid, state, blob):
+        return {"schema_version": 1, "kind": "pin", "item_id": iid,
+                "owner": "pins", "state": state, "evidence_ref": "",
+                "details": blob}
+
+    def B(**kw):
+        return pins._detail_blob(kw.get("question", ""), kw.get("outcome", ""),
+                                 kw.get("context", ""), kw.get("reason", ""),
+                                 kw.get("source", "session"))
+
+    def catch(fn):
+        try:
+            return fn()
+        except Exception as exc:
+            return {"py_error_type": type(exc).__name__, "py_error_msg": str(exc)}
+
+    # A. unpin lifecycle (found)
+    seed([rec("t1", "open", B(question="q1", outcome="o1", context="c1"))])
+    out.append(catch(lambda: pins.unpin("t1", reason="done")))
+    out.append(Path(pins._store_path()).read_text())
+
+    # B. most recent open pin wins
+    seed([rec("dup", "open", B(question="q1", outcome="o1", context="c1")),
+          rec("dup", "open", B(question="q2", outcome="o2", context="c2"))])
+    out.append(catch(lambda: pins.unpin("dup", reason="rr")))
+    out.append("".join(r["state"] for r in
+                       json.loads(Path(pins._store_path()).read_text())))
+
+    # C. not-found envelope
+    seed([rec("absent", "open", B(question="q", outcome="o", context="c"))])
+    before = Path(pins._store_path()).read_text()
+    out.append(catch(lambda: pins.unpin("nope topic")))
+    out.append(Path(pins._store_path()).read_text() == before)
+
+    # D. junk blob folds to {}
+    seed([rec("j", "open", "junk{")])
+    out.append(catch(lambda: pins.unpin("j", reason="r")))
+    out.append(json.loads(Path(pins._store_path()).read_text())[0]["details"])
+
+    # E. list blob → TypeError crash parity
+    seed([rec("l", "open", "[1,2]")])
+    out.append(catch(lambda: pins.unpin("l", reason="r")))
+
+    # F. int details VALUE → json.loads TypeError
+    r = rec("v", "open", None)
+    r["details"] = 5
+    seed([r])
+    out.append(catch(lambda: pins.unpin("v", reason="r")))
+
+    # G/H. list: board, filter, limits (0 / 1 / neg / None)
+    seed([rec("t1", "open", B(question="q1", outcome="o1", context="shared")),
+          rec("t2", "open", B(question="q2", outcome="o2", context="b-only")),
+          rec("t3", "resolved", B(question="q3", outcome="o3", context="shared"))])
+    out.append([p["topic"] for p in pins.list_pins()])
+    out.append([p["topic"] for p in pins.list_pins(open_only=True)])
+    out.append([p["topic"] for p in pins.list_pins(open_only=True, limit=1)])
+    out.append([p["topic"] for p in pins.list_pins(limit=2)])
+    out.append([p["topic"] for p in pins.list_pins(limit=0)])
+    out.append([p["topic"] for p in pins.list_pins(limit=-1)])
+    out.append(len(pins.list_pins(limit=None)))
+    out.append(len(pins.list_pins(limit=True)))
+
+    # I. list crash rows (string blobs parsing to non-dicts RAISE)
+    for blob in ("null", "5", "[1]"):
+        seed([rec("x", "open", blob)])
+        out.append(catch(lambda: pins.list_pins()))
+
+    # J. detail blob bytes (unicode + astral + separators)
+    out.append(pins._detail_blob("qué?", "out\U0001f600", "c<>&"))
+    out.append(pins._detail_blob("d\"q", "b\\", "n\nl\tt"))
+    out.append(pins._detail_blob("q", 3, "c", True, "session"))
+
+    # K. store path derivation (live kernel vs pinned constant — same env)
+    out.append(str(pins._store_path()))
+    out.append(str(pins._store_path()) == str(pins.PINS_STORE) or
+               str(Path(pins._store_path()).name))
+
+    # L. pin creation (kernel blob rides the record on save)
+    seed([])
+    out.append(catch(lambda: pins.pin("néw topic", "q?", "outcome!", "ctx")))
+    first = json.loads(Path(pins._store_path()).read_text())[0]
+    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("")))
+
+    # N. handle_hook (surface intact)
+    out.append(catch(lambda: pins.handle_hook("on_message", {"message": "hi"})))
+    out.append(catch(lambda: pins.handle_hook("on_tick")))
+
+    return out
+
+
+if __name__ == "__main__":
+    print(json.dumps(run_scenarios()))
+'''
+
+
+def run_mode(mode: str, src_dir: Path) -> subprocess.CompletedProcess:
+    env = dict(os.environ)
+    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")
+    )
+    return subprocess.run(
+        [sys.executable, "-c", RUNNER], capture_output=True, text=True, env=env,
+        timeout=300,
+    )
+
+
+def truth_rows() -> list:
+    return json.loads((CRATE / "src" / "frozen_pins_truth.json").read_text())["cases"]
+
+
+def ffi_replay(rows: list) -> tuple[int, list[str]]:
+    """Every frozen row replayed through the FFI; crash channels raise."""
+    import importlib
+
+    sys.path.insert(0, str(REPO / "src"))
+    from rune import cc_loader
+
+    if not cc_loader.available():
+        raise RuntimeError("cognition_core .so missing under CC_RUST_LIB")
+    ok = 0
+    errors: list[str] = []
+    for i, case in enumerate(rows):
+        fn = case["fn"]
+        args = case.get("args", [])
+        kwargs = case.get("kwargs", {})
+        expected = case["expected"]
+        label = f"row {i}: {fn}"
+        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:
+                errors.append(f"{label}: unexpected crash {exc!r}")
+                continue
+            if type(exc).__name__ == etype and str(exc) == emsg:
+                ok += 1
+                continue
+            errors.append(f"{label}: crash {type(exc).__name__}: {exc} != {etype}: {emsg}")
+            continue
+        if "py_error_type" in json.dumps(expected):
+            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}")
+    return ok, errors
+
+
+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"),
+        )
+    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", ""
+            )
+        )
+        return out
+    if fn == "pn_freeze_unpin":
+        records, topic = args[0], args[1]
+        reason = kwargs.get("reason", "")
+        out = cc_loader.pins_unpin(records, topic, reason)
+        err = out.get("__py_error__")
+        if err:
+            if err["type"] == "TypeError":
+                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"])}
+    if fn == "pn_freeze_list":
+        records = args[0]
+        open_only = kwargs.get("open_only", False)
+        limit = kwargs.get("limit", 30)
+        out = cc_loader.pins_list(records, open_only, limit)
+        err = out.get("__py_error__")
+        if err:
+            if err["type"] == "AttributeError":
+                raise AttributeError(err["msg"])
+            raise TypeError(err["msg"])
+        return {"rows": out["rows"]}
+    raise ValueError(f"unknown row fn {fn}")
+
+
+def _dump(records: list) -> str:
+    import json
+
+    return json.dumps(records, indent=2)
+
+
+def main() -> int:
+    live = run_mode("auto", REPO / "src")
+    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 ""),
+        )
+    )
+    if live.returncode or pinned.returncode:
+        print("PARITY RUN FAILED: live rc", live.returncode, "pinned rc", pinned.returncode)
+        return 1
+    a = json.loads(live.stdout)
+    b = json.loads(pinned.stdout)
+    failures = []
+    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}")
+    rows = truth_rows()
+    ok, kernel_errors = ffi_replay(rows)
+    print(f"facade scenarios: {total}, mismatches: {len(failures)}")
+    for f in failures:
+        print("  FAIL", f[:220])
+    print(f"kernel truth rows: {ok}/{len(rows)} ok; mismatches: {len(kernel_errors)}")
+    for e in kernel_errors:
+        print("  FAIL", e[:220])
+    verdict = not failures and not kernel_errors
+    print("PARITY 100%" if verdict else "PARITY FAILURES")
+    return 0 if verdict else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/gen_pins_truth.py

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+"""gen_pins_truth.py — factory wave GAP truth for pins.
+
+Freezes the oracle behavior of the 4 audit-gap functions (dichotomy wave
+#2479: _store_path, _detail_blob, unpin, list_pins) through the SHARED
+harness (scripts/freeze_truth.py — one law).
+
+The store-mutating surfaces run through the freeze rig
+(scripts/pn_freeze_seams/sitecustomize.py — PYTHONPATH-imported by BOTH
+this generator and --verify): pinned floor (RUNE_COGNITION_CORE=python),
+hermetic per-row scratch store seeded exactly as the oracle _save writes
+it, frozen RUNE_HOME set BEFORE the oracle import (PINS_STORE is an
+import-time derivation — pin_home law), and machine-path sentinelization
+(__PN_FREEZE__ — no absolute paths ride frozen truth).
+
+Run (from the repo root):
+    RUNE_COGNITION_CORE=python PYTHONPATH=scripts/pn_freeze_seams:src \
+      python3 scripts/gen_pins_truth.py
+
+Verify (byte-identical, same rig):
+    PYTHONPATH=scripts/pn_freeze_seams:src python3 scripts/freeze_truth.py \
+      --verify src/rust/cognition_core/src/frozen_pins_truth.json \
+      --module rune.pins
+
+Case map (every gap branch):
+  * _detail_blob          — defaults / reason+source kwargs / unicode +
+                            astral + control escape / empty / non-str
+                            scalar values (int/bool/None/0.5-float).
+  * pn_freeze_store_path  — plain / trailing-slash / empty / root / double
+                            slash / nested-double-slash / relative.
+  * pn_freeze_canonical_store_path — the REAL _store_path() (frozen home,
+                            sentinelized prefix — the import-time derivation, captured before the seam rebind).
+  * pn_freeze_unpin       — found+reason / default reason / not-found /
+                            duplicate topic (most recent open wins) /
+                            resolved-only / junk blob / empty blob /
+                            missing / None details / extra keys re-sorted /
+                            "null"-int-list-str-float-bool blob crash rows /
+                            non-str detail VALUES (json.loads TypeError) /
+                            int + unicode reason / int topic.
+  * pn_freeze_list        — empty store / full board (reversal) /
+                            open_only filter / limit truncation (2, 0, 1,
+                            99) / bool-as-int limit / junk + empty +
+                            "{}" + partial + non-str values / "null"-int
+                            -list-str blob AttributeError crash rows /
+                            negative limit (-1, -99) / non-integral limit
+                            TypeError / unicode board.
+"""
+
+import json
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "scripts"))
+
+from freeze_truth import freeze, write_envelope  # noqa: E402
+
+NOW = "2026-10-13T12:00:00.123456+00:00"
+
+
+def rec(iid, state, blob, i=0):
+    """One canonical PinRecord dict (the _load() surface the kernel takes)."""
+    return {
+        "schema_version": 1,
+        "kind": "pin",
+        "item_id": iid,
+        "owner": "pins",
+        "state": state,
+        "evidence_ref": "",
+        "details": blob,
+    }
+
+
+def rec_raw(iid, details, extra=None):
+    """A record whose `details` is a NON-STRING value (corrupt-store class)."""
+    r = rec(iid, "open", None)
+    r["details"] = details
+    return r
+
+
+def blob(question, outcome, context, reason="", source="session"):
+    return json.dumps(
+        {
+            "question": question,
+            "outcome": outcome,
+            "context": context,
+            "reason": reason,
+            "source": source,
+        },
+        sort_keys=True,
+    )
+
+
+B1 = blob("q1", "o1", "c1")
+B2 = blob("q2", "o2", "shared ctx")
+B3 = blob("q3", "o3", "resolved ctx", reason="r3", source="wave")
+
+cases = [
+    # ── _detail_blob (the store blob is a BYTE contract: separators,
+    #    sort_keys, ensure_ascii) ──────────────────────────────────────────
+    ("_detail_blob", ("who?", "fixed", "ctx"), {}),
+    ("_detail_blob", ("who?", "fixed", "ctx"), {"reason": "r2", "source": "wave"}),
+    ("_detail_blob", ("qué?", "out\U0001f600", "c<>&"), {}),
+    ("_detail_blob", ('d"q', "b\\", "n\nl\tt"), {}),
+    ("_detail_blob", ("", "", ""), {}),
+    ("_detail_blob", ("q", "o", "c"), {"reason": "qué \U0001f600", "source": "éwave"}),
+    # non-str scalar payloads: raw (the blob serializes them verbatim)
+    ("_detail_blob", ("q", 3, "c", True, "session"), {}),
+    ("_detail_blob", ("q", 0.5, "c", None, None), {}),
+    # ── _store_path derivation (sentinelized; kernel replays the join) ──
+    ("pn_freeze_store_path", ("__MEM__",), {}),
+    ("pn_freeze_store_path", ("__MEM__/",), {}),
+    ("pn_freeze_store_path", ("",), {}),
+    ("pn_freeze_store_path", ("/",), {}),
+    ("pn_freeze_store_path", ("a//b//",), {}),
+    ("pn_freeze_store_path", ("/a//",), {}),
+    ("pn_freeze_store_path", ("rel",), {}),
+    ("pn_freeze_canonical_store_path", (), {}),
+    # ── pins_unpin: kernel contract = updated records + found flag;
+    #    the seeded bytes come back as post-call store text ────────────────
+    # found resolution: state flip open → resolved, reason merged, blob
+    # re-encoded (sort_keys, ", "/": " separators).
+    ("pn_freeze_unpin", ([rec("t1", "open", B1)], "t1"), {"reason": "done"}),
+    # default reason "" still merges ({"reason": ""} re-encoded).
+    ("pn_freeze_unpin", ([rec("t1", "open", B1)], "t1"), {}),
+    # not-found: store unchanged (kernel must return the records untouched).
+    ("pn_freeze_unpin", ([rec("absent", "open", B1)], "absent"), {"reason": "why"}),
+    # duplicate topic: only the MOST RECENT open pin flips (reverse scan).
+    (
+        "pn_freeze_unpin",
+        ([rec("dup", "open", B1), rec("dup", "open", B2)], "dup"),
+        {"reason": "rr"},
+    ),
+    # a resolved pin is never re-touched: open-only scan.
+    (
+        "pn_freeze_unpin",
+        ([rec("t1", "open", B1), rec("gone", "resolved", B2)], "gone"),
+        {"reason": "again"},
+    ),
+    # corrupt blobs fold to {} then merge the reason.
+    ("pn_freeze_unpin", ([rec("j", "open", "junk{")], "j"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec("e", "open", "")], "e"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec("m", "open", None)], "m"), {"reason": "r"}),
+    (
+        "pn_freeze_unpin",
+        ([rec("x", "open", json.dumps({"zz": "last", "aa": "first"}))], "x"),
+        {"reason": "r"},
+    ),
+    # crash-parity rows: string blobs parsing to non-dicts raise the exact
+    # Python TypeError (store unchanged).
+    ("pn_freeze_unpin", ([rec("n", "open", "null")], "n"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec("i", "open", "5")], "i"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec("l", "open", "[1,2]")], "l"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec("s", "open", '"abc"')], "s"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec("f", "open", "5.5")], "f"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec("b", "open", "true")], "b"), {"reason": "r"}),
+    # non-str detail VALUES: json.loads TypeError (uncatched in unpin).
+    ("pn_freeze_unpin", ([rec_raw("v", 5)], "v"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec_raw("b2", True)], "b2"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec_raw("a2", [1])], "a2"), {"reason": "r"}),
+    ("pn_freeze_unpin", ([rec_raw("d2", {"a": 1})], "d2"), {"reason": "r"}),
+    # raw reason values ride the blob verbatim.
+    ("pn_freeze_unpin", ([rec("r3i", "open", B1)], "r3i"), {"reason": 3}),
+    ("pn_freeze_unpin", ([rec("r3u", "open", B1)], "r3u"), {"reason": "qué \U0001f600"}),
+    # int topic (non-str): not-found decision + envelope built Python-side.
+    ("pn_freeze_unpin", ([rec("t1", "open", B1)], 5), {}),
+    # ── pins_list: kernel contract = projected rows (project → filter →
+    #    reverse → slice); the FULL projection surface ─────────────────────
+    ("pn_freeze_list", ([],), {}),
+    (
+        "pn_freeze_list",
+        (
+            [
+                rec("t1", "open", B1),
+                rec("t2", "open", B2),
+                rec("t3", "resolved", B3),
+                rec("t4", "open", blob("q4", "o4", "t4 only", reason="r4")),
+            ],
+        ),
+        {},
+    ),
+    (
+        "pn_freeze_list",
+        ([rec("t1", "open", B1), rec("t2", "open", B2), rec("t3", "resolved", B3)],),
+        {"open_only": True, "limit": 2},
+    ),
+    (
+        "pn_freeze_list",
+        ([rec("t1", "open", B1), rec("t2", "open", B2), rec("t3", "resolved", B3)],),
+        {"limit": 0},
+    ),
+    (
+        "pn_freeze_list",
+        ([rec("t1", "open", B1), rec("t2", "open", B2), rec("t3", "resolved", B3)],),
+        {"limit": 1},
+    ),
+    (
+        "pn_freeze_list",
+        ([rec("t1", "open", B1), rec("t2", "open", B2), rec("t3", "resolved", B3)],),
+        {"limit": 99},
+    ),
+    # bool-as-int limit ([:True] == [:1] — Python bool subclass quirk).
+    ("pn_freeze_list", ([rec("t1", "open", B1), rec("t2", "open", B2)],), {"limit": True}),
+    # negative limits: slice-from-tail semantics.
+    ("pn_freeze_list", ([rec("t1", "open", B1), rec("t2", "open", B2), rec("t3", "open", B3)],), {"limit": -1}),
+    ("pn_freeze_list", ([rec("t1", "open", B1), rec("t2", "open", B2), rec("t3", "open", B3)],), {"limit": -99}),
+    # non-integral limit → TypeError (slice indices message).
+    ("pn_freeze_list", ([rec("t1", "open", B1)],), {"limit": 1.5}),
+    # corrupt / empty / partial blobs project with defaults.
+    ("pn_freeze_list", ([rec("j", "open", "junk{")],), {}),
+    ("pn_freeze_list", ([rec("e", "open", "")],), {}),
+    ("pn_freeze_list", ([rec("o", "open", "{}")],), {}),
+    ("pn_freeze_list", ([rec("p", "open", '{"reason": "only-reason"}')],), {}),
+    ("pn_freeze_list", ([rec_raw("v", 5)],), {}),
+    ("pn_freeze_list", ([rec_raw("d", {})],), {}),
+    ("pn_freeze_list", ([rec_raw("w", None)],), {}),
+    # crash rows: string blobs parsing to non-dicts → .get AttributeError.
+    ("pn_freeze_list", ([rec("n", "open", "null")],), {}),
+    ("pn_freeze_list", ([rec("i", "open", "5")],), {}),
+    ("pn_freeze_list", ([rec("l", "open", "[1]")],), {}),
+    ("pn_freeze_list", ([rec("s", "open", '"abc"')],), {}),
+    # unicode contents + state mix on one board.
+    (
+        "pn_freeze_list",
+        (
+            [
+                rec("té", "open", blob("qué", "o\U0001f600", "c")),
+                rec("t2", "resolved", blob("q", "o", "shared – dash")),
+                rec("t3", "open", B2),
+            ],
+        ),
+        {},
+    ),
+]
+
+truth = freeze("rune.pins", cases, oracle_module="rune.pins")
+dest = REPO / "src" / "rust" / "cognition_core" / "src" / "frozen_pins_truth.json"
+write_envelope(NOW, truth, dest)
+print(f"wrote {len(truth)} cases -> {dest}")

scripts/pn_freeze_seams/sitecustomize.py

@@ -0,0 +1,108 @@
+"""sitecustomize — the pins freeze rig (dichotomy wave gap-close).
+
+Auto-imported via PYTHONPATH=scripts/pn_freeze_seams:src by BOTH the truth
+generator and `freeze_truth.py --verify`, so the --verify law (replay
+against the pristine oracle) still holds: the seams only make the oracle's
+already-existing behavior reachable with JSON-safe shapes.
+
+  * RUNE_COGNITION_CORE=python pinned (the oracle floor; the harness's
+    _pin_oracle_cores does this too — belt and braces).
+  * RUNE_HOME frozen to a hermetic scratch home BEFORE rune.pins imports
+    (expanduser-at-import quirk: PINS_STORE is computed at import time, so
+    the env must predate the import — pin_home law).
+  * The store path is rebound onto a hermetic fixed-path scratch
+    (/tmp/pn_gap_freeze — wiped per seam call, so rows are self-contained
+    and unpin's mutation can never leak into the next row).
+  * pn_freeze_store_path parameterizes the oracle's own derivation
+    expression (os.path.join(memory_dir, "store", "pins.json") — the exact
+    code PINS_STORE evaluates at import) with sentinel memory_dir values,
+    so no machine path rides frozen truth; pn_freeze_canonical_store_path
+    exercises the REAL pins._store_path() under the frozen home and
+    substitutes its absolute prefix for the __PIN_HOME__ sentinel.
+  * pn_freeze_unpin / pn_freeze_list seed the scratch store from the row
+    args, drive the REAL oracle fns, and echo the result (or the exact
+    exception type+message — crash-parity law) plus the post-call store
+    text the oracle wrote.
+"""
+
+import json
+import os
+import shutil
+import sys
+from pathlib import Path
+
+REPO = Path(os.environ.get("PN_FREEZE_REPO", "") or os.getcwd())
+os.environ["RUNE_COGNITION_CORE"] = "python"
+
+SCRATCH = "/tmp/pn_gap_freeze"
+if os.path.isdir(SCRATCH):
+    shutil.rmtree(SCRATCH)
+os.makedirs(os.path.join(SCRATCH, "pin-home", "memory"), exist_ok=True)
+os.environ["RUNE_HOME"] = os.path.join(SCRATCH, "pin-home")
+
+if str(REPO / "src") not in sys.path:
+    sys.path.insert(0, str(REPO / "src"))
+
+import rune.pins as _pins  # noqa: E402
+
+# The IMPORT-TIME derivation (PINS_STORE as the oracle computes it from the
+# frozen RUNE_HOME) — captured BEFORE the seam rebind below.
+PINS_IMPORT_PATH = _pins.PINS_STORE
+
+STORE_PATH = os.path.join(SCRATCH, "store", "pins.json")
+_pins.PINS_STORE = STORE_PATH
+os.makedirs(os.path.dirname(STORE_PATH), exist_ok=True)
+
+
+def _seed(records):
+    """Reseed the scratch store exactly as the oracle _save writes it."""
+    _pins._save(records)
+
+
+def _store_text():
+    return Path(_pins._store_path()).read_text()
+
+
+def pn_freeze_unpin(records, topic, reason=""):
+    """Drive the REAL oracle unpin against a row-private store."""
+    _seed(records)
+    out = {"result": None}
+    try:
+        result = _pins.unpin(topic, reason=reason)
+        out["result"] = result
+    except Exception as exc:  # noqa: BLE001 — crash-parity capture
+        out["result"] = {"py_error_type": type(exc).__name__, "py_error_msg": str(exc)}
+    out["store"] = _store_text()
+    return out
+
+
+def pn_freeze_list(records, open_only=False, limit=30):
+    """Drive the REAL oracle list_pins against a row-private store."""
+    _seed(records)
+    try:
+        rows = _pins.list_pins(open_only=open_only, limit=limit)
+    except Exception as exc:  # noqa: BLE001 — crash-parity capture
+        return {"py_error_type": type(exc).__name__, "py_error_msg": str(exc)}
+    return {"rows": rows}
+
+
+def pn_freeze_store_path(memory_dir):
+    """The oracle's own PINS_STORE derivation, parameterized on memory_dir
+    (sentinel values keep absolute paths out of frozen truth)."""
+    return os.path.join(memory_dir, "store", "pins.json")
+
+
+def pn_freeze_canonical_store_path():
+    """The oracle's REAL import-time PINS_STORE derivation (the frozen-home
+    computation that ran when rune.pins imported), with the machine prefix
+    substituted for the __PN_FREEZE__ sentinel."""
+    raw = str(PINS_IMPORT_PATH)
+    return raw.replace(SCRATCH, "__PN_FREEZE__")
+
+
+# Binding law: the harness resolves seam names as oracle-module attributes
+# (import must predate freeze(), so sys.modules["rune.pins"] carries them).
+_pins.pn_freeze_unpin = pn_freeze_unpin
+_pins.pn_freeze_list = pn_freeze_list
+_pins.pn_freeze_store_path = pn_freeze_store_path
+_pins.pn_freeze_canonical_store_path = pn_freeze_canonical_store_path

src/rune/cc_loader.py

@@ -77,7 +77,9 @@ def _load() -> ctypes.CDLL:
                 "cc_announcement_patterns", "cc_pronoun_patterns",
                 "cc_hedge_patterns", "cc_sycophancy_patterns",
                 "cc_pattern_triples", "cc_cc_core", "cc_check_genuineness",
-                "cc_identity_handle_hook"):
+                "cc_identity_handle_hook",
+                "cc_pins_store_path", "cc_pins_detail_blob",
+                "cc_pins_unpin", "cc_pins_list"):
         fn = getattr(lib, sym)
         fn.argtypes = [ctypes.c_char_p]
         fn.restype = ctypes.c_void_p
@@ -270,6 +272,36 @@ def identity_handle_hook(event: str) -> dict:
 def pins_recall(pins: list, term: str, limit: int = 4) -> list:
     return _call("cc_pins_recall", {"pins": pins, "term": term, "limit": limit})
 
+
+# ── pins gap-close kernels (dichotomy wave: _store_path, _detail_blob,
+#    unpin, list_pins) — crash channels ride {"__py_error__": {type, msg}} ──
+
+def pins_store_path(memory_dir: str) -> str:
+    """The canonical store join (the PINS_STORE import-time derivation)."""
+    return str(_call("cc_pins_store_path", {"memory_dir": memory_dir}))
+
+
+def pins_detail_blob(question, outcome, context, reason="", source="session") -> str:
+    """The details blob bytes (json.dumps sort_keys contract)."""
+    return str(_call("cc_pins_detail_blob", {"question": question,
+                                             "outcome": outcome,
+                                             "context": context,
+                                             "reason": reason,
+                                             "source": source}))
+
+
+def pins_unpin(records: list, topic, reason="") -> dict:
+    """Unpin kernel: updated records + found flag (the store IO stays in
+    the facade); unresolved pins return the records untouched."""
+    return _call("cc_pins_unpin", {"records": records, "topic": topic,
+                                   "reason": reason})
+
+
+def pins_list(records: list, open_only=False, limit=30) -> dict:
+    """The pinboard projection (project → filter → reverse → slice)."""
+    return _call("cc_pins_list", {"records": records, "open_only": open_only,
+                                  "limit": limit})
+
 def liveness_canary(last: float, now: float, stall_limit: float = 300.0) -> dict:
     return _call("cc_liveness_canary", {"last": last, "now": now,
                                         "stall_limit": stall_limit})

src/rune/pins.py

@@ -54,6 +54,11 @@ PINS_STORE = os.path.join(memory_env.memory_dir(), "store", "pins.json")
 
 def _store_path():
     """Return the canonical pins store path (configurable for tests)."""
+    from rune.identity_anchors import _cc_core
+
+    core = _cc_core()
+    if core is not None:
+        return Path(core.pins_store_path(str(memory_env.memory_dir())))
     return Path(PINS_STORE)
 
 
@@ -76,6 +81,11 @@ def _save(records):
 
 def _detail_blob(question, outcome, context, reason="", source="session"):
     """Encode the pin's detail payload as the canonical record's details blob."""
+    from rune.identity_anchors import _cc_core
+
+    core = _cc_core()
+    if core is not None:
+        return core.pins_detail_blob(question, outcome, context, reason, source)
     return json.dumps(
         {
             "question": question,
@@ -135,7 +145,19 @@ def pin(topic, question, outcome, context="", source="session"):
 
 def unpin(topic, reason=""):
     """Resolve a pin in place (status open -> resolved, journaled reason)."""
+    from rune.identity_anchors import _cc_core
+
     records = _load()
+    core = _cc_core()
+    if core is not None:
+        out = core.pins_unpin(records, topic, reason)
+        crash = out.get("__py_error__")
+        if crash:
+            raise TypeError(crash.get("msg", ""))
+        if out["found"]:
+            _save(out["records"])
+            return {"unpinned": True, "topic": topic, "reason": reason}
+        return {"unpinned": False, "error": f"no open pin: {topic}"}
     for i in range(len(records) - 1, -1, -1):  # most recent open pin wins
         rec = records[i]
         if rec.get("item_id") == topic and rec.get("state") == "open":
@@ -182,6 +204,22 @@ def _py_pin_recall(recs, term, limit=4):
 
 def list_pins(open_only=False, limit=30):
     """The pinboard: all pins (open + resolved) by default, newest first."""
+    from rune.identity_anchors import _cc_core
+
+    core = _cc_core()
+    if core is not None:
+        if limit is not None and not isinstance(limit, (bool, int)):
+            raise TypeError(
+                "slice indices must be integers or None or have an __index__ method"
+            )
+        out = core.pins_list(_load(), open_only, limit)
+        crash = out.get("__py_error__")
+        if crash:
+            etype = {"TypeError": TypeError, "AttributeError": AttributeError}[
+                crash.get("type", "AttributeError")
+            ]
+            raise etype(crash.get("msg", ""))
+        return out["rows"]
     out = [_project(rec) for rec in _load()]
     if open_only:
         out = [p for p in out if p["status"] == "open"]

src/rust/cognition_core/src/broker.rs

@@ -798,6 +798,293 @@ pub fn pins_recall(pins: &[Json], term: &str, limit: usize) -> Vec<Json> {
     out
 }
 
+// ── pins gap-close kernels (dichotomy wave: _store_path, _detail_blob,
+//    unpin, list_pins) — the store IO itself stays in the facade
+//    (AtomicStore); the kernels are pure over the parsed record rows. ──────
+
+/// Python truthiness of a decoded JSON value (dict/set ordering quirks
+/// aside: `if open_only:` and `if blob:` are value truthiness).
+fn py_falsy(v: &Json) -> bool {
+    match v {
+        Json::Null => true,
+        Json::Bool(b) => !*b,
+        Json::Num(n) => *n == 0.0,
+        Json::Str(s) => s.is_empty(),
+        Json::Arr(a) => a.is_empty(),
+        Json::Obj(o) => o.is_empty(),
+    }
+}
+
+/// 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 {
+    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(),
+    }
+}
+
+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();
+        }
+    }
+    py_json_type_name(v)
+}
+
+fn py_error_json(type_name: &str, msg: &str) -> Json {
+    Json::Obj(vec![(
+        "__py_error__".to_string(),
+        Json::Obj(vec![
+            ("type".to_string(), Json::Str(type_name.to_string())),
+            ("msg".to_string(), Json::Str(msg.to_string())),
+        ]),
+    )])
+}
+
+/// json.loads on a non-str value: "the JSON object must be str, bytes or
+/// bytearray, not {T}" (unpin catches only JSONDecodeError — TypeError
+/// escapes).
+fn loads_non_str_error(v: &Json) -> Json {
+    py_error_json(
+        "TypeError",
+        &format!(
+            "the JSON object must be str, bytes or bytearray, not {}",
+            py_json_type_name(v)
+        ),
+    )
+}
+
+/// 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() {
+        "list" => "list indices must be integers or slices, not str".to_string(),
+        other => format!("'{other}' object does not support item assignment"),
+    };
+    py_error_json("TypeError", &msg)
+}
+
+/// _store_path: the canonical store join — os.path.join(memory_dir,
+/// "store", "pins.json") (posixpath semantics; the import-time PINS_STORE
+/// derivation with the memory_dir seam freed).
+pub fn pins_store_path(memory_dir: &str) -> String {
+    let mut path = memory_dir.to_string();
+    for comp in ["store", "pins.json"] {
+        if path.is_empty() || path.ends_with('/') {
+            path.push_str(comp);
+        } else {
+            path.push('/');
+            path.push_str(comp);
+        }
+    }
+    path
+}
+
+/// _detail_blob: the pin's detail payload as the canonical record's
+/// `details` blob — json.dumps({...}, sort_keys=True) bytes.
+pub fn pins_detail_blob(
+    question: &Json,
+    outcome: &Json,
+    context: &Json,
+    reason: &Json,
+    source: &Json,
+) -> String {
+    let obj = Json::Obj(vec![
+        ("context".to_string(), context.clone()),
+        ("outcome".to_string(), outcome.clone()),
+        ("question".to_string(), question.clone()),
+        ("reason".to_string(), reason.clone()),
+        ("source".to_string(), source.clone()),
+    ]);
+    pyjson::dumps_sorted(&obj)
+}
+
+/// In-place `obj[key] = value` (replace keeps position; append new keys —
+/// position moot where the write is re-serialized sorted).
+fn obj_upsert(v: &mut Json, key: &str, val: Json) {
+    if let Json::Obj(pairs) = v {
+        for pair in pairs.iter_mut() {
+            if pair.0 == key {
+                pair.1 = val;
+                return;
+            }
+        }
+        pairs.push((key.to_string(), val));
+    }
+}
+
+/// The details state for one unpin resolve: Ok(parsed-or-default obj) or
+/// Err(the oracle's TypeError channel). Falsy blobs ("" None) take the
+/// "{}" default; truthy non-str values raise "the JSON object must be
+/// str..."; junk folds to {}; parsed non-dicts raise the item-assignment
+/// message for the parsed class.
+fn unpin_details_state(details: Option<&Json>) -> Result<Json, Json> {
+    let empty_blob = Json::Str("{}".to_string());
+    let blob = details.unwrap_or(&empty_blob);
+    if py_falsy(blob) {
+        return Ok(Json::Obj(vec![]));
+    }
+    let text = match blob {
+        Json::Str(s) => s.clone(),
+        other => return Err(loads_non_str_error(other)),
+    };
+    match pyjson::parse(&text) {
+        Ok(Json::Obj(pairs)) => Ok(Json::Obj(pairs)),
+        Ok(other) => Err(item_assign_error(&other, &text)),
+        Err(_) => Ok(Json::Obj(vec![])),
+    }
+}
+
+/// unpin: resolve a pin in place — most recent open pin wins; the records
+/// return updated (state flip + reason merge, the blob re-encoded sorted)
+/// when found; the oracle's TypeError crash channels ride the
+/// __py_error__ envelope (store untouched by construction).
+pub fn pins_unpin(records: &[Json], topic: &Json, reason: &Json) -> Json {
+    let mut idx: Option<usize> = None;
+    for (i, rec) in records.iter().enumerate().rev() {
+        if matches!(get(rec, "item_id"), Some(t) if t == topic)
+            && matches!(get(rec, "state"), Some(Json::Str(s)) if s == "open")
+        {
+            idx = Some(i);
+            break;
+        }
+    }
+    let i = match idx {
+        Some(i) => i,
+        None => {
+            return Json::Obj(vec![
+                ("records".to_string(), Json::Arr(records.to_vec())),
+                ("found".to_string(), Json::Bool(false)),
+            ])
+        }
+    };
+    let details = match unpin_details_state(get(&records[i], "details")) {
+        Ok(d) => d,
+        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 updated[i], "state", Json::Str("resolved".to_string()));
+    obj_upsert(
+        &mut updated[i],
+        "details",
+        Json::Str(pyjson::dumps_sorted(&details_out)),
+    );
+    Json::Obj(vec![
+        (
+            "records".to_string(),
+            pyjson::Json::Arr(updated),
+        ),
+        ("found".to_string(), Json::Bool(true)),
+    ])
+}
+
+/// _project over one record: detail-parse-or-default, AttributeError crash
+/// for string blobs parsing to non-dicts (the oracle's .get miss), {}
+/// defaults for non-str values (the (JSONDecodeError, TypeError) catch).
+fn project_pin(rec: &Json) -> Result<Json, Json> {
+    let details = {
+        match get(rec, "details") {
+            Some(Json::Str(text)) if !text.is_empty() => match pyjson::parse(text) {
+                Ok(Json::Obj(pairs)) => Json::Obj(pairs),
+                Ok(other) => {
+                    return Err(py_error_json(
+                        "AttributeError",
+                        &format!(
+                            "'{}' object has no attribute 'get'",
+                            py_json_type_name_from_text(&other, text)
+                        ),
+                    ))
+                }
+                Err(_) => Json::Obj(vec![]),
+            },
+            Some(other) if other != &Json::Null && !py_falsy(other) => {
+                // truthy non-str value: json.loads raises TypeError, which
+                // _project catches → the {} default
+                Json::Obj(vec![])
+            }
+            _ => Json::Obj(vec![]),
+        }
+    };
+    let field = |details: &Json, key: &str| -> Json {
+        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())),
+        ),
+        ("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())),
+        ),
+        ("reason".to_string(), field(&details, "reason")),
+    ]))
+}
+
+/// list_pins: the pinboard — project all records, open_only filter, then
+/// reverse, then the Python slice semantics (bool-as-int limit, negative
+/// slice-from-tail, [:None] whole-list; non-integral/non-index limits are
+/// the oracle's slice TypeError).
+pub fn pins_list(records: &[Json], open_only: &Json, limit: &Json) -> Json {
+    let limit_i: Option<i64> = match limit {
+        Json::Null => None,
+        Json::Bool(b) => Some(if *b { 1 } else { 0 }),
+        Json::Num(n) if n.fract() != 0.0 => {
+            return py_error_json(
+                "TypeError",
+                "slice indices must be integers or None or have an __index__ method",
+            )
+        }
+        Json::Num(n) => Some(*n as i64),
+        _ => {
+            return py_error_json(
+                "TypeError",
+                "slice indices must be integers or None or have an __index__ method",
+            )
+        }
+    };
+    let mut out: Vec<Json> = vec![];
+    for rec in records {
+        match project_pin(rec) {
+            Ok(p) => out.push(p),
+            Err(err_doc) => return err_doc,
+        }
+    }
+    if !py_falsy(open_only) {
+        out.retain(|p| matches!(get(p, "status"), Some(Json::Str(s)) if s == "open"));
+    }
+    out.reverse();
+    let take: Vec<Json> = match limit_i {
+        None => out,
+        Some(n) if n >= 0 => out.into_iter().take(n as usize).collect(),
+        Some(n) => {
+            let skip = n.unsigned_abs() as usize;
+            if skip >= out.len() {
+                vec![]
+            } else {
+                out[..out.len() - skip].to_vec()
+            }
+        }
+    };
+    Json::Obj(vec![("rows".to_string(), Json::Arr(take))])
+}
+
 // ── tier_manager kernels (W3.4) ─────────────────────────────────────────────
 
 /// get_tier: thresholds arrive as inputs (hot_days/warm_days) + now (no-clock);

src/rust/cognition_core/src/lib.rs

@@ -12,6 +12,9 @@ mod branch_battery;
 #[cfg(test)]
 mod growth_battery;
 
+#[cfg(test)]
+mod pins_battery;
+
 #[cfg(test)]
 mod safety_battery;
 
@@ -291,6 +294,77 @@ pub extern "C" fn cc_pins_recall(arg: *const c_char) -> *mut c_char {
     envelope(result)
 }
 
+// ── pins gap-close wrappers (dichotomy wave: _store_path, _detail_blob,
+//    unpin, list_pins) — raw-Json passthrough; the __py_error__ crash
+//    channels ride the envelope verbatim for the facade to raise. ──────────
+
+#[no_mangle]
+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))
+            }
+            _ => pyjson::Json::Str(broker::pins_store_path("")),
+        });
+    envelope(result)
+}
+
+#[no_mangle]
+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()))
+        };
+        pyjson::Json::Str(broker::pins_detail_blob(
+            &raw("question"),
+            &raw("outcome"),
+            &raw("context"),
+            &get(&a, "reason")
+                .cloned()
+                .unwrap_or_else(|| pyjson::Json::Str(String::new())),
+            &get(&a, "source")
+                .cloned()
+                .unwrap_or_else(|| pyjson::Json::Str("session".to_string())),
+        ))
+    });
+    envelope(result)
+}
+
+#[no_mangle]
+pub extern "C" fn cc_pins_unpin(arg: *const c_char) -> *mut c_char {
+    let result = read_arg(arg).and_then(|t| arg_json(&t)).map(|a| {
+        let topic = get(&a, "topic").cloned().unwrap_or(pyjson::Json::Null);
+        let reason = get(&a, "reason")
+            .cloned()
+            .unwrap_or_else(|| pyjson::Json::Str(String::new()));
+        let pins: Vec<pyjson::Json> = match get(&a, "records") {
+            Some(pyjson::Json::Arr(x)) => x.clone(),
+            _ => vec![],
+        };
+        broker::pins_unpin(&pins, &topic, &reason)
+    });
+    envelope(result)
+}
+
+#[no_mangle]
+pub extern "C" fn cc_pins_list(arg: *const c_char) -> *mut c_char {
+    let result = read_arg(arg).and_then(|t| arg_json(&t)).map(|a| {
+        let open_only = get(&a, "open_only")
+            .cloned()
+            .unwrap_or(pyjson::Json::Bool(false));
+        // mirror the facade signature default (30)
+        let limit = get(&a, "limit").cloned().unwrap_or(pyjson::Json::Num(30.0));
+        let pins: Vec<pyjson::Json> = match get(&a, "records") {
+            Some(pyjson::Json::Arr(x)) => x.clone(),
+            _ => vec![],
+        };
+        broker::pins_list(&pins, &open_only, &limit)
+    });
+    envelope(result)
+}
+
 #[no_mangle]
 pub extern "C" fn cc_tier_get(arg: *const c_char) -> *mut c_char {
     let result = read_arg(arg).and_then(|t| arg_json(&t)).map(|a| {

src/rust/cognition_core/src/pins_battery.rs

@@ -0,0 +1,327 @@
+//! pins gap-close battery — parity vs the frozen Python-oracle truth,
+//! embedded VERBATIM (never transcribed): frozen_pins_truth.json — the
+//! freeze_truth harness format replayed against the LIVE oracle under the
+//! pn_freeze_seams rig (pinned floor RUNE_COGNITION_CORE=python, hermetic
+//! per-row store, frozen RUNE_HOME import-time derivation, machine paths
+//! sentinelized as __MEM__/__PN_FREEZE__). Re-verifiable byte-identical:
+//!   PYTHONPATH=scripts/pn_freeze_seams:src python3 scripts/freeze_truth.py
+//!     --verify src/rust/cognition_core/src/frozen_pins_truth.json
+//!     --module rune.pins
+//!
+//! Case map: _detail_blob (8 rows), pn_freeze_store_path (7), canonical
+//! store path (1), pn_freeze_unpin (22 — found/not-found/most-recent/
+//! crash channels), pn_freeze_list (22 — projection/filter/slice/crash).
+
+use crate::broker::{pins_detail_blob, pins_list, pins_store_path, pins_unpin};
+use crate::pyjson::{self, Json};
+
+const TRUTH: &str = include_str!("frozen_pins_truth.json");
+
+const STORE_TAIL: &str = "/store/pins.json";
+
+fn j(text: &str) -> Json {
+    pyjson::parse(text).expect("embedded truth must parse")
+}
+
+fn s(v: &Json, key: &str) -> String {
+    match v.get(key) {
+        Some(Json::Str(x)) => x.clone(),
+        _ => String::new(),
+    }
+}
+
+fn cases() -> Vec<Json> {
+    match j(TRUTH).get("cases") {
+        Some(Json::Arr(items)) => items.clone(),
+        _ => panic!("missing cases"),
+    }
+}
+
+/// The unpin/list truth rows carry the store as post-call TEXT (the oracle
+/// _save's exact json.dump(indent=2) bytes) — the battery re-serializes the
+/// kernel's returned records with dump_indent2 and byte-compares (the
+/// growth save_deltas kernel's byte law, now over the pins store).
+fn store_bytes(records: &[Json]) -> String {
+    pyjson::dump_indent2(&Json::Arr(records.to_vec()))
+}
+
+fn crash_checked(got: &Json, expected: &Json, expected_result: &Json, label: &str) {
+    match got.get("__py_error__") {
+        Some(err) => {
+            assert_eq!(
+                s(err, "type"),
+                s(expected_result, "py_error_type"),
+                "{label}: crash type diverged"
+            );
+            assert_eq!(
+                s(err, "msg"),
+                s(expected_result, "py_error_msg"),
+                "{label}: crash message diverged"
+            );
+        }
+        _ => panic!("{label}: kernel resolved where the oracle raised"),
+    }
+    let _ = expected;
+}
+
+fn replay_unpin(case: &Json, label: &str) {
+    let args = match case.get("args") {
+        Some(Json::Arr(items)) => items.clone(),
+        _ => panic!("{label}: missing args"),
+    };
+    let records_arg = match args.first() {
+        Some(Json::Arr(x)) => x.clone(),
+        _ => panic!("{label}: rows arg"),
+    };
+    let topic = args.get(1).cloned().unwrap_or(Json::Str(String::new()));
+    let reason = match case.get("kwargs").and_then(|k| k.get("reason")) {
+        Some(r) => r.clone(),
+        _ => 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![]));
+    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
+        // replaying the INPUT records through the same serializer.
+        let got = pins_unpin(&records_arg, &topic, &reason);
+        crash_checked(&got, &expected, &expected_result, label);
+        assert_eq!(
+            store_bytes(&records_arg),
+            s(&expected, "store"),
+            "{label}: crashing unpin must leave the store untouched"
+        );
+        return;
+    }
+    let got = pins_unpin(&records_arg, &topic, &reason);
+    let found = matches!(got.get("found"), Some(Json::Bool(true)));
+    assert_eq!(
+        found,
+        matches!(expected_result.get("unpinned"), Some(Json::Bool(true))),
+        "{label}: found flag diverged"
+    );
+    let updated: Vec<Json> = match got.get("records") {
+        Some(Json::Arr(x)) => x.clone(),
+        _ => panic!("{label}: missing updated records"),
+    };
+    assert_eq!(
+        store_bytes(&updated),
+        s(&expected, "store"),
+        "{label}: store bytes diverged"
+    );
+}
+
+fn replay_list(case: &Json, label: &str) {
+    let args = match case.get("args") {
+        Some(Json::Arr(items)) => items.clone(),
+        _ => panic!("{label}: missing args"),
+    };
+    let records_arg = match args.first() {
+        Some(Json::Arr(x)) => x.clone(),
+        _ => panic!("{label}: rows arg"),
+    };
+    let kwargs = case.get("kwargs").cloned().unwrap_or(Json::Obj(vec![]));
+    let open_only = kwargs
+        .get("open_only")
+        .cloned()
+        .unwrap_or(Json::Bool(false));
+    let limit = kwargs.get("limit").cloned().unwrap_or(Json::Num(30.0));
+    let expected = case.get("expected").cloned().unwrap_or(Json::Obj(vec![]));
+    if expected.get("py_error_type").is_some() {
+        let got = pins_list(&records_arg, &open_only, &limit);
+        crash_checked(&got, &expected, &expected, label);
+        return;
+    }
+    let got = pins_list(&records_arg, &open_only, &limit);
+    let rows: Vec<Json> = match got.get("rows") {
+        Some(Json::Arr(x)) => x.clone(),
+        _ => panic!("{label}: missing rows"),
+    };
+    let want: Vec<Json> = match expected.get("rows") {
+        Some(Json::Arr(x)) => x.clone(),
+        _ => panic!("{label}: missing expected rows"),
+    };
+    assert_eq!(&Json::Arr(rows), &Json::Arr(want), "{label}: rows diverged");
+}
+
+fn replay_detail_blob(case: &Json, label: &str) {
+    let args = match case.get("args") {
+        Some(Json::Arr(items)) => items.clone(),
+        _ => panic!("{label}: missing args"),
+    };
+    let raw = |i: usize, default: &str| -> Json {
+        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
+    // reason/source POSITIONALLY (args[3]/args[4]) — kwargs win.
+    let reason = kwargs
+        .get("reason")
+        .cloned()
+        .or_else(|| args.get(3).cloned())
+        .unwrap_or_else(|| Json::Str(String::new()));
+    let source = kwargs
+        .get("source")
+        .cloned()
+        .or_else(|| args.get(4).cloned())
+        .unwrap_or_else(|| Json::Str("session".to_string()));
+    let got = pins_detail_blob(&raw(0, ""), &raw(1, ""), &raw(2, ""), &reason, &source);
+    match case.get("expected") {
+        Some(Json::Str(want)) => assert_eq!(&got, want, "{label}: blob bytes diverged"),
+        other => panic!("{label}: unexpected expected shape {other:?}"),
+    }
+}
+
+fn replay_store_path(case: &Json, label: &str) {
+    let expect = case.get("expected").cloned().unwrap_or(Json::Null);
+    // The canonical row carries `args: []` — only a NON-empty args array is
+    // the parameterized join form; an empty one means the canonical row.
+    let items = match case.get("args") {
+        Some(Json::Arr(items)) if !items.is_empty() => Some(items),
+        _ => None,
+    };
+    if let Some(items) = items {
+        let dir = match items.first() {
+            Some(Json::Str(s)) => s.clone(),
+            _ => String::new(),
+        };
+        assert_eq!(
+            &Json::Str(pins_store_path(&dir)),
+            &expect,
+            "{label}: store path join diverged"
+        );
+        return;
+    }
+    // canonical row: the frozen sentinelized derivation — the kernel
+    // must reproduce it from its memory_dir suffix (join of the parent
+    // of the sentinelized output).
+    let want = match &expect {
+        Json::Str(s) => s.clone(),
+        other => panic!("{label}: unexpected expected shape {other:?}"),
+    };
+    let parent = &want[..want.len() - STORE_TAIL.len()];
+    assert_eq!(
+        pins_store_path(parent),
+        want,
+        "{label}: canonical derivation diverged"
+    );
+}
+
+#[test]
+fn pins_frozen_truth_replay() {
+    let rows = cases();
+    assert!(rows.len() >= 55, "truth row-count drift: {}", rows.len());
+    for (i, case) in rows.iter().enumerate() {
+        let label = format!("[row {i}: {}]", s(case, "fn"));
+        match s(case, "fn").as_str() {
+            "_detail_blob" => replay_detail_blob(case, &label),
+            "pn_freeze_store_path" | "pn_freeze_canonical_store_path" => {
+                replay_store_path(case, &label)
+            }
+            "pn_freeze_unpin" => replay_unpin(case, &label),
+            "pn_freeze_list" => replay_list(case, &label),
+            other => panic!("unknown truth fn: {other}"),
+        }
+    }
+}
+
+#[test]
+fn pins_kernel_units() {
+    // store-path join edge classes (posixpath semantics)
+    assert_eq!(pins_store_path("/home/x"), "/home/x/store/pins.json");
+    assert_eq!(pins_store_path("/home/x/"), "/home/x/store/pins.json");
+    assert_eq!(pins_store_path(""), "store/pins.json");
+    assert_eq!(pins_store_path("/"), "/store/pins.json");
+    assert_eq!(pins_store_path("//"), "//store/pins.json");
+    assert_eq!(pins_store_path("a//b//"), "a//b//store/pins.json");
+    // detail blob: separators + sort_keys + ensure_ascii (byte contract)
+    assert_eq!(
+        pins_detail_blob(
+            &Json::Str("who?".into()),
+            &Json::Str("fixed".into()),
+            &Json::Str("ctx".into()),
+            &Json::Str(String::new()),
+            &Json::Str("session".into())
+        ),
+        "{\"context\": \"ctx\", \"outcome\": \"fixed\", \"question\": \"who?\", \
+         \"reason\": \"\", \"source\": \"session\"}"
+    );
+    assert_eq!(
+        pins_detail_blob(
+            &Json::Str("qué?".into()),
+            &Json::Str("out😀".into()),
+            &Json::Str("c<>&".into()),
+            &Json::Str(String::new()),
+            &Json::Str("session".into())
+        ),
+        "{\"context\": \"c<>&\", \"outcome\": \"out\\ud83d\\ude00\", \
+         \"question\": \"qu\\u00e9?\", \"reason\": \"\", \"source\": \"session\"}"
+    );
+    // unpin: found + not-found + most-recent-wins
+    let rec = Json::Obj(vec![
+        ("kind".into(), Json::Str("pin".into())),
+        ("item_id".into(), Json::Str("t".into())),
+        ("state".into(), Json::Str("open".into())),
+        (
+            "details".into(),
+            Json::Str("{\"question\": \"q\", \"outcome\": \"o\"}".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(),
+        _ => panic!("records"),
+    };
+    let upd_rec = &upd[0];
+    assert_eq!(s(upd_rec, "state"), "resolved");
+    assert_eq!(
+        s(upd_rec, "details"),
+        "{\"outcome\": \"o\", \"question\": \"q\", \"reason\": \"d\"}"
+    );
+    let miss = pins_unpin(&[rec], &Json::Str("nope".into()), &Json::Str("".into()));
+    assert!(matches!(miss.get("found"), Some(Json::Bool(false))));
+    // crash channel: non-dict blob
+    let bad = Json::Obj(vec![
+        ("kind".into(), Json::Str("pin".into())),
+        ("item_id".into(), Json::Str("b".into())),
+        ("state".into(), Json::Str("open".into())),
+        ("details".into(), Json::Str("5".into())),
+    ]);
+    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");
+    // list: projection defaults + open filter + reversal
+    let rows = pins_list(
+        &[Json::Obj(vec![
+            ("item_id".into(), Json::Str("t".into())),
+            ("state".into(), Json::Str("open".into())),
+            ("details".into(), Json::Str("junk{".into())),
+        ])],
+        &Json::Bool(false),
+        &Json::Num(30.0),
+    );
+    let list_rows: Vec<Json> = match rows.get("rows") {
+        Some(Json::Arr(x)) => x.clone(),
+        _ => panic!("rows"),
+    };
+    let first = &list_rows[0];
+    assert_eq!(s(first, "topic"), "t");
+    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_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/pins_messages.out

@@ -0,0 +1,30 @@
+assign_list: TypeError: list indices must be integers or slices, not str
+assign_list_direct: TypeError: list indices must be integers or slices, not str
+unpin_blob_null: TypeError: 'NoneType' object does not support item assignment
+unpin_blob_int: TypeError: 'int' object does not support item assignment
+unpin_blob_float: TypeError: 'float' object does not support item assignment
+unpin_blob_bool: TypeError: 'bool' object does not support item assignment
+unpin_blob_str: TypeError: 'str' object does not support item assignment
+unpin_blob_junk: JSONDecodeError: Expecting property name enclosed in double quotes: line 1 column 2 (char 1)
+unpin_blob_empty: OK blob='{"reason": "r"}'
+unpin_value_int: TypeError: the JSON object must be str, bytes or bytearray, not int
+unpin_value_bool: TypeError: the JSON object must be str, bytes or bytearray, not bool
+unpin_value_list: TypeError: the JSON object must be str, bytes or bytearray, not list
+unpin_value_dict: TypeError: the JSON object must be str, bytes or bytearray, not dict
+unpin_value_noneval: OK 'ok'
+project_blob_null: AttributeError: 'NoneType' object has no attribute 'get'
+project_blob_int: AttributeError: 'int' object has no attribute 'get'
+project_blob_float: AttributeError: 'float' object has no attribute 'get'
+project_blob_bool: AttributeError: 'bool' object has no attribute 'get'
+project_blob_str: AttributeError: 'str' object has no attribute 'get'
+project_blob_arr: AttributeError: 'list' object has no attribute 'get'
+project_value_intval: OK 'MISSING'
+project_value_boolval: OK 'MISSING'
+project_value_arrval: OK 'MISSING'
+project_value_dictval: OK 'MISSING'
+project_value_noneval: OK 'MISSING'
+project_value_emptystr: OK 'MISSING'
+slice_float: TypeError: slice indices must be integers or None or have an __index__ method
+slice_bool_true: OK [1]
+slice_neg: [1, 2] | slice_neg_big: []
+slice_pos_big: [1, 2, 3] | slice_0: []

verification/pins_messages_gen.py

@@ -0,0 +1,80 @@
+#!/usr/bin/env python3
+"""Exception-message probes for the pins kernel contract (crash parity).
+Evidence: verification/pins_messages.out
+"""
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+CHILD = r'''
+import json
+
+def msg(fn):
+    try:
+        v = fn()
+        return f"OK {v!r}"
+    except Exception as e:
+        return f"{type(e).__name__}: {e}"
+
+# item-assignment crashes (unpin: details["reason"] = reason)
+print("assign_list:", msg(lambda: [1, 2].__setitem__("reason", "r")))
+try:
+    d = [1, 2]
+    d["reason"] = "r"
+except Exception as e:
+    print("assign_list_direct:", f"{type(e).__name__}: {e}")
+for label, blob in [("null", "null"), ("int", "5"), ("float", "5.5"),
+                    ("bool", "true"), ("str", '"abc"'), ("junk", "{,"),
+                    ("empty", "")]:
+    try:
+        d = json.loads(blob) if blob else {}
+        d["reason"] = "r"
+        print(f"unpin_blob_{label}:", f"OK blob={json.dumps(d, sort_keys=True)!r}")
+    except Exception as e:
+        print(f"unpin_blob_{label}:", f"{type(e).__name__}: {e}")
+
+# json.loads of non-str VALUES (unpin: no TypeError catch)
+for label, val in [("int", 5), ("bool", True), ("list", [1]), ("dict", {"a": 1}),
+                   ("noneval", None)]:
+    def f(val=val):
+        v = val or "{}"
+        v = json.loads(v)
+        v["reason"] = "r"
+        return "ok"
+    print(f"unpin_value_{label}:", msg(f))
+
+# _project .get crashes: str blob parsing to non-dict
+for label, blob in [("null", "null"), ("int", "5"), ("float", "5.5"),
+                    ("bool", "true"), ("str", '"abc"'), ("arr", "[1,2]")]:
+    def g(blob=blob):
+        try:
+            details = json.loads(blob)
+        except (json.JSONDecodeError, TypeError):
+            details = {}
+        return details.get("question", "")
+    print(f"project_blob_{label}:", msg(g))
+
+# _project with non-str detail VALUES (TypeError caught -> defaults)
+for label, val in [("intval", 5), ("boolval", True), ("arrval", [1]), ("dictval", {"a": 1}), ("noneval", None), ("emptystr", "")]:
+    def h(val=val):
+        blob = val
+        if not blob:
+            details = {}
+        else:
+            try:
+                details = json.loads(blob)
+            except (json.JSONDecodeError, TypeError):
+                details = {}
+        return details.get("question", "MISSING")
+    print(f"project_value_{label}:", msg(h))
+
+# slice-by-float message
+print("slice_float:", msg(lambda: [1, 2, 3][:3.0]))
+print("slice_bool_true:", msg(lambda: [1, 2, 3][:True]))
+print("slice_neg:", [1, 2, 3][:-1], "| slice_neg_big:", [1, 2, 3][:-99])
+print("slice_pos_big:", [1, 2, 3][:99], "| slice_0:", [1, 2, 3][:0])
+'''
+p = subprocess.run([sys.executable, "-c", CHILD], capture_output=True, text=True)
+Path("verification/pins_messages.out").write_text(p.stdout + ("\nSTDERR:\n" + p.stderr if p.stderr else ""))
+print(p.stdout)

verification/pins_probe.out

@@ -0,0 +1,71 @@
+rc: 1
+CHILD-STDOUT:
+blob1 {"context": "ctx", "outcome": "fixed", "question": "who?", "reason": "", "source": "session"}
+blob2 {"context": "ctx", "outcome": "fixed", "question": "who?", "reason": "r2", "source": "wave"}
+blob_unicode {"context": "c<>&", "outcome": "out\ud83d\ude00", "question": "qu\u00e9?", "reason": "", "source": "session"}
+blob_empty {"context": "", "outcome": "", "question": "", "reason": "", "source": "session"}
+pin_result {'pinned': True, 'wing': 'pins', 'topic': 'topic a', 'outcome': 'outcome!'}
+store_after_pin [
+  {
+    "schema_version": 1,
+    "kind": "pin",
+    "item_id": "topic a",
+    "owner": "pins",
+    "state": "open",
+    "evidence_ref": "",
+    "details": "{\"context\": \"ctx here\", \"outcome\": \"outcome!\", \"question\": \"q?\", \"reason\": \"\", \"source\": \"session\"}"
+  }
+]
+unpin_result {'unpinned': True, 'topic': 'topic a', 'reason': 'done'}
+store_after_unpin [
+  {
+    "schema_version": 1,
+    "kind": "pin",
+    "item_id": "topic a",
+    "owner": "pins",
+    "state": "resolved",
+    "evidence_ref": "",
+    "details": "{\"context\": \"ctx here\", \"outcome\": \"outcome!\", \"question\": \"q?\", \"reason\": \"done\", \"source\": \"session\"}"
+  }
+]
+unpin_missing {'unpinned': False, 'error': 'no open pin: nope topic'}
+unpin_missing_reason {'unpinned': False, 'error': 'no open pin: nope topic'}
+unpin_junk {'unpinned': True, 'topic': 'junky', 'reason': 'r'}
+store_after_junk [
+  {
+    "schema_version": 1,
+    "kind": "pin",
+    "item_id": "topic a",
+    "owner": "pins",
+    "state": "resolved",
+    "evidence_ref": "",
+    "details": "{\"context\": \"ctx here\", \"outcome\": \"outcome!\", \"question\": \"q?\", \"reason\": \"done\", \"source\": \"session\"}"
+  },
+  {
+    "schema_version": 1,
+    "kind": "pin",
+    "item_id": "junky",
+    "owner": "pins",
+    "state": "resolved",
+    "evidence_ref": "",
+    "details": "{\"reason\": \"r\"}"
+  }
+]
+unpin_list_exc TypeError: list indices must be integers or slices, not str
+unpin_int_exc TypeError: 'int' object does not support item assignment
+unpin_dup {'unpinned': True, 'topic': 'dup', 'reason': 'rr'}
+dup_states_order ['open', 'resolved']
+dup_last_blob {"context": "c2", "outcome": "o2", "question": "q2", "reason": "rr", "source": "session"}
+
+STDERR:
+Traceback (most recent call last):
+  File "<string>", line 71, in <module>
+    rows = pins.list_pins()
+  File "/mnt/workdrive/rune-wave-ct2/src/rune/pins.py", line 185, in list_pins
+    out = [_project(rec) for rec in _load()]
+           ~~~~~~~~^^^^^
+  File "/mnt/workdrive/rune-wave-ct2/src/rune/pins.py", line 102, in _project
+    "question": details.get("question", ""),
+                ^^^^^^^^^^^
+AttributeError: 'int' object has no attribute 'get'
+

verification/pins_probe2.out

@@ -0,0 +1,27 @@
+CHILD-STDOUT:
+
+STDERR:
+Traceback (most recent call last):
+  File "<string>", line 24, in <module>
+    rows = pins.list_pins()
+  File "/mnt/workdrive/rune-wave-ct2/src/rune/pins.py", line 185, in list_pins
+    out = [_project(rec) for rec in _load()]
+           ~~~~~~~~^^^^^
+  File "/mnt/workdrive/rune-wave-ct2/src/rune/pins.py", line 102, in _project
+    "question": details.get("question", ""),
+                ^^^^^^^^^^^
+AttributeError: 'NoneType' object has no attribute 'get'
+
+JOIN-STDOUT:
+join '/home/x' -> '/home/x/store/pins.json'
+join '/home/x/' -> '/home/x/store/pins.json'
+join '' -> 'store/pins.json'
+join '/' -> '/store/pins.json'
+join '//' -> '//store/pins.json'
+join 'rel' -> 'rel/store/pins.json'
+join 'a//b//' -> 'a//b//store/pins.json'
+join '/a//' -> '/a//store/pins.json'
+join 'x' -> 'x/store/pins.json'
+rune_home_trail: /tmp/ph_mem
+home_trail: /tmp/ph_home/memory
+

verification/pins_probe2_gen.py

@@ -0,0 +1,92 @@
+#!/usr/bin/env python3
+"""Second oracle probe pass: list/recall sections (int-blob record removed
+first) + os.path.join semantics for pins_store_path. Evidence:
+verification/pins_probe2.out
+"""
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+OUT = []
+
+CHILD = r'''
+import json, sys
+sys.path.insert(0, "src")
+from rune import pins
+from pathlib import Path
+import os as _os
+_os.makedirs(pins._store_path().parent, exist_ok=True)
+
+# fresh store
+recs = []
+def rec(iid, state, blob, i=0):
+    return {"schema_version": 1, "kind": "pin", "item_id": iid,
+            "owner": "pins", "state": state, "evidence_ref": "",
+            "details": blob}
+
+recs = [rec("t1", "open", json.dumps({"q": "q1", "o": "o1", "c": "c1", "reason": "", "source": "session"})),
+        rec("t2", "open", json.dumps({"q": "q2", "o": "o2", "c": "shared", "reason": "", "source": "session"})),
+        rec("t3", "resolved", json.dumps({"q": "q3", "o": "o3", "c": "shared", "reason": "r3"})),
+        rec("t4", "open", json.dumps({"q": "q4", "o": "o4", "c": "t4 only"})),
+        rec("t5", "open", "junk{"),
+        rec("t6", "resolved", ""),
+        rec("t7", "open", "null")]
+Path(pins._store_path()).write_text(json.dumps(recs, indent=2))
+rows = pins.list_pins()
+print("list_all:", [r["topic"] for r in rows])
+print("list_all_limit2:", [r["topic"] for r in pins.list_pins(limit=2)])
+print("list_open:", [r["topic"] for r in pins.list_pins(open_only=True)])
+print("list_open_limit2:", [r["topic"] for r in pins.list_pins(open_only=True, limit=2)])
+print("list_limit0:", pins.list_pins(limit=0))
+print("row_t1:", rows[-4] if len(rows) > 4 else None)
+print("row_t3:", [r for r in rows if r["topic"] == "t3"])
+print("row_t5:", [r for r in rows if r["topic"] == "t5"])
+print("row_t6:", [r for r in rows if r["topic"] == "t6"])
+# recall via delegation (python-pinned floor)
+print("recall_shared:", pins.recall("shared"))
+print("recall_t1:", pins.recall("t1"))
+print("recall_case:", pins.recall("SHARED"))
+print("recall_none:", pins.recall(""))
+print("recall_limit:", len(pins.recall("shared", limit=1)))
+# unpin reason unicode + topic not found formatting
+print("unpin_uni:", pins.unpin("t4", reason="qu\u00e9 \U0001f680"))
+print("store_t4_blob:", [r["details"] for r in json.loads(Path(pins._store_path()).read_text()) if r["item_id"] == "t4"])
+# _project row shapes for battery
+print("proj_full:", pins._project(recs[1]))
+'''
+
+env = dict(os.environ)
+env["RUNE_HOME"] = tempfile.mkdtemp(prefix="pins_probe2_")
+env["RUNE_COGNITION_CORE"] = "python"
+p = subprocess.run([sys.executable, "-c", CHILD],
+                   capture_output=True, text=True, env=env,
+                   cwd="/mnt/workdrive/rune-wave-ct2")
+OUT.append("CHILD-STDOUT:\n" + p.stdout)
+if p.stderr:
+    OUT.append("STDERR:\n" + p.stderr[-1500:])
+
+# os.path.join semantics for the store-path kernel
+JOINS = r'''
+import os, sys
+sys.path.insert(0, "src/rune")
+import memory_env
+cases = ["/home/x", "/home/x/", "", "/", "//", "rel", "a//b//", "/a//", "x"]
+for c in cases:
+    print(f"join {c!r} -> {os.path.join(c, 'store', 'pins.json')!r}")
+os.environ["RUNE_HOME"] = "/tmp/ph_home/"
+os.environ["RUNE_MEMORY_DIR"] = "/tmp/ph_mem"
+print("rune_home_trail:", memory_env.memory_dir())
+del os.environ["RUNE_MEMORY_DIR"]
+print("home_trail:", memory_env.memory_dir())
+'''
+p2 = subprocess.run([sys.executable, "-c", JOINS], capture_output=True, text=True, env=dict(os.environ))
+OUT.append("JOIN-STDOUT:\n" + p2.stdout)
+if p2.stderr:
+    OUT.append("JOIN-STDERR:\n" + p2.stderr[-800:])
+
+Path("verification").mkdir(exist_ok=True)
+Path("verification/pins_probe2.out").write_text("\n".join(OUT) + "\n")
+print("\n".join(OUT))

verification/pins_probe_gen.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python3
+"""Oracle probes for the pins gap-close (evidence protocol): exact bytes,
+crash messages, join semantics. Output -> verification/pins_probe.out
+"""
+import json
+import os
+import subprocess
+import sys
+import tempfile
+from pathlib import Path
+
+OUT = []
+
+
+def note(tag, value):
+    OUT.append(f"{tag}: {value!r}")
+
+
+home = tempfile.mkdtemp(prefix="pins_probe_")
+env = dict(os.environ)
+env["RUNE_HOME"] = home
+env["RUNE_COGNITION_CORE"] = "python"
+p = subprocess.run(
+    [sys.executable, "-c", r'''
+import json, os, sys
+sys.path.insert(0, "src")
+from rune import pins
+from pathlib import Path
+
+note = print
+
+# 1. detail blob bytes
+note("blob1", pins._detail_blob("who?", "fixed", "ctx"))
+note("blob2", pins._detail_blob("who?", "fixed", "ctx", reason="r2", source="wave"))
+note("blob_unicode", pins._detail_blob("qu\u00e9?", "out\ud83d\ude00", "c<>&"))
+note("blob_empty", pins._detail_blob("", "", ""))
+
+# 2. pin + store bytes
+res = pins.pin("topic a", "q?", "outcome!", "ctx here")
+note("pin_result", res)
+note("store_after_pin", Path(pins._store_path()).read_text())
+
+# 3. unpin happy path: result + store bytes
+res = pins.unpin("topic a", reason="done")
+note("unpin_result", res)
+note("store_after_unpin", Path(pins._store_path()).read_text())
+
+# 4. unpin not-found
+note("unpin_missing", pins.unpin("nope topic"))
+note("unpin_missing_reason", pins.unpin("nope topic", reason="why"))
+
+# 5. unpin with junk blob
+pins.pin("junky", "q", "o", "c")
+recs = json.loads(Path(pins._store_path()).read_text())
+recs[-1]["details"] = "not json {"
+Path(pins._store_path()).write_text(json.dumps(recs, indent=2))
+note("unpin_junk", pins.unpin("junky", reason="r"))
+note("store_after_junk", Path(pins._store_path()).read_text())
+
+# 6. unpin with list blob (crash?)
+pins.pin("lister", "q", "o", "c")
+recs = json.loads(Path(pins._store_path()).read_text())
+recs[-1]["details"] = "[1, 2]"
+Path(pins._store_path()).write_text(json.dumps(recs, indent=2))
+try:
+    note("unpin_list", pins.unpin("lister", reason="r"))
+except Exception as e:
+    note("unpin_list_exc", f"{type(e).__name__}: {e}")
+
+# 7. unpin with int blob
+recs = json.loads(Path(pins._store_path()).read_text())
+for rec in recs:
+    if rec.get("item_id") == "lister" and rec.get("state") == "open":
+        rec["details"] = "5"
+Path(pins._store_path()).write_text(json.dumps(recs, indent=2))
+try:
+    note("unpin_int", pins.unpin("lister", reason="r"))
+except Exception as e:
+    note("unpin_int_exc", f"{type(e).__name__}: {e}")
+
+# 8. most recent open pin wins: duplicate topics
+pins.pin("dup", "q1", "o1", "c1")
+pins.pin("dup", "q2", "o2", "c2")
+note("unpin_dup", pins.unpin("dup", reason="rr"))
+recs = json.loads(Path(pins._store_path()).read_text())
+states = [r["state"] for r in recs if r.get("item_id") == "dup"]
+note("dup_states_order", states)
+note("dup_last_blob", recs[-1]["details"])
+
+# 9. list_pins projection + order + limits
+pins.pin("alpha", "q", "out-a", "ctx shared")
+pins.pin("beta", "q", "out-b", "beta only")
+pins.unpin("beta", reason="closed-b")
+rows = pins.list_pins()
+note("list_all", [r["topic"] for r in rows])
+note("list_openonly", [r["topic"] for r in pins.list_pins(open_only=True)])
+note("list_limit1", [r["topic"] for r in pins.list_pins(limit=1)])
+note("list_limit0", pins.list_pins(limit=0))
+note("list_row", rows[-1])
+note("list_limit_big", len(pins.list_pins(limit=99)))
+note("list_resolved_row", [r for r in rows if r["topic"] == "beta"][0])
+
+# 10. _project missing keys
+proj = pins._project({"item_id": "x", "state": "open", "details": json.dumps({"reason": "only-reason"})})
+note("project_partial", proj)
+note("project_no_blob", pins._project({"item_id": "y"}))
+note("project_bad_blob", pins._project({"item_id": "z", "details": "junk{"}))
+note("project_none_blob", pins._project({"item_id": "w", "details": None}))
+
+# 11. recall delegation path (kernel absent w/ python pin)
+note("recall_1", pins.recall("shared"))
+note("recall_empty", pins.recall(""))
+
+# 12. handle_hook
+note("hook_msg", pins.handle_hook("on_message", {"message": "hello"}))
+note("hook_other", pins.handle_hook("on_tick"))
+'''],
+    capture_output=True, text=True, env=env, cwd="/mnt/workdrive/rune-wave-ct2")
+note("rc", p.returncode)
+OUT.append("CHILD-STDOUT:\n" + p.stdout)
+if p.stderr:
+    OUT.append("STDERR:\n" + p.stderr[-2000:])
+Path("verification").mkdir(exist_ok=True)
+Path("verification/pins_probe.out").write_text("\n".join(OUT) + "\n")
+print("\n".join(OUT))

verification/pn-build-release.exit

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

verification/pn-build.exit

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

verification/pn-cargo-test.exit

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

verification/pn-clippy.exit

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

verification/pn-freeze-verify.exit

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

verification/pn-gen1.exit

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

verification/pn-gen2.exit

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

verification/pn-parity.exit

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

verification/pn-pytest-live.exit

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

verification/pn-pytest-python.exit

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

verification/pn-pytest-shims-pins.exit

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

verification/pn-pytest-shims.exit

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

verification/pn-summary.md

@@ -0,0 +1,63 @@
+# pins gap-close (dichotomy wave #2479) — verification summary
+
+Module: src/rune/pins.py (oracle 282L/11 fns) → EXTENDS src/rust/cognition_core
+(audit alias). Worktree rune-wave-ct2. Baseline 7/11 PARTIAL → **11/11 FULL
+RUST** (docs/dichotomy-rebase-honest-audit.md regenerated: PARTIAL 51→50,
+FULL 102→103, fn-level 1588→1592).
+
+Ported kernels (broker.rs + lib.rs FFI + cc_loader.py ctypes + pins.py facade
+delegation): `_store_path` (pins_store_path), `_detail_blob` (pins_detail_blob),
+`unpin` (pins_unpin), `list_pins` (pins_list). Store IO (AtomicStore) stays in
+the facade. Crash channels ride {"__py_error__": {type, msg}}.
+
+## Verification ladder evidence (all re-run after Phase B)
+
+| Stage | Command | Exit | Evidence |
+|---|---|---|---|
+| cargo build | `cargo build` | 0 | pn-build.log/.exit |
+| cargo build --release | `cargo build --release` | 0 | pn-build-release.log/.exit |
+| clippy | `cargo clippy --all-targets -- -D warnings` | 0 | pn-clippy.log/.exit |
+| cargo test | `cargo test` | 0 (64 lib + 5 int) | pn-cargo-test.log/.exit |
+| truth gen ×2 | `RUNE_COGNITION_CORE=python PYTHONPATH=scripts/pn_freeze_seams:src python3 scripts/gen_pins_truth.py` | 0, byte-identical | pn-gen1.log pn-gen2.log/.exit |
+| freeze --verify | `PYTHONPATH=scripts/pn_freeze_seams:src python3 scripts/freeze_truth.py --verify src/rust/cognition_core/src/frozen_pins_truth.json --module rune.pins` | 0 (60 cases) | pn-freeze-verify.log/.exit |
+| live parity | `python3 scripts/check_pins_parity.py` (CC_RUST_LIB=…release .so) | 0 — **190/190** both pins | pn-parity.log/.exit |
+| pytest live | `PYTHONPATH=src python3 -m pytest tests/test_legacy_pins.py` | 0 — 5 passed | pn-pytest-live.log/.exit |
+| pytest python pin | `RUNE_COGNITION_CORE=python …` | 2 failed, 3 passed | pn-pytest-python.log/.exit |
+| shims (pins subset) | `PYTHONPATH=src python3 -m pytest tests/test_rust_core_shims.py -k pins` | 0 — 2 passed | pn-pytest-shims-pins.log/.exit |
+| ruff | `ruff check` + `ruff format` on all changed .py | clean | — |
+
+Frozen truth: src/rust/cognition_core/src/frozen_pins_truth.json — 60 rows
+(_detail_blob 8, pn_freeze_store_path 7, canonical 1, pn_freeze_unpin 22,
+pn_freeze_list 22), embedded VERBATIM in pins_battery.rs via include_str!
+(row-count drift assertion ≥55).
+
+Rig: scripts/pn_freeze_seams/sitecustomize.py (pinned floor
+RUNE_COGNITION_CORE=python; RUNE_HOME frozen pre-import — PINS_STORE is an
+import-time derivation; per-row hermetic scratch store wiped per call;
+__PN_FREEZE__ sentinelization).
+
+## Documented divergences (not bugs)
+
+1. **pytest python-pin leg: 2 failures in test_legacy_pins.py** — `recall`
+   raises RuntimeError("cognition_core kernel not built — …") under
+   RUNE_COGNITION_CORE=python because the pins_recall Python oracle was
+   RETIRED 2026-10-10 (strangler exit; delegation-only stub). Pre-existing on
+   the base tree: this port's diff does not touch recall/_py_pin_recall (the
+   failure path is byte-identical at HEAD). Convention per sibling lanes
+   (witness/spine notes: retired-oracle raises under the strict pin are BY
+   DESIGN).
+2. **test_rust_core_shims.py full file: 9 failures** — all other crates'
+   kernels (memory_scoring, spine, construction, …) not built in this
+   worktree; the pins subset (-k pins) passes 2/2. Pre-existing/environmental.
+3. **Generator bug found & fixed during resume:** 8 pn_freeze_list rows in
+   gen_pins_truth.py had the kwargs dict inside the args tuple (oracle was
+   actually called with open_only=<dict>, limit=30). Fixed to real kwargs;
+   truth regenerated ×2 byte-identical; battery replay now matches.
+
+## Phase B (language-advantage pass, second commit)
+
+broker.rs pins block: `py_json_type_name`/`py_json_type_name_from_text`
+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.