chore(release): auto-bump version to 0.271.0 — earned by # (minor: converged release (push-to-master bump))

Promotion record: (unknown)

README.md

@@ -1,6 +1,6 @@
 # Rune
 
-**Current release: v0.270.0**
+**Current release: v0.271.0**
 
 **Rune is a persistent personal agent designed to accumulate experience, develop reusable competence, and remain recognisably the same agent even when its underlying foundation model changes.**
 

docs/dichotomy-rebase-honest-audit.md

@@ -2,14 +2,14 @@
 
 _Regenerate with `python3 scripts/dichotomy_audit.py --write`. A module is FULL only when 80%+ of its Python functions have name-preserving counterparts in the Go/Rust implementation — directories without the functions are facades and count as NOT TOUCHED._
 
-**FULL completion: 28%** (53/192 runtime modules at FULL)
+**FULL completion: 28%** (54/192 runtime modules at FULL)
 
-Function-level detail: 53% (1268/2376 runtime functions matched) — secondary; includes in-flight work
+Function-level detail: 54% (1290/2376 runtime functions matched) — secondary; includes in-flight work
 
 | Status | Modules | Lines | Meaning |
 |---|---|---|---|
-| FULL | 53 | 20189 | 80%+ of functions ported |
-| PARTIAL | 95 | 41837 | 20–80% — real logic exists, gaps named below |
+| FULL | 54 | 21059 | 80%+ of functions ported |
+| PARTIAL | 94 | 40967 | 20–80% — real logic exists, gaps named below |
 | NOT TOUCHED | 44 | 11346 | no name-preserving implementation |
 | Total runtime | 192 | 73372 | |
 
@@ -68,7 +68,6 @@ Function-level detail: 53% (1268/2376 runtime functions matched) — secondary;
 |---|---|---|---|
 | memory_store | 2333 | 30/78 | _ctx_budget, now_iso, ensure_memory_type_column, _embed, _temp_json, _entities_to_json |
 | rune_operator | 2746 | 25/68 | _build_model_router, _sorted_tool_names, _filter_safety_noise, _wire_auto_filer, _reconcile_guarded, get_operator |
-| context_broker | 870 | 6/28 | _py_default_estimator, _cc_core, _py_default_budget, _py_default_memory_budget, _py_memory_disabled_value, reset_discovered_windows |
 | construction_interface | 1193 | 6/27 | _to_stage_check, _guard_specs_from_obligations, t_layout, _obligations_projection, _obligation_scope_files, main |
 | structural_store | 690 | 13/34 | _ss_core, _sorted, new_doc, _redact_secrets, save, canonical_digest |
 | provider_registry | 679 | 11/28 | _ctx_budget, _format_tools_openai, _execute_tool_calls, _execute_tool_calls_anthropic, get_registry, connect_provider |
@@ -168,6 +167,7 @@ Function-level detail: 53% (1268/2376 runtime functions matched) — secondary;
 |---|---|---|---|
 | curator | 1838 | 43/43 | GO |
 | sleep_time | 1056 | 26/27 | GO |
+| context_broker | 870 | 28/28 | GO |
 | local_tool_runtime | 809 | 30/30 | GO |
 | memory_consolidation | 770 | 23/23 | GO |
 | persona_gate | 724 | 26/26 | GO |

docs/factory-claims.json

@@ -11,6 +11,7 @@
     "concept_topology",
     "construction_intent",
     "construction_patterns_builtin",
+    "context_broker",
     "curator",
     "drift-ledger",
     "drift_ledger",
@@ -68,7 +69,6 @@
     "research_budget": "rune-wave-be",
     "evidence_capsule": "rune-wave-be",
     "builder_bridge_cli": "rune-wave-be",
-    "context_broker": "rune-wave-bf",
     "structural_store": "rune-wave-bh",
     "growth_delta": "rune-wave-bl",
     "safety_override": "rune-wave-bn",

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "rune"
-version = "0.270.0"
+version = "0.271.0"
 description = "Rune — a stateful coding assistant with persistent memory, orchestration, and TUI"
 requires-python = ">=3.10"
 

scripts/check_structural_store_parity.py

@@ -0,0 +1,452 @@
+#!/usr/bin/env python3
+"""Live parity: structural_store Rust kernels vs Python oracle (dichotomy
+wave, contract BUILDER-2026-10-13T00:32:22Z-bc3aae-factory-gap-ss). Layers:
+
+  1. FROZEN — the frozen truth corpus (frozen_ss_truth.json, generated by
+     scripts/gen_structural_store_truth.py against the LIVE oracle with
+     RUNE_STRUCTURAL_CORE=python pinned) replayed through the ctypes FFI,
+     case by case, deep-copied inputs.
+  2. FUZZ   — 300 seeded random docs (seed 2479) per kernel: oracle floor
+     vs FFI, exact compare; save() compared BYTE-EXACT on real files.
+  3. LIVE   — dual-mode facade battery: the same facade calls executed with
+     RUNE_STRUCTURAL_CORE=python (oracle floor) and with the Rust core live;
+     results must be identical (timestamps masked).
+
+Exit 0 only at 100% parity on the gap-closing functions.
+"""
+import calendar
+import copy
+import json
+import os
+import random
+import string
+import sys
+import tempfile
+import time as _time
+import types
+from pathlib import Path
+
+os.environ["RUNE_STRUCTURAL_CORE"] = "python"   # oracle floor for generation
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+
+from rune import structural_store as ss  # noqa: E402
+from rune import ss_loader  # noqa: E402
+
+FIXED_NOW = "2026-10-13T01:02:03Z"
+FIXED_EPOCH = calendar.timegm(_time.strptime(FIXED_NOW, "%Y-%m-%dT%H:%M:%SZ"))
+ss.now = lambda: FIXED_NOW
+ss.time = types.SimpleNamespace(   # snapshot() stamps straight off time.*
+    gmtime=lambda: _time.gmtime(FIXED_EPOCH),
+    strftime=lambda fmt, t=None: _time.strftime(fmt, t if t is not None else _time.gmtime()),
+)
+
+TRUTH = json.loads((REPO / "src" / "rust" / "structural_store" / "src"
+                    / "frozen_ss_truth.json").read_text())
+
+fails: list[str] = []
+counts = {"frozen": 0, "fuzz": 0, "live": 0}
+
+
+def fail(layer, name, detail):
+    fails.append(f"[{layer}] {name}: {detail}")
+    print(f"FAIL[{layer}] {name}: {detail}")
+    if len(fails) <= 3:
+        import difflib
+        w, _, g = detail.partition(" got=")
+        want_s = w[5:] if w.startswith("want=") else w
+        a = want_s.replace(r"\u", "\\u").splitlines()
+        b = g.splitlines()
+        dump = "\n".join(difflib.unified_diff(a, b, lineterm=""))[:2000]
+        (REPO / "parity_first_failures.log").open("a").write(
+            f"=== {layer}/{name} ===\n{dump}\n")
+
+
+def j(x):
+    return json.dumps(x, sort_keys=True, ensure_ascii=True)
+
+
+def check(layer, name, want, got):
+    counts[layer] += 1
+    if j(want) != j(got):
+        fail(layer, name, f"want={j(want)[:300]} got={j(got)[:300]}")
+
+
+# ── oracle dispatch (python floor) ──────────────────────────────────────────
+
+def oracle_call(fn, a):
+    return {
+        "sorted": lambda: ss._sorted(a["value"]),
+        "new_doc": lambda: ss.new_doc(),
+        "redact": lambda: ss._redact_secrets(a["value"]),
+        "digest": lambda: ss.canonical_digest(a["doc"]),
+        "alias": lambda: _mut(lambda d: ss.alias(d, a["cid"], a["old"]), a["doc"]),
+        "spine_uri": lambda: ss.spine_uri(a["entity_id"]),
+        "resolve_uri": lambda: ss.resolve_uri(a["uri"]),
+        "doc_authorities": lambda: _mut(ss._doc_authorities, a["doc"]),
+        "authority_health": lambda: ss.authority_health(a["doc"]),
+        "bind_task": lambda: _mut(lambda d: ss.bind_task(
+            d, a["task_id"], a["entity_ids"], a["evidence_ref"]), a["doc"]),
+        "task_entities": lambda: ss.task_entities(a["doc"], a["task_id"]),
+        "impact_frontier": lambda: ss.impact_frontier(
+            a["doc"], a["seeds"], max_level=a.get("max_level", 2),
+            change_kind=a.get("change_kind")),
+        "observed_impact": lambda: _mut_ret(lambda d: ss.record_observed_impact(
+            d, a["task_id"], a["items"]), a["doc"]),
+        "predicted_vs_observed": lambda: ss.predicted_vs_observed(
+            a["predicted"], a["observed_doc"], a["seeds"]),
+        "events": lambda: ss.structural_events(a["doc"]),
+        "diff": lambda: ss.diff(a["old"], a["new"]),
+        "dedupe": lambda: _mut(ss.dedupe_observations, a["doc"]),
+        "promote_fix": lambda: _mut(lambda d: ss.promote(
+            d, a["id"], a["state"], a["ev"]), a["doc"]),
+        "resolve_status_fix": lambda: ss.resolve_status(a["entity"]),
+        "supersede_fix": lambda: _mut(lambda d: ss.supersede(
+            d, a["new"], a["old"], a["ev"]), a["doc"]),
+    }[fn]()
+
+
+def _mut(call, doc):
+    d = copy.deepcopy(doc)
+    try:
+        ret = call(d)
+    except ValueError as exc:
+        return {"__err__": str(exc)}
+    return {"__doc__": d, "__ret__": ret}
+
+
+def _mut_ret(call, doc):
+    d = copy.deepcopy(doc)
+    ret = call(d)
+    return {"__doc__": d, "__ret__": ret}
+
+
+# ── FFI dispatch ────────────────────────────────────────────────────────────
+
+def ffi_call(fn, a, tmp):
+    return {
+        "sorted": lambda: ss_loader.sorted_value(a["value"]),
+        "new_doc": lambda: ss_loader.new_doc(FIXED_NOW),
+        "redact": lambda: ss_loader.redact_secrets(a["value"]),
+        "digest": lambda: ss_loader.canonical_digest(a["doc"]),
+        "alias": lambda: _ffi_mut(
+            lambda d: ss_loader.alias(d, a["cid"], a["old"], FIXED_NOW), a["doc"],
+            ret_fn=lambda d: ss.entity(d, a["cid"])),
+        "spine_uri": lambda: ss_loader.spine_uri(a["entity_id"]),
+        "resolve_uri": lambda: ss_loader.resolve_uri(a["uri"]),
+        "doc_authorities": lambda: _ffi_mut(
+            ss_loader.doc_authorities, a["doc"],
+            ret_fn=lambda d: copy.deepcopy(ss._doc_authorities(d))),
+        "authority_health": lambda: ss_loader.authority_health(a["doc"]),
+        "bind_task": lambda: _ffi_mut(
+            lambda d: ss_loader.bind_task(
+                d, a["task_id"], a["entity_ids"], a["evidence_ref"], FIXED_NOW), a["doc"],
+            ret_fn=lambda d: ss.entity(d, f"task:{a['task_id']}") or {"id": f"task:{a['task_id']}"}),
+        "task_entities": lambda: ss_loader.task_entities(a["doc"], a["task_id"]),
+        "impact_frontier": lambda: ss_loader.impact_frontier(
+            a["doc"], a["seeds"], max_level=a.get("max_level", 2),
+            change_kind=a.get("change_kind")),
+        "observed_impact": lambda: _ffi_mut_ret(lambda d: ss_loader.record_observed_impact(
+            d, a["task_id"], a["items"], FIXED_NOW), a["doc"]),
+        "predicted_vs_observed": lambda: ss_loader.predicted_vs_observed(
+            a["predicted"], a["observed_doc"], a["seeds"]),
+        "events": lambda: ss_loader.structural_events(a["doc"]),
+        "diff": lambda: ss_loader.diff(a["old"], a["new"]),
+        "dedupe": lambda: _ffi_mut(ss_loader.dedupe_observations, a["doc"]),
+        "promote_fix": lambda: _ffi_mut(
+            lambda d: ss_loader.promote(d, a["id"], a["state"], a["ev"], FIXED_NOW), a["doc"],
+            ret_fn=lambda d: ss.entity(d, a["id"])),
+        "resolve_status_fix": lambda: ss_loader.resolve_status(a["entity"]),
+        "supersede_fix": lambda: _ffi_mut(
+            lambda d: ss_loader.supersede(d, a["new"], a["old"], a["ev"], FIXED_NOW), a["doc"],
+            ret_fn=lambda d: ss.entity(d, a["old"])),
+    }[fn]()
+
+
+def _ret_entity(_updated_doc, d, cid):
+    return ss.entity(d, cid)
+
+
+def _ffi_mut(call, doc, ret_fn=None):
+    """FFI mutation contract: the kernel returns the UPDATED doc; the
+    oracle-shaped __ret__ is re-derived from it via ret_fn."""
+    d = copy.deepcopy(doc)
+    try:
+        out = call(d)
+    except RuntimeError as exc:
+        return {"__err__": str(exc).split("core error: ", 1)[-1]}
+    if isinstance(out, dict):
+        d = out
+    return {"__doc__": d, "__ret__": ret_fn(copy.deepcopy(d)) if ret_fn else None}
+
+
+def _ffi_mut_ret(call, doc):
+    d = copy.deepcopy(doc)
+    out = call(d)
+    if isinstance(out.get("doc"), dict):
+        d = out["doc"]
+    return {"__doc__": d, "__ret__": out["result"]}
+
+
+def frozen():
+    for fn, cases in TRUTH["groups"].items():
+        for c in cases:
+            a = copy.deepcopy(c["args"])
+            if fn == "save" or fn == "snapshot":
+                continue  # byte-law layer below
+            try:
+                want = oracle_call(fn, a)
+            except ValueError as exc:
+                want = {"__err__": str(exc)}
+            got_raw = None
+            try:
+                got = ffi_call(fn, a, None)
+            except RuntimeError as exc:
+                got = {"__err__": str(exc).split("core error: ", 1)[-1]}
+            counts["frozen"] += 1
+            if j(want) != j(got):
+                fail("frozen", f"{fn}/{c['name']}", f"want={j(want)[:280]} got={j(got)[:280]}")
+
+
+def byte_law(tmp):
+    """save()/snapshot() must be byte-exact through the FFI."""
+    save_docs = TRUTH["groups"]["save"]
+    for c in save_docs:
+        d = copy.deepcopy(c["args"]["doc"])
+        name = c["name"]
+        try:
+            p = ss.save(d, tmp / f"py/{name}/store.json")
+            want = {"content": Path(p).read_text(), "doc": d}
+        except ValueError as exc:
+            want = {"err": str(exc)}
+        d2 = copy.deepcopy(c["args"]["doc"])
+        try:
+            out = ss_loader.save(d2, str(tmp / f"rs/{name}/store.json"))
+            got = {"content": Path(out["path"]).read_text(), "doc": out["doc"]}
+        except RuntimeError as exc:
+            got = {"err": str(exc).split("core error: ", 1)[-1]}
+        check("frozen", f"save-bytes/{name}", want, got)
+    for c in TRUTH["groups"]["snapshot"]:
+        a = copy.deepcopy(c["args"])
+        d = a["doc"]
+        p = ss.snapshot(d, tmp / f"py-snap/{c['name']}", a["label"], tag=a["tag"])
+        want = {"path": Path(p).name, "content": Path(p).read_text(), "doc": d}
+        d2 = copy.deepcopy(a["doc"])
+        out = ss_loader.snapshot(d2, str(tmp / f"rs-snap/{c['name']}"), a["label"], a["tag"], FIXED_NOW)
+        got = {"path": Path(out["path"]).name, "content": Path(out["path"]).read_text(), "doc": out["doc"]}
+        check("frozen", f"snapshot-bytes/{c['name']}", want, got)
+
+
+# ── fuzz (seeded) ───────────────────────────────────────────────────────────
+
+WORDS = ["m:a", "m:b", "m:c", "c:api", "task:T1", "é-unit", "x" * 40, ""]
+RELTYPES = ["depends_on", "implements", "consumes", "consumed_by", "verified_by",
+            "obligated_canary", "deployed_as", "documented_by", "supersedes",
+            "historical_relation", "related_memory", "belongs_to", "weird_rel"]
+SECRETS = ["token", "api_key", "password", "not_a_secret", "client_secret", "TOKEN"]
+
+
+def rand_value(rnd, depth=0):
+    kind = rnd.randint(0, 6 if depth < 2 else 4)
+    if kind == 0:
+        return rnd.randint(-5, 10 ** 6)
+    if kind == 1:
+        return rnd.choice([0.5, 1.0, -2.25, 3.0, 1e-7, 1e21, 12345.6789])
+    if kind == 2:
+        return rnd.choice([True, False, None])
+    if kind == 3:
+        return "".join(rnd.choice(string.printable[:80] + "é☕φ") for _ in range(rnd.randint(0, 12)))
+    if kind == 4:
+        return rnd.choice(WORDS)
+    if kind == 5:
+        return [rand_value(rnd, depth + 1) for _ in range(rnd.randint(0, 4))]
+    return {rnd.choice("abcdefé") + str(i): rand_value(rnd, depth + 1)
+            for i in range(rnd.randint(0, 4))}
+
+
+def rand_doc(rnd):
+    d = {"schema_version": 1, "generated_at": FIXED_NOW,
+         "entities": [], "relationships": [], "observations": []}
+    for i in range(rnd.randint(0, 5)):
+        e = {"id": rnd.choice(WORDS) or f"m:e{i}", "kind": rnd.choice(["module", "contract", "widget"]),
+             "maturity": rnd.choice(["DECLARED", "PRESENT", "WIRED", "REACHABLE", "EXERCISED", "LIVE_PROVEN"]),
+             "evidence_refs": sorted({f"ev-{rnd.randint(0, 3)}" for _ in range(rnd.randint(0, 2))}),
+             "observed_at": FIXED_NOW}
+        if rnd.random() < 0.3:
+            e["aliases"] = [f"m:old{i}"]
+        if rnd.random() < 0.3:
+            e["ext"] = {"conflicts": rand_value(rnd, 1)} if rnd.random() < 0.5 else {"n": i}
+        if rnd.random() < 0.3:
+            e["identities"] = {"p": rand_value(rnd, 1)}
+        if rnd.random() < 0.2:
+            e[rnd.choice(SECRETS)] = "s3cret"
+        d["entities"].append(e)
+    for _ in range(rnd.randint(0, 6)):
+        rel = {"type": rnd.choice(RELTYPES), "from": rnd.choice(WORDS), "to": rnd.choice(WORDS),
+               "evidence_refs": [f"ev-{rnd.randint(0, 2)}"] if rnd.random() < 0.8 else [],
+               "observed_at": FIXED_NOW}
+        if rnd.random() < 0.2:
+            rel["status"] = rnd.choice(["STALE", "CANDIDATE", "CURRENT"])
+        d["relationships"].append(rel)
+    for _ in range(rnd.randint(0, 4)):
+        o = {"kind": rnd.choice(["superseded", "k1"]), "x": rand_value(rnd, 1), "observed_at": FIXED_NOW}
+        d["observations"].append(o)
+    if rnd.random() < 0.4:
+        d["authority_generations"] = {
+            rnd.choice(["p1", "p2", "é-proj"]): {
+                "generation": rnd.randint(0, 5),
+                "observed": [f"{rnd.choice(RELTYPES)}|{rnd.choice(WORDS)}|{rnd.choice(WORDS)}"
+                             for _ in range(rnd.randint(0, 3))],
+                **({"status": rnd.choice(["CURRENT", "DEGRADED", "UNKNOWN"])} if rnd.random() < 0.7 else {}),
+            } for _ in range(rnd.randint(1, 3))}
+    return d
+
+
+def fuzz():
+    rnd = random.Random(2479)
+    for i in range(300):
+        d = rand_doc(rnd)
+        d2 = copy.deepcopy(d)
+        check("fuzz", f"sorted/{i}", ss._sorted(copy.deepcopy(d)), ss_loader.sorted_value(copy.deepcopy(d)))
+        check("fuzz", f"redact/{i}", ss._redact_secrets(copy.deepcopy(d)), ss_loader.redact_secrets(copy.deepcopy(d)))
+        check("fuzz", f"digest/{i}", ss.canonical_digest(copy.deepcopy(d)), ss_loader.canonical_digest(copy.deepcopy(d)))
+        check("fuzz", f"health/{i}", ss.authority_health(copy.deepcopy(d)), ss_loader.authority_health(copy.deepcopy(d)))
+        check("fuzz", f"events/{i}", ss.structural_events(copy.deepcopy(d)), ss_loader.structural_events(copy.deepcopy(d)))
+        d3 = copy.deepcopy(d)
+        ss.dedupe_observations(d3)
+        d4 = copy.deepcopy(d)
+        d4 = ss_loader.dedupe_observations(d4)
+        check("fuzz", f"dedupe/{i}", d3, d4)
+        seed = rnd.choice(WORDS) or "m:a"
+        ck = rnd.choice([None, "contract_change", "implementation_change", "contract_violation"])
+        check("fuzz", f"frontier/{i}",
+              ss.impact_frontier(copy.deepcopy(d), [seed], change_kind=ck),
+              ss_loader.impact_frontier(copy.deepcopy(d), [seed], change_kind=ck))
+        check("fuzz", f"diff/{i}", ss.diff(copy.deepcopy(d), d2), ss_loader.diff(copy.deepcopy(d), copy.deepcopy(d2)))
+        ent = rnd.choice(d["entities"]) if d["entities"] else {}
+        check("fuzz", f"resolve_status/{i}", ss.resolve_status(copy.deepcopy(ent)),
+              ss_loader.resolve_status(copy.deepcopy(ent)))
+        # mutation kernels
+        want = _mut(lambda x: ss.bind_task(x, "T1", ["m:a", "c:api"], "ev"), copy.deepcopy(d))
+        got = _ffi_mut(lambda x: ss_loader.bind_task(x, "T1", ["m:a", "c:api"], "ev", FIXED_NOW),
+                       copy.deepcopy(d),
+                       ret_fn=lambda x: ss.entity(x, "task:T1") or {"id": "task:T1"})
+        check("fuzz", f"bind_task/{i}", want, got)
+        want = _mut(lambda x: ss.alias(x, "m:b", "m:b-old"), copy.deepcopy(d)) if any(
+            e["id"] == "m:b" for e in d["entities"]) else {"__err__": "alias target not found: m:b"}
+        got = _ffi_mut(lambda x: ss_loader.alias(x, "m:b", "m:b-old", FIXED_NOW), copy.deepcopy(d),
+                       ret_fn=lambda x: ss.entity(x, "m:b"))
+        check("fuzz", f"alias/{i}", want, got)
+        want = _mut(lambda x: ss.promote(x, "m:a", "WIRED", "ev-f"), copy.deepcopy(d))
+        got = _ffi_mut(lambda x: ss_loader.promote(x, "m:a", "WIRED", "ev-f", FIXED_NOW), copy.deepcopy(d),
+                       ret_fn=lambda x: ss.entity(x, "m:a"))
+        check("fuzz", f"promote/{i}", want, got)
+        want = _mut(lambda x: ss.supersede(x, "m:b", "m:a", "ev-s"), copy.deepcopy(d))
+        got = _ffi_mut(lambda x: ss_loader.supersede(x, "m:b", "m:a", "ev-s", FIXED_NOW), copy.deepcopy(d),
+                       ret_fn=lambda x: ss.entity(x, "m:a"))
+        check("fuzz", f"supersede/{i}", want, got)
+        items = [{"entity": rnd.choice(WORDS), "source": "t", "detail": rand_value(rnd, 1)} for _ in range(3)]
+        want = _mut_ret(lambda x: ss.record_observed_impact(x, "T1", copy.deepcopy(items)), copy.deepcopy(d))
+        got = _ffi_mut_ret(lambda x: ss_loader.record_observed_impact(x, "T1", copy.deepcopy(items), FIXED_NOW), copy.deepcopy(d))
+        check("fuzz", f"observed_impact/{i}", want, got)
+
+
+def save_bytes_fuzz(tmp):
+    rnd = random.Random(4242)
+    for i in range(50):
+        d = rand_doc(rnd)
+        p1 = ss.save(copy.deepcopy(d), tmp / f"fz-py/{i}.json")
+        out = ss_loader.save(copy.deepcopy(d), str(tmp / f"fz-rs/{i}.json"))
+        b1 = Path(p1).read_bytes()
+        b2 = Path(out["path"]).read_bytes()
+        counts["fuzz"] += 1
+        if b1 != b2:
+            fail("fuzz", f"save-bytes/{i}", f"py={b1[:160]!r} rs={b2[:160]!r}")
+        check("fuzz", f"digest-bytes/{i}", ss.canonical_digest(copy.deepcopy(d)),
+              ss_loader.canonical_digest(copy.deepcopy(d)))
+
+
+# ── live dual-mode facade battery ───────────────────────────────────────────
+
+def facade_battery():
+    def run(tag):
+        import subprocess
+        code = r'''
+import copy, json, os, sys, tempfile, calendar, time as _time, types
+sys.path.insert(0, "src")
+from rune import structural_store as ss
+FIXED = "2026-10-13T01:02:03Z"
+ss.now = lambda: FIXED
+_EPOCH = calendar.timegm(_time.strptime(FIXED, "%Y-%m-%dT%H:%M:%SZ"))
+ss.time = types.SimpleNamespace(   # freeze snapshot()'s wall-clock stamp
+    gmtime=lambda: _time.gmtime(_EPOCH),
+    strftime=lambda fmt, t=None: _time.strftime(fmt, t if t is not None else _time.gmtime()),
+)
+out = {}
+d = ss.new_doc()
+ss.add_entity(d, id="m:a", kind="module")
+ss.add_entity(d, id="m:b", kind="module")
+ss.add_entity(d, id="c:api", kind="contract")
+ss.add_relationship(d, type="implements", from_id="m:a", to_id="c:api", evidence_refs=["ev"])
+ss.add_relationship(d, type="depends_on", from_id="m:b", to_id="m:a", evidence_refs=["ev"])
+ss.alias(d, "m:b", "m:b-old")
+ss.promote(d, "m:a", "WIRED", "ev-2")
+ss.bind_task(d, "T1", ["m:a", "m:b"], "ev-t")
+with tempfile.TemporaryDirectory() as td:
+    p = ss.save(d, td + "/s.json")
+    out["save_bytes_sha"] = __import__("hashlib").sha256(open(p, "rb").read()).hexdigest()
+    out["snapshot_name"] = ss.snapshot(d, td + "/sn", "l1", tag="cafe1234").name
+out["digest"] = ss.canonical_digest(d)
+out["entities"] = sorted(e["id"] for e in d["entities"])
+out["via_alias"] = ss.entity(d, "m:b-old")["maturity"]
+out["health"] = ss.authority_health(d)
+out["frontier"] = ss.impact_frontier(d, ["c:api"], change_kind=None)
+out["frontier_impl"] = ss.impact_frontier(d, ["m:a"], change_kind=None)
+out["events_n"] = len(ss.structural_events(d))
+out["diff"] = ss.diff(d, copy.deepcopy(d))
+out["task_entities"] = ss.task_entities(d, "T1")
+ss.dedupe_observations(d)
+out["obs_n"] = len(d["observations"])
+r = ss.record_observed_impact(d, "T2", [{"entity": "m:a", "detail": "x"}])
+out["observed"] = r
+out["pvo"] = ss.predicted_vs_observed(out["frontier"], d, ["c:api"])
+print(json.dumps(out, sort_keys=True))
+'''
+        env = dict(os.environ)
+        if tag == "python":
+            env["RUNE_STRUCTURAL_CORE"] = "python"
+        else:
+            env.pop("RUNE_STRUCTURAL_CORE", None)
+        r = subprocess.run([sys.executable, "-c", code], capture_output=True,
+                           text=True, cwd=str(REPO), env=env, timeout=120)
+        if r.returncode != 0:
+            fail("live", tag, r.stderr[-300:])
+            return None
+        return json.loads(r.stdout)
+
+    py = run("python")
+    rs = run("rust")
+    if py is None or rs is None:
+        return
+    check("live", "dual-mode-facade", py, rs)
+
+
+def main():
+    if not ss_loader.available():
+        print("Rust core not built — run cargo build --release in src/rust/structural_store")
+        return 2
+    with tempfile.TemporaryDirectory() as td:
+        frozen()
+        byte_law(Path(td))
+        fuzz()
+        save_bytes_fuzz(Path(td))
+    facade_battery()
+    total = sum(counts.values())
+    print(f"\nPARITY: {total} checks, {len(fails)} failures "
+          f"(frozen={counts['frozen']} fuzz={counts['fuzz']} live={counts['live']})")
+    return 1 if fails else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/gen_structural_store_truth.py

@@ -0,0 +1,378 @@
+#!/usr/bin/env python3
+"""Freeze the dichotomy-wave gap-closing truth corpus for
+src/rune/structural_store.py (contract BUILDER-2026-10-13T00:32:22Z-bc3aae-
+factory-gap-ss, issue #2479 tranche).
+
+LAW (poisoned-truth): RUNE_STRUCTURAL_CORE=python is pinned at the top —
+truth is generated by the PYTHON ORACLE ONLY. Inputs are deep-copied before
+every oracle call (no leakage between cases). The clock is pinned by
+monkeypatching structural_store.now (no-clock law on the Rust side).
+
+The frozen corpus is embedded VERBATIM in the crate's cfg(test) battery
+(src/frozen_battery.rs, include_str!) and replayed by
+scripts/check_structural_store_parity.py through the live FFI.
+"""
+import calendar
+import copy
+import json
+import os
+import sys
+import tempfile
+import time as _time
+import types
+from pathlib import Path
+
+os.environ["RUNE_STRUCTURAL_CORE"] = "python"   # poisoned-truth law
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+
+from rune import structural_store as ss  # noqa: E402
+
+FIXED_NOW = "2026-10-13T01:02:03Z"
+FIXED_EPOCH = calendar.timegm(_time.strptime(FIXED_NOW, "%Y-%m-%dT%H:%M:%SZ"))
+ss.now = lambda: FIXED_NOW  # pin the clock
+# snapshot() formats its stamp straight off time.strftime/gmtime — pin the
+# module reference the oracle actually uses.
+ss.time = types.SimpleNamespace(
+    gmtime=lambda: _time.gmtime(FIXED_EPOCH),
+    strftime=lambda fmt, t=None: _time.strftime(fmt, t if t is not None else _time.gmtime()),
+)
+
+OUT = (REPO / "src" / "rust" / "structural_store" / "src"
+       / "frozen_ss_truth.json")
+
+groups: dict[str, list] = {}
+
+
+def case(group, name, args, expect):
+    groups.setdefault(group, []).append(
+        {"fn": group, "name": name, "args": args, "expect": expect})
+
+
+def run(group, name, args, call):
+    """Freeze one case: input deep-copied, oracle called, outcome recorded."""
+    frozen_args = copy.deepcopy(args)
+    try:
+        expect = {"ok": call(copy.deepcopy(args))}
+    except ValueError as exc:
+        expect = {"err": str(exc)}
+    case(group, name, frozen_args, expect)
+
+
+def ent(id, kind="module", maturity="DECLARED", **kw):
+    e = {"id": id, "kind": kind, "maturity": maturity, "evidence_refs": []}
+    e.update(kw)
+    return e
+
+
+def doc(*entities, rels=(), obs=(), **kw):
+    d = {"schema_version": 1, "generated_at": FIXED_NOW,
+         "entities": list(entities), "relationships": list(rels),
+         "observations": list(obs)}
+    d.update(kw)
+    return d
+
+
+def rel(t, f, to, **kw):
+    r = {"type": t, "from": f, "to": to, "evidence_refs": ["ev"],
+         "observed_at": FIXED_NOW}
+    r.update(kw)
+    return r
+
+
+def main():
+    # ── _sorted ─────────────────────────────────────────────────────────
+    for name, v in [
+        ("dict_keys_sorted", {"b": 1, "a": {"z": 2, "y": [3, {"c": 4, "b": 5}]}}),
+        ("list_of_dicts_by_canonical_form", [{"b": 2, "a": 1}, {"a": 9}, {"a": 1, "b": 2}, {}]),
+        ("list_of_scalars", [3, 1, 2, "b", "a", True, None]),
+        ("nested_lists", [[2], [1, 2], [1], []]),
+        ("unicode_keys_and_values", {"é": "ü", "z": "a", "a": ["z", "é"]}),
+        ("floats_and_ints", [1.5, 0.5, 2, 1, -1.25]),
+        ("empty_containers", {"e": [], "f": {}, "g": [[]]}),
+        ("dict_of_scalars_passthrough", {"x": 1, "y": "s", "z": None, "w": False}),
+    ]:
+        run("sorted", name, {"value": v}, lambda a: ss._sorted(a["value"]))
+
+    # ── new_doc ─────────────────────────────────────────────────────────
+    run("new_doc", "basic", {"now": FIXED_NOW}, lambda a: ss.new_doc())
+
+    # ── _redact_secrets ─────────────────────────────────────────────────
+    for name, v in [
+        ("exact_names", {"token": "t", "api_token": "t", "api_key": "t",
+                         "apikey": "t", "password": "t", "secret": "t",
+                         "auth_header": "t", "authorization": "t",
+                         "credential": "t", "credentials": "t",
+                         "private_key": "t", "access_token": "t",
+                         "bearer": "t", "safe": "keep"}),
+        ("substring_names", {"my_token": "t", "API_KEY": "t",
+                             "user_password_hash": "t", "client_secret": "t",
+                             "my_credentials_v2": "t", "token_count": "t"}),
+        ("nested", {"a": {"token": "deep"}, "b": [{"password": "x"}, {"ok": 1}]}),
+        ("non_secret_passthrough", {"a": 1, "b": "s", "c": None, "d": [1, 2]}),
+        ("case_insensitive", {"Token": "t", "PASSWORD": "t"}),
+    ]:
+        run("redact", name, {"value": v}, lambda a: ss._redact_secrets(a["value"]))
+
+    # ── save (byte law frozen from the real filesystem) ────────────────
+    with tempfile.TemporaryDirectory() as td:
+        def save_call(args):
+            p = ss.save(copy.deepcopy(args["doc"]),
+                        Path(td) / args["name"] / "store.json")
+            return {"content": Path(p).read_text(), "doc": p and None}
+        # capture the mutated doc separately (save mutates schema_version in)
+        def save_case(name, d, expect_err=False):
+            args = {"doc": copy.deepcopy(d), "name": name}
+            try:
+                probe = copy.deepcopy(d)
+                p = ss.save(probe, Path(td) / name / "store.json")
+                content = Path(p).read_text()
+                expect = {"ok": {"content": content, "doc": probe}}
+            except ValueError as exc:
+                expect = {"err": str(exc)}
+            case("save", name, args, expect)
+
+        base = doc(ent("m:a"), rels=[rel("depends_on", "m:a", "m:b")], generated_at=FIXED_NOW)
+        save_case("full_doc", base)
+        save_case("no_schema_version", {k: v for k, v in base.items() if k != "schema_version"})
+        save_case("schema_one_float", {**base, "schema_version": 1.0})
+        save_case("schema_ninety_nine", {**base, "schema_version": 99})
+        save_case("schema_string", {**base, "schema_version": "x"})
+        save_case("secrets_redacted_on_disk", {**base, "credentials": {"token": "s3cret", "note": "x"}})
+        save_case("unicode_escaped", {**base, "note": "café ☕ — ok"})
+        save_case("empty_doc", {"schema_version": 1})
+        save_case("nested_empty", {**base, "ext": {"a": [], "b": {}}})
+
+    # ── canonical_digest ────────────────────────────────────────────────
+    dd = doc(ent("m:a"), rels=[rel("depends_on", "m:a", "m:b")])
+    run("digest", "basic", {"doc": dd}, lambda a: ss.canonical_digest(a["doc"]))
+    run("digest", "key_order_irrelevant",
+        {"doc": dd, "doc_reordered": {"entities": dd["entities"], "schema_version": 1,
+                                      "observations": dd["observations"],
+                                      "relationships": dd["relationships"],
+                                      "generated_at": dd["generated_at"]}},
+        lambda a: [ss.canonical_digest(a["doc"]), ss.canonical_digest(a["doc_reordered"])])
+    run("digest", "secret_never_changes_digest",
+        {"doc": {**dd, "token": "s3cret"}},
+        lambda a: [ss.canonical_digest(a["doc"]), ss.canonical_digest({**a["doc"], "token": "other"})])
+    run("digest", "unicode", {"doc": {**dd, "note": "é"}}, lambda a: ss.canonical_digest(a["doc"]))
+
+    # ── snapshot ────────────────────────────────────────────────────────
+    with tempfile.TemporaryDirectory() as td:
+        def snap_case(name, d, label, tag):
+            args = {"doc": copy.deepcopy(d), "dir": "snaps", "label": label, "tag": tag}
+            probe = copy.deepcopy(d)
+            p = ss.snapshot(probe, Path(td) / name / "snaps", label, tag=tag)
+            content = Path(p).read_text()
+            case("snapshot", name, args,
+                 {"ok": {"path": "snaps/" + Path(p).name, "content": content, "doc": probe}})
+        snap_case("label_and_tag", dd, "l1", "deadbeef1234")
+        snap_case("no_label", dd, "", "deadbeef1234")
+        snap_case("empty_tag_is_noident", dd, "l1", "")
+        snap_case("short_tag", dd, "", "abc")
+
+    # ── alias ───────────────────────────────────────────────────────────
+    adoc = doc(ent("m:a"), ent("m:b"))
+    run("alias", "new_alias", {"doc": adoc, "cid": "m:a", "old": "m:old"},
+        lambda a: ss.alias(a["doc"], a["cid"], a["old"]))
+    run("alias", "idempotent",
+        {"doc": doc(ent("m:a", aliases=["m:old"])), "cid": "m:a", "old": "m:old"},
+        lambda a: ss.alias(a["doc"], a["cid"], a["old"]))
+    run("alias", "target_missing", {"doc": adoc, "cid": "m:zz", "old": "m:old"},
+        lambda a: ss.alias(a["doc"], a["cid"], a["old"]))
+    run("alias", "self_alias", {"doc": adoc, "cid": "m:a", "old": "m:a"},
+        lambda a: ss.alias(a["doc"], a["cid"], a["old"]))
+    run("alias", "via_alias_resolution",
+        {"doc": doc(ent("m:a", aliases=["m:old"])), "cid": "m:old", "old": "m:older"},
+        lambda a: ss.alias(a["doc"], a["cid"], a["old"]))
+
+    # ── spine_uri / resolve_uri ─────────────────────────────────────────
+    for name, eid in [("kind_name", "project:alpha"), ("no_colon", "alpha"),
+                      ("two_colons", "project:al:pha")]:
+        run("spine_uri", name, {"entity_id": eid}, lambda a: ss.spine_uri(a["entity_id"]))
+    for name, uri in [("ok", "spine://project/alpha"), ("no_slash", "spine://alpha"),
+                      ("wrong_scheme", "memory://project/alpha"), ("no_scheme", "project/alpha"),
+                      ("nested_slash", "spine://project/a/b")]:
+        run("resolve_uri", name, {"uri": uri}, lambda a: ss.resolve_uri(a["uri"]))
+
+    # ── _doc_authorities ────────────────────────────────────────────────
+    for name, d in [
+        ("defaults_in", doc(ent("m:a"))),
+        ("existing_untouched", {**doc(ent("m:a")), "authority_generations": {"p": {"generation": 2}}}),
+    ]:
+        a = {"doc": copy.deepcopy(d)}
+        d2 = copy.deepcopy(d)
+        r = ss._doc_authorities(d2)
+        case("doc_authorities", name, a, {"ok": {"doc": d2, "ret": r}})
+
+    # ── authority_health ────────────────────────────────────────────────
+    run("authority_health", "empty", {"doc": doc(ent("m:a"))}, lambda a: ss.authority_health(a["doc"]))
+    run("authority_health", "populated",
+        {"doc": {**doc(ent("m:a")), "authority_generations": {
+            "z-proj": {"generation": 3, "status": "CURRENT", "last_success": FIXED_NOW,
+                       "last_attempt": FIXED_NOW, "last_error": None, "observed": ["owns|a|b", "owns|c|d"]},
+            "a-proj": {"generation": 1, "status": "DEGRADED", "observed": []}}}},
+        lambda a: ss.authority_health(a["doc"]))
+    run("authority_health", "missing_fields",
+        {"doc": {**doc(ent("m:a")), "authority_generations": {"p": {}}}},
+        lambda a: ss.authority_health(a["doc"]))
+
+    # ── bind_task / task_entities ───────────────────────────────────────
+    for name, d, task_id, ids, ev in [
+        ("with_evidence", doc(ent("m:a"), ent("m:b")), "T1", ["m:a", "m:b"], "ev-1"),
+        ("fallback_evidence", doc(ent("m:a")), "T2", ["m:a"], ""),
+        ("dedupes_relationships", doc(ent("m:a")), "T3", ["m:a", "m:a"], "ev"),
+    ]:
+        a = {"doc": copy.deepcopy(d), "task_id": task_id, "entity_ids": ids, "evidence_ref": ev}
+        d2 = copy.deepcopy(d)
+        r = ss.bind_task(d2, task_id, copy.deepcopy(ids), ev)
+        case("bind_task", name, a, {"ok": {"doc": d2, "ret": r}})
+    tdoc = doc(ent("m:a"), ent("m:b"),
+               rels=[rel("affects", "task:T1", "m:b"), rel("affects", "task:T1", "m:a"),
+                     rel("affects", "task:T9", "m:b")])
+    run("task_entities", "sorted_for_task", {"doc": tdoc, "task_id": "T1"},
+        lambda a: ss.task_entities(a["doc"], a["task_id"]))
+    run("task_entities", "no_bindings", {"doc": tdoc, "task_id": "TX"},
+        lambda a: ss.task_entities(a["doc"], a["task_id"]))
+
+    # ── impact_frontier (branch battery) ────────────────────────────────
+    def frontier_case(name, d, seeds, change_kind=None, max_level=2):
+        args = {"doc": copy.deepcopy(d), "seeds": seeds,
+                "change_kind": change_kind, "max_level": max_level}
+        run("impact_frontier", name, args,
+            lambda a: ss.impact_frontier(a["doc"], a["seeds"], max_level=a["max_level"],
+                                         change_kind=a["change_kind"]))
+
+    impl_doc = doc(
+        ent("m:core"), ent("m:app"), ent("m:lib"), ent("m:tests"), ent("m:deploy"),
+        rels=[
+            rel("depends_on", "m:app", "m:core"),
+            rel("implements", "m:app", "c:api"),
+            rel("implements", "m:lib", "c:api"),
+            rel("consumes", "m:core", "c:api"),
+            rel("verified_by", "m:app", "k:canary"),
+            rel("deployed_as", "m:core", "d:live"),
+            rel("documented_by", "m:core", "doc:r"),
+            rel("supersedes", "m:lib", "m:old"),
+            rel("consumed_by", "m:lib", "m:core"),
+        ])
+    frontier_case("implementation_change", impl_doc, ["m:app"])
+    frontier_case("contract_change", impl_doc, ["c:api"])
+    frontier_case("contract_violation", impl_doc, ["c:api"], change_kind="contract_violation")
+    frontier_case("inferred_from_contract_kind", impl_doc, ["c:api"], change_kind=None)
+    frontier_case("stale_and_candidate_skipped",
+                  doc(ent("m:a"), ent("m:b"), ent("m:c"),
+                      rels=[rel("depends_on", "m:a", "m:b", status="STALE"),
+                            rel("depends_on", "m:a", "m:c", status="CANDIDATE")]),
+                  ["m:a"])
+    frontier_case("non_obligation_never_expands",
+                  doc(ent("m:a"), ent("m:b"),
+                      rels=[rel("documented_by", "m:a", "m:b")]),
+                  ["m:a"])
+    frontier_case("consumed_by_reverse",
+                  doc(ent("m:a"), ent("m:b"),
+                      rels=[rel("consumed_by", "m:b", "m:a")]),
+                  ["m:a"])
+    frontier_case("level2_from_level1_and_contextual", impl_doc, ["m:core"])
+    frontier_case("multi_seed_dedup", impl_doc, ["m:app", "m:app", "m:core"])
+    frontier_case("alias_seed_kind",
+                  doc(ent("c:api", kind="contract", aliases=["c:old"]),
+                      ent("m:app"),
+                      rels=[rel("implements", "m:app", "c:api")]),
+                  ["c:old"])
+
+    # ── record_observed_impact ──────────────────────────────────────────
+    for name, items in [
+        ("basic", [{"entity": "m:a", "source": "test", "detail": "d", "authority": "ci"},
+                   {"entity": "m:b"}]),
+        ("empty_entity_skipped", [{"entity": "", "source": "x"}, {"entity": "m:a"}]),
+        ("detail_truncated_300", [{"entity": "m:a", "detail": "x" * 500}]),
+        ("detail_multibyte_truncated", [{"entity": "m:a", "detail": "é" * 400}]),
+        ("defaults", [{"entity": "m:a"}]),
+    ]:
+        d0 = doc(ent("m:a"), ent("m:b"))
+        a = {"doc": copy.deepcopy(d0), "task_id": "T1", "items": copy.deepcopy(items)}
+        d2 = copy.deepcopy(d0)
+        result = ss.record_observed_impact(d2, "T1", copy.deepcopy(items))
+        case("observed_impact", name, a, {"ok": {"doc": d2, "result": result}})
+
+    # ── predicted_vs_observed ───────────────────────────────────────────
+    pdoc = doc(ent("m:a"), ent("m:b"), ent("m:c"),
+               rels=[rel("depends_on", "m:a", "m:b"), rel("depends_on", "m:a", "m:c")])
+    pred = {"level0_direct": ["m:a"], "level1_obligations": ["m:b"],
+            "contracts": [], "level2_verification": []}
+    run("predicted_vs_observed", "miss_and_needless",
+        {"predicted": pred, "observed_doc": pdoc, "seeds": ["m:a"]},
+        lambda a: ss.predicted_vs_observed(a["predicted"], a["observed_doc"], a["seeds"]))
+    run("predicted_vs_observed", "exact_match",
+        {"predicted": {"level0_direct": ["m:a"], "level1_obligations": ["m:b", "m:c"],
+                       "contracts": [], "level2_verification": []},
+         "observed_doc": pdoc, "seeds": ["m:a"]},
+        lambda a: ss.predicted_vs_observed(a["predicted"], a["observed_doc"], a["seeds"]))
+
+    # ── structural_events ───────────────────────────────────────────────
+    run("events", "full",
+        {"doc": doc(ent("m:a", observed_at=FIXED_NOW), ent("m:b", observed_at=FIXED_NOW),
+                    rels=[rel("depends_on", "m:a", "m:b")],
+                    obs=[{"kind": "superseded", "old": "x", "new": "y", "observed_at": FIXED_NOW}])},
+        lambda a: ss.structural_events(a["doc"]))
+    run("events", "observation_kind_default_and_type_override",
+        {"doc": doc(ent("m:a"),
+                    obs=[{"note": "no kind"}, {"kind": "k1", "type": "explicit_type"}])},
+        lambda a: ss.structural_events(a["doc"]))
+
+    # ── diff ────────────────────────────────────────────────────────────
+    old = doc(ent("m:a"), ent("m:gone"), ent("m:same", identities={"p": 1}, maturity="DECLARED"),
+              rels=[rel("depends_on", "m:a", "m:gone")])
+    new = doc(ent("m:a"), ent("m:new"), ent("m:same", identities={"p": 2}, maturity="WIRED"),
+              rels=[rel("depends_on", "m:a", "m:new"), rel("depends_on", "m:a", "m:a")])
+    run("diff", "all_categories", {"old": old, "new": new}, lambda a: ss.diff(a["old"], a["new"]))
+    run("diff", "no_changes", {"old": old, "new": copy.deepcopy(old)},
+        lambda a: ss.diff(a["old"], a["new"]))
+
+    # ── dedupe_observations ─────────────────────────────────────────────
+    for name, d in [
+        ("exact_dups", doc(ent("m:a"), obs=[{"kind": "k", "x": 1}, {"x": 1, "kind": "k"},
+                                            {"kind": "k", "x": 2}])),
+        ("nested_dups", doc(ent("m:a"), obs=[{"a": {"z": 1, "b": [2]}}, {"a": {"b": [2], "z": 1}}])),
+        ("empty", doc(ent("m:a"))),
+    ]:
+        a = {"doc": copy.deepcopy(d)}
+        d2 = copy.deepcopy(d)
+        r = ss.dedupe_observations(d2)
+        case("dedupe", name, a, {"ok": {"doc": d2, "ret": r}})
+
+    # ── fixed-kernel regressions (oracle parity proofs) ─────────────────
+    for name, d, eid, state, ev, ret in [
+        ("unknown_entity_returns_none", doc(ent("m:a")), "m:zz", "PRESENT", "e", None),
+        ("via_alias", doc(ent("m:a", aliases=["m:old"])), "m:old", "PRESENT", "e", "PRESENT"),
+        ("missing_entity_beats_bad_state", doc(ent("m:a")), "m:zz", "BOGUS", "", None),
+        ("demotion_recorded", doc(ent("m:a", maturity="PRESENT")), "m:a", "DECLARED", "e2", "PRESENT"),
+    ]:
+        a = {"doc": copy.deepcopy(d), "id": eid, "state": state, "ev": ev}
+        d2 = copy.deepcopy(d)
+        r = ss.promote(d2, eid, state, ev)
+        case("promote_fix", name, a, {"ok": {"ret": r, "doc": d2}})
+    run("resolve_status_fix", "status_verbatim",
+        {"entity": {"status": "WEIRD"}}, lambda a: ss.resolve_status(a["entity"]))
+    run("resolve_status_fix", "empty_conflicts_not_conflicting",
+        {"entity": {"maturity": "DECLARED", "ext": {"conflicts": []}}},
+        lambda a: ss.resolve_status(a["entity"]))
+    for name, d, nid, oid in [
+        ("rel_has_observed_at", doc(ent("m:a"), ent("m:b")), "m:b", "m:a"),
+        ("via_alias", doc(ent("m:a", aliases=["m:old"]), ent("m:b")), "m:b", "m:old"),
+    ]:
+        a = {"doc": copy.deepcopy(d), "new": nid, "old": oid, "ev": "ev-s"}
+        d2 = copy.deepcopy(d)
+        ss.supersede(d2, nid, oid, "ev-s")
+        case("supersede_fix", name, a, {"ok": {"doc": d2}})
+
+    out = {"module": "structural_store", "now": FIXED_NOW, "groups": groups}
+    OUT.write_text(json.dumps(out, indent=1, sort_keys=True) + "\n")
+    total = sum(len(v) for v in groups.values())
+    print(f"frozen {total} cases across {len(groups)} groups -> {OUT}")
+
+
+if __name__ == "__main__":
+    main()

scripts/open_contract_gap_ss.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+"""Open the #2479 dichotomy-wave gap-closing construction contract for
+structural_store (PARTIAL → ≥80% fn coverage). Walks draft→approved→building
+in ONE process (shared-store clobber hazard).
+
+Usage: open_contract_gap_ss.py
+"""
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from rune.construction_interface import ConstructionInterface
+
+iface = ConstructionInterface(builder="factory-gap-ss")
+
+r = iface.submit_need(
+    "Close structural_store PARTIAL gaps (dichotomy wave, issue #2479 "
+    "tranche): port the 21 audit-missing functions name-preserving to the "
+    "existing Rust crate (doc-envelope FFI ss_*, ctypes loader mirroring "
+    "broker_loader.py, facade delegation with RUNE_STRUCTURAL_CORE=python "
+    "pin), frozen-truth battery generated against the live oracle, live "
+    "parity 100% on new fns, Phase B language-advantage pass.",
+    origin="builder",
+)
+print("submit_need:", json.dumps(r)[:400])
+if not r.get("success"):
+    print("NEED FAILED — aborting")
+    sys.exit(1)
+c = r["ids"]["contract_id"]
+print("contract_id:", c)
+
+for phase in ("approved", "building"):
+    t = iface.transition(c, phase)
+    print(f"transition->{phase}:", json.dumps(t)[:300])
+    if not t.get("success"):
+        sys.exit(1)
+print("OPENED", c)

src/rune/__init__.py

@@ -6,7 +6,7 @@ This is the main entry point for the Rune program.
 import json
 from pathlib import Path
 
-__version__ = "0.270.0"
+__version__ = "0.271.0"
 
 # One build identity (handoff 2026-09-10 §4): the semantic version above is
 # the repo-level truth (auto-bumped at merge); the DEPLOYED revision (exact

src/rune/canary_gitea_automation.py

@@ -233,7 +233,7 @@ def main() -> int:
     )
     check(
         "Rune minor version advertises provisioning and restore verification",
-        '__version__ = "0.270.0"' in rune_version,
+        '__version__ = "0.271.0"' in rune_version,
         failures,
     )
     # #169 policy update: auto-bump on merge is sanctioned (Nick directive:

src/rune/rune_tui.py

@@ -8,7 +8,7 @@ mode system, subject header, shared memory and identity.
 
 This is not an interface. This is how I see, speak, and interact with the world.
 
-Version: v0.270.0
+Version: v0.271.0
 Features: 41 (24 original + 17 new native commands)
 
 Usage:
@@ -2037,7 +2037,7 @@ class RuneApp(App):
         # Show version info
         self.chat_channel.add_message(
             "system",
-            f"[dim]Rune v0.270.0 | Textual TUI | {len(loaded)} native modules[/dim]",
+            f"[dim]Rune v0.271.0 | Textual TUI | {len(loaded)} native modules[/dim]",
         )
 
         # Auto-show night mode suggestion if evening/night hours

src/rune/ss_loader.py

@@ -1,11 +1,32 @@
 #!/usr/bin/env python3
-"""ctypes loader for the Rust structural_store core (facade-1 burn-down).
+"""ctypes loader for the Rust structural_store core (facade-1 burn-down;
+dichotomy-wave gap-closing tranche mirrors broker_loader.py's contract).
 
-FFI contract (doc-envelope law):
+FFI contract (doc-envelope law — every call returns {"ok", "value"|"err"}):
   * ss_maturity_rank({state}) -> int (spine ladder)
   * ss_promote({doc, id, state, evidence_ref, now}) -> updated doc
   * ss_supersede({doc, new_id, old_id, evidence_ref, now}) -> updated doc
   * ss_resolve_status({entity}) -> string
+  * ss_sorted({value}) -> normalized value
+  * ss_new_doc({now}) -> doc
+  * ss_redact_secrets({value}) -> redacted value
+  * ss_save({doc, path}) -> {"doc", "path"} (writes the file)
+  * ss_canonical_digest({doc}) -> hex digest
+  * ss_snapshot({doc, snapshots_dir, label, tag, now}) -> {"doc", "path"}
+  * ss_alias({doc, canonical_id, old_id, now}) -> updated doc
+  * ss_spine_uri({entity_id}) -> string
+  * ss_resolve_uri({uri}) -> string
+  * ss_doc_authorities({doc}) -> updated doc
+  * ss_authority_health({doc}) -> per-projector map
+  * ss_bind_task({doc, task_id, entity_ids, evidence_ref, now}) -> updated doc
+  * ss_task_entities({doc, task_id}) -> [entity id]
+  * ss_impact_frontier({doc, seed_ids, max_level, change_kind}) -> frontier
+  * ss_record_observed_impact({doc, task_id, evidence_items, now})
+        -> {"doc", "result"}
+  * ss_predicted_vs_observed({predicted, observed_doc, seed_ids}) -> delta
+  * ss_structural_events({doc}) -> [event]
+  * ss_diff({old, new}) -> delta
+  * ss_dedupe_observations({doc}) -> updated doc
   * ss_free_string(ptr)
 
 FFI segfault law: explicit argtypes + restype (c_void_p for char*).
@@ -26,6 +47,16 @@ _LIB_PATH = Path(
     )
 )
 
+_SYMBOLS = (
+    "ss_maturity_rank", "ss_promote", "ss_supersede", "ss_resolve_status",
+    "ss_sorted", "ss_new_doc", "ss_redact_secrets", "ss_save",
+    "ss_canonical_digest", "ss_snapshot", "ss_alias", "ss_spine_uri",
+    "ss_resolve_uri", "ss_doc_authorities", "ss_authority_health",
+    "ss_bind_task", "ss_task_entities", "ss_impact_frontier",
+    "ss_record_observed_impact", "ss_predicted_vs_observed",
+    "ss_structural_events", "ss_diff", "ss_dedupe_observations",
+)
+
 _lib = None
 
 
@@ -42,10 +73,9 @@ def _load() -> ctypes.CDLL:
     lib = ctypes.CDLL(str(_LIB_PATH))
     lib.ss_free_string.argtypes = [ctypes.c_void_p]
     lib.ss_free_string.restype = None
-    for sym in ("ss_maturity_rank", "ss_promote", "ss_supersede",
-                "ss_resolve_status"):
+    for sym in _SYMBOLS:
         fn = getattr(lib, sym)
-        fn.argtypes = [ctypes.c_char_p]
+        fn.argtypes = [ctypes.c_void_p]
         fn.restype = ctypes.c_void_p
     _lib = lib
     return lib
@@ -53,16 +83,20 @@ def _load() -> ctypes.CDLL:
 
 def _call(sym: str, payload: dict):
     lib = _load()
-    ptr = getattr(lib, sym)(json.dumps(payload).encode("utf-8"))
+    data = json.dumps(payload).encode()
+    buf = ctypes.create_string_buffer(data)
+    ptr = getattr(lib, sym)(ctypes.cast(buf, ctypes.c_void_p))
     if not ptr:
         raise RuntimeError(f"{sym} returned null")
     try:
-        text = ctypes.cast(ptr, ctypes.c_char_p).value.decode("utf-8")
+        text = ctypes.string_at(ptr).decode("utf-8")
     finally:
         lib.ss_free_string(ptr)
     env = json.loads(text)
     if not env.get("ok"):
-        raise RuntimeError(f"{sym} core error: {env.get('err')}")
+        # the kernel error travels VERBATIM: the facade translates it into
+        # the oracle's ValueError contract (message text must match 1:1)
+        raise RuntimeError(env.get("err") or f"{sym} failed")
     return env["value"]
 
 
@@ -84,3 +118,93 @@ def supersede(doc: dict, new_id: str, old_id: str, evidence_ref: str,
 
 def resolve_status(entity: dict) -> str:
     return _call("ss_resolve_status", {"entity": entity})
+
+
+def sorted_value(value):
+    return _call("ss_sorted", {"value": value})
+
+
+def new_doc(now: str) -> dict:
+    return _call("ss_new_doc", {"now": now})
+
+
+def redact_secrets(value):
+    return _call("ss_redact_secrets", {"value": value})
+
+
+def save(doc: dict, path: str) -> dict:
+    return _call("ss_save", {"doc": doc, "path": path})
+
+
+def canonical_digest(doc: dict) -> str:
+    return _call("ss_canonical_digest", {"doc": doc})
+
+
+def snapshot(doc: dict, snapshots_dir: str, label: str, tag: str, now: str) -> dict:
+    return _call("ss_snapshot", {"doc": doc, "snapshots_dir": snapshots_dir,
+                                 "label": label, "tag": tag, "now": now})
+
+
+def alias(doc: dict, canonical_id: str, old_id: str, now: str) -> dict:
+    return _call("ss_alias", {"doc": doc, "canonical_id": canonical_id,
+                              "old_id": old_id, "now": now})
+
+
+def spine_uri(entity_id: str) -> str:
+    return _call("ss_spine_uri", {"entity_id": entity_id})
+
+
+def resolve_uri(uri: str) -> str:
+    return _call("ss_resolve_uri", {"uri": uri})
+
+
+def doc_authorities(doc: dict) -> dict:
+    return _call("ss_doc_authorities", {"doc": doc})
+
+
+def authority_health(doc: dict) -> dict:
+    return _call("ss_authority_health", {"doc": doc})
+
+
+def bind_task(doc: dict, task_id: str, entity_ids: list, evidence_ref: str,
+              now: str) -> dict:
+    return _call("ss_bind_task", {"doc": doc, "task_id": task_id,
+                                  "entity_ids": entity_ids,
+                                  "evidence_ref": evidence_ref, "now": now})
+
+
+def task_entities(doc: dict, task_id: str) -> list:
+    return _call("ss_task_entities", {"doc": doc, "task_id": task_id})
+
+
+def impact_frontier(doc: dict, seed_ids: list, max_level: int = 2,
+                    change_kind=None):
+    return _call("ss_impact_frontier", {"doc": doc, "seed_ids": seed_ids,
+                                        "max_level": max_level,
+                                        "change_kind": change_kind})
+
+
+def record_observed_impact(doc: dict, task_id: str, evidence_items: list,
+                           now: str) -> dict:
+    return _call("ss_record_observed_impact", {"doc": doc, "task_id": task_id,
+                                               "evidence_items": evidence_items,
+                                               "now": now})
+
+
+def predicted_vs_observed(predicted: dict, observed_doc: dict,
+                          seed_ids: list) -> dict:
+    return _call("ss_predicted_vs_observed", {"predicted": predicted,
+                                              "observed_doc": observed_doc,
+                                              "seed_ids": seed_ids})
+
+
+def structural_events(doc: dict) -> list:
+    return _call("ss_structural_events", {"doc": doc})
+
+
+def diff(old: dict, new: dict) -> dict:
+    return _call("ss_diff", {"old": old, "new": new})
+
+
+def dedupe_observations(doc: dict) -> dict:
+    return _call("ss_dedupe_observations", {"doc": doc})

src/rune/structural_store.py

@@ -85,6 +85,9 @@ def _sorted(obj):
 
 
 def new_doc() -> dict:
+    core = _ss_core()
+    if core is not None:
+        return core.new_doc(now())
     return {
         "schema_version": SCHEMA_VERSION,
         "generated_at": now(),
@@ -143,6 +146,18 @@ def _redact_secrets(obj):
 
 
 def save(doc: dict, path: Path | str) -> Path:
+    core = _ss_core()
+    if core is not None:
+        try:
+            out = core.save(doc, str(path))
+        except RuntimeError as exc:
+            # Kernel errors carry the oracle's ValueError semantics through
+            # the envelope — translate back to the documented contract.
+            raise ValueError(str(exc)) from exc
+        if isinstance(out.get("doc"), dict):
+            doc.clear()
+            doc.update(out["doc"])
+        return Path(out["path"])
     incoming = doc.get("schema_version", SCHEMA_VERSION)
     if incoming != SCHEMA_VERSION:
         raise ValueError(f"schema_version {incoming} != {SCHEMA_VERSION}")
@@ -160,6 +175,10 @@ def canonical_digest(doc: dict) -> str:
     identical semantic content always yields an identical digest."""
     import hashlib
 
+    core = _ss_core()
+    if core is not None:
+        return core.canonical_digest(doc)
+
     payload = json.dumps(_sorted(_redact_secrets(doc)), sort_keys=True)
     return hashlib.sha256(payload.encode("utf-8")).hexdigest()
 
@@ -168,6 +187,13 @@ def snapshot(doc: dict, snapshots_dir: Path, label: str = "", tag: str = "") ->
     """Persist an immutable snapshot. `tag` is a caller-supplied identity
     fragment (Spine uses the deployed SHA; a Project Scaffold may use its
     repo HEAD or 'noident')."""
+    core = _ss_core()
+    if core is not None:
+        out = core.snapshot(doc, str(snapshots_dir), label, tag, now())
+        if isinstance(out.get("doc"), dict):
+            doc.clear()
+            doc.update(out["doc"])
+        return Path(out["path"])
     snapshots_dir = Path(snapshots_dir)
     snapshots_dir.mkdir(parents=True, exist_ok=True)
     name = f"snap-{time.strftime('%Y%m%dT%H%M%SZ', time.gmtime())}-{(tag or 'noident')[:8]}"
@@ -312,6 +338,16 @@ def promote(doc: dict, id: str, state: str, evidence_ref: str) -> dict | None:
 def alias(doc: dict, canonical_id: str, old_id: str) -> dict:
     """RENAME: same conceptual entity, identifier changed. Old references
     stay resolvable (entity() follows alias entries)."""
+    core = _ss_core()
+    if core is not None:
+        try:
+            out = core.alias(doc, canonical_id, old_id, now())
+        except RuntimeError as exc:
+            raise ValueError(str(exc)) from exc
+        if isinstance(out, dict):
+            doc.clear()
+            doc.update(out)
+        return entity(doc, canonical_id)
     e = entity(doc, canonical_id)
     if e is None:
         raise ValueError(f"alias target not found: {canonical_id}")
@@ -344,11 +380,20 @@ def supersede(doc: dict, new_id: str, old_id: str, evidence_ref: str) -> dict:
 def spine_uri(entity_id: str) -> str:
     """Stable cross-graph identity: spine://<kind>/<name>. A future Memory
     record cites this URI without duplicating structural authority."""
+    core = _ss_core()
+    if core is not None:
+        return core.spine_uri(entity_id)
     kind, _, name = entity_id.partition(":")
     return f"spine://{kind}/{name}"
 
 
 def resolve_uri(uri: str) -> str:
+    core = _ss_core()
+    if core is not None:
+        try:
+            return core.resolve_uri(uri)
+        except RuntimeError as exc:
+            raise ValueError(str(exc)) from exc
     scheme, _, rest = uri.partition("://")
     if scheme != "spine":
         raise ValueError(f"not a spine URI: {uri}")
@@ -460,6 +505,9 @@ def resolve_status(e: dict) -> str:
 
 def authority_health(doc: dict) -> dict:
     """Provenance/freshness per projector — provenance only, not monitoring."""
+    core = _ss_core()
+    if core is not None:
+        return core.authority_health(doc)
     return {
         name: {
             "generation": a.get("generation", 0),
@@ -478,6 +526,13 @@ def authority_health(doc: dict) -> dict:
 
 def bind_task(doc: dict, task_id: str, entity_ids: list[str], evidence_ref: str = "") -> dict:
     """TASK/ISSUE/INTENT BINDING: task → affects → entities. Deterministic."""
+    core = _ss_core()
+    if core is not None:
+        out = core.bind_task(doc, task_id, list(entity_ids), evidence_ref, now())
+        if isinstance(out, dict):
+            doc.clear()
+            doc.update(out)
+        return entity(doc, f"task:{task_id}") or {"id": f"task:{task_id}"}
     for eid in entity_ids:
         add_relationship(
             doc,
@@ -490,6 +545,9 @@ def bind_task(doc: dict, task_id: str, entity_ids: list[str], evidence_ref: str
 
 
 def task_entities(doc: dict, task_id: str) -> list[str]:
+    core = _ss_core()
+    if core is not None:
+        return core.task_entities(doc, task_id)
     return sorted(r["to"] for r in relationships(doc, from_id=f"task:{task_id}", type="affects"))
 
 
@@ -502,6 +560,9 @@ def impact_frontier(doc: dict, seed_ids: list[str], max_level: int = 2, change_k
     contract_violation (escalates siblings to obligations). None infers from
     seed kind. Unknown relationship types are obligation-bearing (fail-loud);
     NON_OBLIGATION types and CANDIDATE/STALE facts are skipped."""
+    core = _ss_core()
+    if core is not None:
+        return core.impact_frontier(doc, list(seed_ids), max_level, change_kind)
     seeds = list(dict.fromkeys(seed_ids))
     level0 = sorted(set(seeds))
     level1: set[str] = set()
@@ -578,6 +639,13 @@ def record_observed_impact(doc: dict, task_id: str, evidence_items: list[dict])
     """INDEPENDENT OBSERVED IMPACT: derive what was ACTUALLY affected from
     external evidence (tests, canaries, runtime, deploy) — never by re-running
     the frontier being evaluated. Nothing auto-promotes into structure."""
+    core = _ss_core()
+    if core is not None:
+        out = core.record_observed_impact(doc, task_id, list(evidence_items), now())
+        if isinstance(out.get("doc"), dict):
+            doc.clear()
+            doc.update(out["doc"])
+        return out["result"]
     implicated: set[str] = set()
     for item in evidence_items:
         eid = item.get("entity", "")
@@ -599,6 +667,9 @@ def record_observed_impact(doc: dict, task_id: str, evidence_items: list[dict])
 
 
 def predicted_vs_observed(predicted: dict, observed_doc: dict, seed_ids: list[str]) -> dict:
+    core = _ss_core()
+    if core is not None:
+        return core.predicted_vs_observed(predicted, observed_doc, list(seed_ids))
     observed = impact_frontier(observed_doc, seed_ids)
     pred_set = {
         e
@@ -629,6 +700,9 @@ def predicted_vs_observed(predicted: dict, observed_doc: dict, seed_ids: list[st
 def structural_events(doc: dict) -> list[dict]:
     """GENERIC EVENT/QUERY SEAM: dumb typed structural facts for future
     subscribers. No salience, no policy."""
+    core = _ss_core()
+    if core is not None:
+        return core.structural_events(doc)
     events = []
     for e in doc["entities"]:
         events.append(
@@ -657,6 +731,9 @@ def structural_events(doc: dict) -> list[dict]:
 
 def diff(old: dict, new: dict) -> dict:
     """Deterministic structural delta. No interpretation."""
+    core = _ss_core()
+    if core is not None:
+        return core.diff(old, new)
     oe = {e["id"]: e for e in old.get("entities", [])}
     ne = {e["id"]: e for e in new.get("entities", [])}
     ors = {(r["type"], r["from"], r["to"]) for r in old.get("relationships", [])}
@@ -680,6 +757,13 @@ def diff(old: dict, new: dict) -> dict:
 def dedupe_observations(doc: dict) -> None:
     """Exact-duplicate observation suppression (carry-forward + re-derivation
     both record; distinct repeats are history and stay)."""
+    core = _ss_core()
+    if core is not None:
+        out = core.dedupe_observations(doc)
+        if isinstance(out, dict):
+            doc.clear()
+            doc.update(out)
+        return
     seen: set[str] = set()
     unique = []
     for o in doc.get("observations", []):

src/rust/structural_store/Cargo.lock

@@ -2,6 +2,32 @@
 # It is not intended for manual editing.
 version = 4
 
+[[package]]
+name = "equivalent"
+version = "1.0.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "877a4ace8713b0bcf2a4e7eec82529c029f1d0619886d18145fea96c3ffe5c0f"
+
+[[package]]
+name = "evidence_provenance"
+version = "0.1.0"
+
+[[package]]
+name = "hashbrown"
+version = "0.17.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ed5909b6e89a2db4456e54cd5f673791d7eca6732202bbf2a9cc504fe2f9b84a"
+
+[[package]]
+name = "indexmap"
+version = "2.14.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cc4e190f5d26ca7051642629da2c52fc03bde85a03197c99408dcd291734c855"
+dependencies = [
+ "equivalent",
+ "hashbrown",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.18"
@@ -68,6 +94,7 @@ version = "1.0.151"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14"
 dependencies = [
+ "indexmap",
  "itoa",
  "memchr",
  "serde",
@@ -79,6 +106,7 @@ dependencies = [
 name = "structural_store"
 version = "0.1.0"
 dependencies = [
+ "evidence_provenance",
  "serde",
  "serde_json",
 ]

src/rust/structural_store/Cargo.toml

@@ -8,4 +8,7 @@ crate-type = ["rlib", "cdylib", "staticlib"]
 
 [dependencies]
 serde = { version = "1", features = ["derive"] }
-serde_json = "1"
+# preserve_order: Python dicts round-trip in insertion order — str() of a
+# detail dict and doc-echo order follow the oracle.
+serde_json = { version = "1", features = ["preserve_order"] }
+evidence_provenance = { path = "../evidence_provenance" }

src/rust/structural_store/src/frozen_battery.rs

@@ -0,0 +1,352 @@
+//! frozen_battery — replay the FROZEN oracle truth VERBATIM
+//! (frozen_ss_truth.json, generated by scripts/gen_structural_store_truth.py
+//! against the LIVE Python oracle with RUNE_STRUCTURAL_CORE=python pinned,
+//! clock pinned via structural_store.now, inputs deep-copied per case).
+//!
+//! The canary never imports the Python oracle — parity stays provable after
+//! strangler exits.
+
+use crate::*;
+use serde_json::Value;
+use std::cell::RefCell;
+
+const TRUTH: &str = include_str!("frozen_ss_truth.json");
+const NOW: &str = "2026-10-13T01:02:03Z";
+
+fn truth() -> Value {
+    serde_json::from_str(TRUTH).expect("frozen truth parses")
+}
+
+fn cases(group: &str) -> Vec<Value> {
+    truth()["groups"][group]
+        .as_array()
+        .cloned()
+        .unwrap_or_default()
+}
+
+fn nm(c: &Value) -> &str {
+    c["name"].as_str().unwrap_or("")
+}
+
+/// Recording I/O seam — save/snapshot write into memory, never the disk.
+struct RecIo {
+    writes: RefCell<Vec<(String, String)>>,
+}
+
+impl RecIo {
+    fn new() -> Self {
+        RecIo { writes: RefCell::new(vec![]) }
+    }
+    fn content(&self, path: &str) -> String {
+        self.writes
+            .borrow()
+            .iter()
+            .find(|(p, _)| p == path)
+            .map(|(_, c)| c.clone())
+            .unwrap_or_default()
+    }
+}
+
+impl SsIo for RecIo {
+    fn mkdir_parents(&self, _dir: &str) -> Result<(), String> {
+        Ok(())
+    }
+    fn write_text(&self, path: &str, contents: &str) -> Result<(), String> {
+        self.writes.borrow_mut().push((path.to_string(), contents.to_string()));
+        Ok(())
+    }
+}
+
+fn expect_err(c: &Value, got: Result<impl std::fmt::Debug, String>) {
+    match c["expect"].get("err").and_then(|e| e.as_str()) {
+        Some(want) => {
+            let e = got.expect_err(&format!("case {} should raise", nm(c)));
+            assert_eq!(e, want, "case {} error message", nm(c));
+        }
+        None => {
+            got.unwrap_or_else(|e| panic!("case {} raised: {}", nm(c), e));
+        }
+    }
+}
+
+fn de<'a>(c: &'a Value, keys: &[&str]) -> &'a Value {
+    let mut cur = c;
+    for k in keys {
+        cur = &cur[k];
+    }
+    cur
+}
+
+// ── pure normalizers ────────────────────────────────────────────────────────
+
+#[test]
+fn replay_sorted() {
+    for c in &cases("sorted") {
+        let got = _sorted(de(c, &["args", "value"]));
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_new_doc() {
+    for c in &cases("new_doc") {
+        let got = new_doc(NOW);
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_redact() {
+    for c in &cases("redact") {
+        let got = _redact_secrets(de(c, &["args", "value"]));
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_save() {
+    for c in &cases("save") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let path = format!("out/{}/store.json", nm(c));
+        let io = RecIo::new();
+        let got = save(&mut doc, &path, &io);
+        expect_err(c, got);
+        if c["expect"].get("ok").is_some() {
+            assert_eq!(io.content(&path), de(c, &["expect", "ok", "content"]).as_str().unwrap(), "case {} content", nm(c));
+            assert_eq!(doc, de(c, &["expect", "ok", "doc"]).clone(), "case {} doc", nm(c));
+        }
+    }
+}
+
+#[test]
+fn replay_digest() {
+    for c in &cases("digest") {
+        let got = canonical_digest(de(c, &["args", "doc"]));
+        let want = de(c, &["expect", "ok"]);
+        if want.is_array() {
+            // multi-assert cases: digests of related docs must agree pairwise
+            let digests: Vec<Value> = if c["args"].get("doc_reordered").is_some() {
+                vec![
+                    json!(canonical_digest(de(c, &["args", "doc"]))),
+                    json!(canonical_digest(de(c, &["args", "doc_reordered"]))),
+                ]
+            } else if c["args"].get("doc").is_some() && c["name"] == "secret_never_changes_digest" {
+                let d = de(c, &["args", "doc"]);
+                let d2 = { let mut x = d.clone(); x["token"] = json!("other"); x };
+                vec![json!(canonical_digest(d)), json!(canonical_digest(&d2))]
+            } else {
+                vec![json!(got.clone())]
+            };
+            assert_eq!(json!(digests), want.clone(), "case {}", nm(c));
+        } else {
+            assert_eq!(json!(got), want.clone(), "case {}", nm(c));
+        }
+    }
+}
+
+#[test]
+fn replay_snapshot() {
+    for c in &cases("snapshot") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let dir = de(c, &["args", "dir"]).as_str().unwrap();
+        let label = de(c, &["args", "label"]).as_str().unwrap();
+        let tag = de(c, &["args", "tag"]).as_str().unwrap();
+        let io = RecIo::new();
+        let got = snapshot(&mut doc, dir, label, tag, NOW, &io).expect("snapshot ok");
+        assert_eq!(json!(got), de(c, &["expect", "ok", "path"]).clone(), "case {} path", nm(c));
+        assert_eq!(io.content(&got), de(c, &["expect", "ok", "content"]).as_str().unwrap(), "case {} content", nm(c));
+        assert_eq!(doc, de(c, &["expect", "ok", "doc"]).clone(), "case {} doc", nm(c));
+    }
+}
+
+#[test]
+fn replay_alias() {
+    for c in &cases("alias") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let cid = de(c, &["args", "cid"]).as_str().unwrap();
+        let old = de(c, &["args", "old"]).as_str().unwrap();
+        let got = alias(&mut doc, cid, old, NOW);
+        match c["expect"].get("err") {
+            Some(_) => expect_err(c, got),
+            None => assert_eq!(got.expect("alias ok"), de(c, &["expect", "ok"]).clone(), "case {}", nm(c)),
+        }
+    }
+}
+
+#[test]
+fn replay_spine_uri() {
+    for c in &cases("spine_uri") {
+        let got = spine_uri(de(c, &["args", "entity_id"]).as_str().unwrap());
+        assert_eq!(json!(got), de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_resolve_uri() {
+    for c in &cases("resolve_uri") {
+        let got = resolve_uri(de(c, &["args", "uri"]).as_str().unwrap());
+        expect_err(c, got.clone());
+        if c["expect"].get("ok").is_some() {
+            assert_eq!(json!(got.unwrap()), de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+        }
+    }
+}
+
+#[test]
+fn replay_doc_authorities() {
+    for c in &cases("doc_authorities") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let ret = _doc_authorities(&mut doc).clone();
+        assert_eq!(json!(ret), de(c, &["expect", "ok", "ret"]).clone(), "case {} ret", nm(c));
+        assert_eq!(doc, de(c, &["expect", "ok", "doc"]).clone(), "case {} doc", nm(c));
+    }
+}
+
+#[test]
+fn replay_authority_health() {
+    for c in &cases("authority_health") {
+        let got = authority_health(de(c, &["args", "doc"]));
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_bind_task() {
+    for c in &cases("bind_task") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let task_id = de(c, &["args", "task_id"]).as_str().unwrap();
+        let ids: Vec<String> = de(c, &["args", "entity_ids"])
+            .as_array().unwrap()
+            .iter().map(|x| x.as_str().unwrap().to_string()).collect();
+        let ev = de(c, &["args", "evidence_ref"]).as_str().unwrap();
+        bind_task(&mut doc, task_id, &ids, ev, NOW);
+        let ret = entity_ref(&doc, &format!("task:{}", task_id))
+            .cloned()
+            .unwrap_or(json!({"id": format!("task:{}", task_id)}));
+        assert_eq!(json!(ret), de(c, &["expect", "ok", "ret"]).clone(), "case {} ret", nm(c));
+        assert_eq!(doc, de(c, &["expect", "ok", "doc"]).clone(), "case {} doc", nm(c));
+    }
+}
+
+#[test]
+fn replay_task_entities() {
+    for c in &cases("task_entities") {
+        let got = task_entities(de(c, &["args", "doc"]), de(c, &["args", "task_id"]).as_str().unwrap());
+        assert_eq!(json!(got), de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_impact_frontier() {
+    for c in &cases("impact_frontier") {
+        let got = impact_frontier(
+            de(c, &["args", "doc"]),
+            &de(c, &["args", "seeds"]).as_array().unwrap()
+                .iter().map(|x| x.as_str().unwrap().to_string()).collect::<Vec<_>>(),
+            de(c, &["args", "max_level"]).as_i64().unwrap_or(2),
+            de(c, &["args", "change_kind"]).as_str(),
+        );
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_observed_impact() {
+    for c in &cases("observed_impact") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let items: Vec<Value> = de(c, &["args", "items"]).as_array().unwrap().clone();
+        let got = record_observed_impact(&mut doc, "T1", &items, NOW);
+        assert_eq!(json!(got), de(c, &["expect", "ok", "result"]).clone(), "case {} result", nm(c));
+        assert_eq!(doc, de(c, &["expect", "ok", "doc"]).clone(), "case {} doc", nm(c));
+    }
+}
+
+#[test]
+fn replay_predicted_vs_observed() {
+    for c in &cases("predicted_vs_observed") {
+        let seeds: Vec<String> = de(c, &["args", "seeds"]).as_array().unwrap()
+            .iter().map(|x| x.as_str().unwrap().to_string()).collect();
+        let got = predicted_vs_observed(de(c, &["args", "predicted"]), de(c, &["args", "observed_doc"]), &seeds);
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_structural_events() {
+    for c in &cases("events") {
+        let got = structural_events(de(c, &["args", "doc"]));
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_diff() {
+    for c in &cases("diff") {
+        let got = diff(de(c, &["args", "old"]), de(c, &["args", "new"]));
+        assert_eq!(got, de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_dedupe() {
+    for c in &cases("dedupe") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        dedupe_observations(&mut doc);
+        assert_eq!(doc, de(c, &["expect", "ok", "doc"]).clone(), "case {} doc", nm(c));
+        assert!(de(c, &["expect", "ok", "ret"]).is_null(), "case {} ret", nm(c));
+    }
+}
+
+// ── fixed-kernel regressions ────────────────────────────────────────────────
+
+#[test]
+fn replay_promote_fix() {
+    for c in &cases("promote_fix") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let id = de(c, &["args", "id"]).as_str().unwrap();
+        let state = de(c, &["args", "state"]).as_str().unwrap();
+        let ev = de(c, &["args", "ev"]).as_str().unwrap();
+        let got = promote(&mut doc, id, state, ev, NOW);
+        let ret = de(c, &["expect", "ok", "ret"]);
+        let want_doc = de(c, &["expect", "ok", "doc"]);
+        if ret.is_null() {
+            assert_eq!(got, Ok(None), "case {} ret", nm(c));
+        } else {
+            // oracle promote returns the ENTITY; the ret maturity is the
+            // observed outcome (doc compare below proves the full shape)
+            let got_str = got.as_ref().ok().and_then(|o| o.as_deref());
+            assert_eq!(got_str, Some(ret["maturity"].as_str().unwrap()), "case {} ret", nm(c));
+        }
+        assert_eq!(doc, want_doc.clone(), "case {} doc", nm(c));
+    }
+}
+
+#[test]
+fn replay_resolve_status_fix() {
+    for c in &cases("resolve_status_fix") {
+        let got = resolve_status(de(c, &["args", "entity"]));
+        assert_eq!(json!(got), de(c, &["expect", "ok"]).clone(), "case {}", nm(c));
+    }
+}
+
+#[test]
+fn replay_supersede_fix() {
+    for c in &cases("supersede_fix") {
+        let mut doc = de(c, &["args", "doc"]).clone();
+        let nid = de(c, &["args", "new"]).as_str().unwrap();
+        let oid = de(c, &["args", "old"]).as_str().unwrap();
+        let ev = de(c, &["args", "ev"]).as_str().unwrap();
+        supersede(&mut doc, nid, oid, ev, NOW).expect("supersede ok");
+        assert_eq!(doc, de(c, &["expect", "ok", "doc"]).clone(), "case {}", nm(c));
+    }
+}
+
+// ── the core probe mirror (RUNE_STRUCTURAL_CORE=python pin) ─────────────────
+
+#[test]
+fn ss_core_pin_semantics() {
+    assert!(_ss_core(Some("python")).is_none(), "python pin disables the core");
+    assert!(_ss_core(None).is_some(), "unset pin activates the core");
+    assert!(_ss_core(Some("auto")).is_some(), "auto activates the core");
+    assert_eq!(_ss_core(Some("auto")).unwrap().id(), "rust");
+}

src/rust/structural_store/src/lib.rs

@@ -2,7 +2,9 @@
 //! PARITY port of the core data structures from src/rune/structural_store.py.
 use serde::{Deserialize, Serialize};
 
-pub const SCHEMA_VERSION: i32 = 3;
+// Oracle law: structural_store.SCHEMA_VERSION == 1 (shared with spine stores).
+// (Previous port carried a stale 3 — corrected against the live oracle.)
+pub const SCHEMA_VERSION: i32 = 1;
 pub const MATURITY_ORDER: &[&str] = &["UNRUN", "DECLARED", "PRESENT", "VERIFIED"];
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -75,7 +77,7 @@ impl Default for StoreDoc {
 use std::ffi::{CStr, CString};
 use std::os::raw::c_char;
 
-fn read_arg(ptr: *const c_char) -> Result<String, String> {
+pub(crate) fn read_arg(ptr: *const c_char) -> Result<String, String> {
     if ptr.is_null() {
         return Err("null argument pointer".to_string());
     }
@@ -85,7 +87,7 @@ fn read_arg(ptr: *const c_char) -> Result<String, String> {
         .map_err(|e| format!("argument is not valid UTF-8: {}", e))
 }
 
-fn envelope(value: Result<serde_json::Value, String>) -> *mut c_char {
+pub(crate) fn envelope(value: Result<serde_json::Value, String>) -> *mut c_char {
     let body = match value {
         Ok(v) => serde_json::json!({"ok": true, "value": v}),
         Err(e) => serde_json::json!({"ok": false, "err": e}),
@@ -95,11 +97,11 @@ fn envelope(value: Result<serde_json::Value, String>) -> *mut c_char {
         .unwrap_or(std::ptr::null_mut())
 }
 
-fn parse_arg(text: &str) -> Result<serde_json::Value, String> {
+pub(crate) fn parse_arg(text: &str) -> Result<serde_json::Value, String> {
     serde_json::from_str(text).map_err(|e| format!("argument JSON parse failed: {}", e))
 }
 
-fn get_str<'a>(v: &'a serde_json::Value, key: &str) -> Option<&'a str> {
+pub(crate) fn get_str<'a>(v: &'a serde_json::Value, key: &str) -> Option<&'a str> {
     v.get(key).and_then(|x| x.as_str())
 }
 
@@ -223,7 +225,9 @@ pub fn is_known_maturity(state: &str) -> bool {
 }
 
 /// Evidence-gated promotion. Demotion is never silent: recorded, not applied.
-/// Returns Ok(Some(updated maturity)) / Ok(demotion-rejected marker) / Err.
+/// Oracle contract order: unknown id → Ok(None) (silent, the caller surfaces
+/// the miss); then evidence required; then known state required. Entity
+/// resolution follows alias entries (oracle entity()).
 pub fn promote(
     doc: &mut serde_json::Value,
     id: &str,
@@ -231,8 +235,8 @@ pub fn promote(
     evidence_ref: &str,
     now_iso: &str,
 ) -> Result<Option<String>, String> {
-    if !is_known_maturity(state) {
-        return Err(format!("unknown maturity state: {}", state));
+    if !crate::gap::has_entity(doc, id) {
+        return Ok(None);
     }
     if evidence_ref.is_empty() {
         return Err(format!(
@@ -240,13 +244,10 @@ pub fn promote(
             id, state
         ));
     }
-    let current = doc
-        .get("entities")
-        .and_then(|e| e.as_array())
-        .and_then(|arr| {
-            arr.iter()
-                .find(|e| e.get("id").and_then(|x| x.as_str()) == Some(id))
-        })
+    if !is_known_maturity(state) {
+        return Err(format!("unknown maturity state: {}", state));
+    }
+    let current = crate::gap::entity_ref(doc, id)
         .and_then(|e| e.get("maturity"))
         .and_then(|m| m.as_str())
         .unwrap_or("DECLARED")
@@ -264,7 +265,7 @@ pub fn promote(
         }
         return Ok(Some(current));
     }
-    let entity = find_entity_mut(doc, id).ok_or_else(|| "entity-not-found".to_string())?;
+    let entity = crate::gap::entity_mut(doc, id).ok_or_else(|| "entity-not-found".to_string())?;
     if let Some(e) = entity.as_object_mut() {
         e.insert("maturity".into(), serde_json::json!(state));
         let mut refs: Vec<String> = e
@@ -285,37 +286,23 @@ pub fn promote(
     Ok(Some(state.to_string()))
 }
 
-fn find_entity_mut<'a>(
-    doc: &'a mut serde_json::Value,
-    id: &str,
-) -> Option<&'a mut serde_json::Value> {
-    doc.get_mut("entities")
-        .and_then(|e| e.as_array_mut())?
-        .iter_mut()
-        .find(|e| e.get("id").and_then(|x| x.as_str()) == Some(id))
-}
-
-fn now_iso() -> String {
-    // oracle: datetime.now(timezone.utc).isoformat() — the digest tests mask
-    // timestamps; the store's monotonicity doesn't depend on the value.
-    "1970-01-01T00:00:00+00:00".into()
-}
-
 /// resolve_status — smallest coherent entity state (oracle-identical).
-pub fn resolve_status(e: &serde_json::Value) -> &'static str {
+/// An explicit `status` passes through VERBATIM (oracle: `if e.get("status")`
+/// returns it as-is); ext.conflicts is Python-truthy (non-empty), not
+/// merely present.
+pub fn resolve_status(e: &serde_json::Value) -> String {
     if let Some(s) = e.get("status").and_then(|s| s.as_str()) {
         if !s.is_empty() {
-            // static return tied to input would need leaks; return via match
-            return match_status(s);
+            return s.to_string();
         }
     }
     let has_conflicts = e
         .get("ext")
         .and_then(|x| x.get("conflicts"))
-        .map(|c| !c.is_null())
+        .map(py_truthy)
         .unwrap_or(false);
     if has_conflicts {
-        return "CONFLICTING";
+        return "CONFLICTING".to_string();
     }
     let maturity = e.get("maturity").and_then(|m| m.as_str()).unwrap_or("");
     let has_evidence = e
@@ -324,25 +311,29 @@ pub fn resolve_status(e: &serde_json::Value) -> &'static str {
         .map(|a| !a.is_empty())
         .unwrap_or(false);
     if (maturity == "EXERCISED" || maturity == "LIVE_PROVEN") && has_evidence {
-        return "CURRENT";
+        return "CURRENT".to_string();
     }
     if !maturity.is_empty() && maturity != "DECLARED" && !has_evidence {
-        return "CONFLICTING";
+        return "CONFLICTING".to_string();
     }
-    "UNKNOWN"
+    "UNKNOWN".to_string()
 }
 
-fn match_status(s: &str) -> &'static str {
-    match s {
-        "SUPERSEDED" => "SUPERSEDED",
-        "CURRENT" => "CURRENT",
-        "CONFLICTING" => "CONFLICTING",
-        "UNKNOWN" => "UNKNOWN",
-        _ => "UNKNOWN",
+/// Python truthiness for the JSON values resolve_status tests.
+fn py_truthy(v: &serde_json::Value) -> bool {
+    match v {
+        serde_json::Value::Null => false,
+        serde_json::Value::Bool(b) => *b,
+        serde_json::Value::Number(n) => n.as_f64().map(|f| f != 0.0).unwrap_or(false),
+        serde_json::Value::String(s) => !s.is_empty(),
+        serde_json::Value::Array(a) => !a.is_empty(),
+        serde_json::Value::Object(o) => !o.is_empty(),
     }
 }
 
 /// supersede — replacement with evidence; old identity stays resolvable.
+/// Entity resolution follows alias entries; the supersedes relationship goes
+/// through the oracle add_relationship law (evidence_refs + observed_at).
 pub fn supersede(
     doc: &mut serde_json::Value,
     new_id: &str,
@@ -353,17 +344,21 @@ pub fn supersede(
     if evidence_ref.is_empty() {
         return Err("supersession requires evidence".into());
     }
-    if find_entity(doc, new_id).is_none() || find_entity(doc, old_id).is_none() {
+    if !crate::gap::has_entity(doc, new_id) || !crate::gap::has_entity(doc, old_id) {
         return Err("supersede requires both entities to exist".into());
     }
-    doc.get_mut("relationships")
-        .and_then(|r| r.as_array_mut())
-        .unwrap_or(&mut Vec::new())
-        .push(serde_json::json!({
-            "type": "supersedes", "from": new_id, "to": old_id,
-            "evidence_refs": [evidence_ref],
-        }));
-    if let Some(old) = find_entity_mut(doc, old_id) {
+    crate::gap::add_relationship(
+        doc,
+        crate::gap::RelSpec {
+            rel_type: "supersedes",
+            from_id: new_id,
+            to_id: old_id,
+            evidence_refs: &[evidence_ref.to_string()],
+            ext: None,
+            now_iso,
+        },
+    );
+    if let Some(old) = crate::gap::entity_mut(doc, old_id) {
         if let Some(o) = old.as_object_mut() {
             o.insert("superseded_by".into(), serde_json::json!(new_id));
             o.insert("status".into(), serde_json::json!("SUPERSEDED"));
@@ -378,6 +373,7 @@ pub fn supersede(
     Ok(())
 }
 
+#[cfg(test)]
 fn find_entity<'a>(doc: &'a serde_json::Value, id: &str) -> Option<&'a serde_json::Value> {
     doc.get("entities")
         .and_then(|e| e.as_array())?
@@ -480,6 +476,7 @@ pub fn authority_observed(
     doc: &mut serde_json::Value,
     rel: Option<&mut serde_json::Value>,
     authority: &str,
+    now_iso: &str,
 ) {
     let auths = doc
         .as_object_mut()
@@ -493,12 +490,12 @@ pub fn authority_observed(
         .or_insert(json!({"generation": 0, "observed": []}));
     let generation = a["generation"].as_i64().unwrap_or(0) + if rel.is_none() { 1 } else { 0 };
     a["generation"] = json!(generation);
-    a["last_success"] = json!(now_iso());
+    a["last_success"] = json!(now_iso);
     a["status"] = json!("CURRENT");
     if let Some(rel) = rel {
         rel["authority"] = json!(authority);
         rel["status"] = json!("CURRENT");
-        rel["observed_at"] = json!(now_iso());
+        rel["observed_at"] = json!(now_iso);
         let oid = format!(
             "{}|{}|{}",
             rel["type"].as_str().unwrap_or(""),
@@ -513,8 +510,9 @@ pub fn authority_observed(
 }
 
 /// authority_failed — the authority's facts go STALE (not silently fresh);
-/// the authority is recorded DEGRADED with the error.
-pub fn authority_failed(doc: &mut serde_json::Value, authority: &str, error: &str) {
+/// the authority is recorded DEGRADED with the error, and the degradation is
+/// OBSERVED (oracle appends the authority_degraded observation).
+pub fn authority_failed(doc: &mut serde_json::Value, authority: &str, error: &str, now_iso: &str) {
     let auths = doc
         .as_object_mut()
         .unwrap()
@@ -525,8 +523,8 @@ pub fn authority_failed(doc: &mut serde_json::Value, authority: &str, error: &st
         .unwrap()
         .entry(authority.to_string())
         .or_insert(json!({"generation": 0, "observed": []}));
-    a["last_attempt"] = json!(now_iso());
-    a["last_error"] = json!(&error[..error.len().min(200)]);
+    a["last_attempt"] = json!(now_iso);
+    a["last_error"] = json!(&error[..error.chars().count().min(200)]);
     a["status"] = json!("DEGRADED");
     if let Some(rels) = doc.get_mut("relationships").and_then(|r| r.as_array_mut()) {
         for rel in rels.iter_mut() {
@@ -535,6 +533,13 @@ pub fn authority_failed(doc: &mut serde_json::Value, authority: &str, error: &st
             }
         }
     }
+    if let Some(obs) = doc.get_mut("observations").and_then(|o| o.as_array_mut()) {
+        obs.push(json!({
+            "kind": "authority_degraded", "authority": authority,
+            "error": &error[..error.chars().count().min(200)],
+            "observed_at": now_iso,
+        }));
+    }
 }
 
 /// reconcile_authorities — facts owned by an authority that did not
@@ -599,7 +604,8 @@ pub fn reconcile_authorities(doc: &mut serde_json::Value) -> Vec<serde_json::Val
 }
 
 /// mark_conflict — BOTH observations preserved; entity CONFLICTING; never
-/// last-wins (oracle-identical).
+/// last-wins (oracle-identical). Entity resolution follows alias entries.
+#[allow(clippy::too_many_arguments)] // oracle signature: the two claims travel together
 pub fn mark_conflict(
     doc: &mut serde_json::Value,
     entity_id: &str,
@@ -608,14 +614,15 @@ pub fn mark_conflict(
     authority_b: &str,
     claim_b: &str,
     evidence_ref: &str,
+    now_iso: &str,
 ) -> Result<(), String> {
-    if find_entity_mut(doc, entity_id).is_none() {
+    if !crate::gap::has_entity(doc, entity_id) {
         return Err(format!("conflict on unknown entity: {}", entity_id));
     }
     if evidence_ref.is_empty() {
         return Err("conflict recording requires evidence".into());
     }
-    let entity = find_entity_mut(doc, entity_id).unwrap();
+    let entity = crate::gap::entity_mut(doc, entity_id).unwrap();
     let e = entity.as_object_mut().unwrap();
     let ext = e.entry("ext").or_insert(json!({}));
     let conflicts = ext
@@ -626,7 +633,7 @@ pub fn mark_conflict(
     conflicts.as_array_mut().unwrap().push(json!({
         "authority_a": authority_a, "claim_a": claim_a,
         "authority_b": authority_b, "claim_b": claim_b,
-        "evidence": evidence_ref, "observed_at": now_iso(),
+        "evidence": evidence_ref, "observed_at": now_iso,
     }));
     e.insert("status".into(), json!("CONFLICTING"));
     if let Some(obs) = doc.get_mut("observations").and_then(|o| o.as_array_mut()) {
@@ -634,7 +641,7 @@ pub fn mark_conflict(
             "kind": "authority_conflict", "entity": entity_id,
             "authority_a": authority_a, "claim_a": claim_a,
             "authority_b": authority_b, "claim_b": claim_b,
-            "observed_at": now_iso(),
+            "observed_at": now_iso,
         }));
     }
     Ok(())
@@ -654,7 +661,7 @@ mod authority_battery {
     fn authority_observed_lifecycle() {
         let mut d = doc();
         let mut rel = json!({"type": "owns", "from": "a", "to": "b"});
-        authority_observed(&mut d, Some(&mut rel), "spine");
+        authority_observed(&mut d, Some(&mut rel), "spine", "2026-10-13T00:00:00Z");
         assert_eq!(rel["status"], "CURRENT");
         assert_eq!(rel["authority"], "spine");
         let gen = d["authority_generations"]["spine"]["generation"]
@@ -662,7 +669,7 @@ mod authority_battery {
             .unwrap();
         assert_eq!(gen, 0); // rel observation does NOT bump the generation
                             // bare observation bumps
-        authority_observed(&mut d, None, "spine");
+        authority_observed(&mut d, None, "spine", "2026-10-13T00:00:00Z");
         assert_eq!(
             d["authority_generations"]["spine"]["generation"]
                 .as_i64()
@@ -678,10 +685,16 @@ mod authority_battery {
             {"type": "owns", "from": "a", "to": "b", "authority": "spine", "status": "CURRENT"},
             {"type": "owns", "from": "c", "to": "d", "status": "CURRENT"},
         ]);
-        authority_failed(&mut d, "spine", "boom");
+        authority_failed(&mut d, "spine", "boom", "2026-10-13T00:00:00Z");
         assert_eq!(d["relationships"][0]["status"], "STALE");
         assert_eq!(d["relationships"][1]["status"], "CURRENT"); // unowned untouched
         assert_eq!(d["authority_generations"]["spine"]["status"], "DEGRADED");
+        // the degradation is OBSERVED, not silent (oracle parity fix)
+        assert!(d["observations"]
+            .as_array()
+            .unwrap()
+            .iter()
+            .any(|o| o["kind"] == "authority_degraded" && o["authority"] == "spine"));
     }
 
     #[test]
@@ -710,6 +723,7 @@ mod authority_battery {
             "b",
             "claim two",
             "ev-1",
+            "2026-10-13T00:00:00Z",
         )
         .expect("conflict ok");
         let e = &d["entities"][0];
@@ -718,7 +732,15 @@ mod authority_battery {
         let c = &e["ext"]["conflicts"][0];
         assert_eq!(c["claim_a"], "claim one");
         assert_eq!(c["claim_b"], "claim two"); // BOTH preserved, never last-wins
-        assert!(mark_conflict(&mut d, "nope", "a", "x", "b", "y", "ev").is_err());
-        assert!(mark_conflict(&mut d, "module:x", "a", "x", "b", "y", "").is_err());
+        assert!(mark_conflict(&mut d, "nope", "a", "x", "b", "y", "ev", "t").is_err());
+        assert!(mark_conflict(&mut d, "module:x", "a", "x", "b", "y", "", "t").is_err());
     }
 }
+
+// ── dichotomy-wave gap-closing port (contract factory-gap-ss) ──────────────
+
+pub mod gap;
+pub use gap::*;
+
+#[cfg(test)]
+mod frozen_battery;

tests/test_structural_store_delegation.py

@@ -0,0 +1,191 @@
+"""test_structural_store_delegation.py — dichotomy-wave gap-closing tests.
+
+The facade must behave identically with the Rust core LIVE (RUNE_STRUCTURAL_CORE
+unset, libstructural_store.so built) and with the PYTHON ORACLE pinned
+(RUNE_STRUCTURAL_CORE=python). Each battery case runs through both modes and
+the results must be identical (timestamps masked by the pinned clock).
+"""
+import copy
+import json
+import os
+import sys
+from pathlib import Path
+
+import pytest
+
+ROOT = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(ROOT / "src"))
+
+from rune import ss_loader  # noqa: E402
+from rune import structural_store as ss  # noqa: E402
+
+FIXED_NOW = "2026-10-13T01:02:03Z"
+
+needs_live = pytest.mark.skipif(
+    not ss_loader.available(), reason="Rust structural_store core not built"
+)
+
+
+@pytest.fixture(autouse=True)
+def _pin_clock(monkeypatch):
+    import calendar
+    import time as _time
+    import types
+
+    monkeypatch.setattr(ss, "now", lambda: FIXED_NOW)
+    epoch = calendar.timegm(_time.strptime(FIXED_NOW, "%Y-%m-%dT%H:%M:%SZ"))
+    # snapshot() stamps straight off time.strftime/gmtime
+    monkeypatch.setattr(
+        ss, "time",
+        types.SimpleNamespace(
+            gmtime=lambda: _time.gmtime(epoch),
+            strftime=lambda fmt, t=None: _time.strftime(
+                fmt, t if t is not None else _time.gmtime()),
+        ),
+    )
+
+
+def run_both_modes(call):
+    """Run `call(ss_module)` under the pinned oracle and the live core;
+    results must be identical."""
+    results = []
+    for pin in (True, False):
+        if pin:
+            monkey = pytest.MonkeyPatch()
+            monkey.setenv("RUNE_STRUCTURAL_CORE", "python")
+        else:
+            if not ss_loader.available():
+                pytest.skip("Rust core not built")
+            monkey = pytest.MonkeyPatch()
+            monkey.delenv("RUNE_STRUCTURAL_CORE", raising=False)
+        try:
+            results.append(call(ss))
+        finally:
+            monkey.undo()
+    assert json.dumps(results[0], sort_keys=True) == json.dumps(
+        results[1], sort_keys=True
+    ), f"modes diverged:\n{json.dumps(results[0], sort_keys=True)}\n{json.dumps(results[1], sort_keys=True)}"
+    return results[0]
+
+
+def make_doc():
+    d = ss.new_doc()
+    ss.add_entity(d, id="m:a", kind="module")
+    ss.add_entity(d, id="m:b", kind="widget_component")
+    ss.add_entity(d, id="c:api", kind="contract")
+    ss.add_relationship(d, type="implements", from_id="m:a", to_id="c:api", evidence_refs=["ev"])
+    ss.add_relationship(d, type="depends_on", from_id="m:b", to_id="m:a", evidence_refs=["ev"])
+    ss.promote(d, "m:a", "WIRED", "ev-2")
+    return d
+
+
+class TestDelegationDualMode:
+    def test_new_doc_and_uris(self):
+        def call(m):
+            out = {"doc": m.new_doc(), "uri": m.spine_uri("project:alpha")}
+            try:
+                out["resolved"] = m.resolve_uri("spine://project/alpha")
+            except ValueError:
+                out["resolved"] = None
+            try:
+                m.resolve_uri("bogus://x")
+                out["bad_uri"] = None
+            except ValueError as exc:
+                out["bad_uri"] = str(exc)
+            return out
+
+        run_both_modes(call)
+
+    def test_save_snapshot_digest_bytes(self, tmp_path):
+        def call(m):
+            d = make_doc.__wrapped__() if hasattr(make_doc, "__wrapped__") else None
+            doc = _make_doc(m)
+            p = m.save(doc, tmp_path / "a" / "s.json")
+            snap = m.snapshot(doc, tmp_path / "snaps", "l1", tag="deadbeef123")
+            return {
+                "bytes": p.read_text(),
+                "snapshot_name": snap.name,
+                "snapshot_bytes": snap.read_text(),
+                "digest": m.canonical_digest(doc),
+                "doc_schema": doc["schema_version"],
+            }
+
+        run_both_modes(call)
+
+    def test_alias_error_and_success(self):
+        def call(m):
+            doc = _make_doc(m)
+            e = m.alias(doc, "m:b", "m:b-old")
+            with pytest.raises(ValueError):
+                m.alias(doc, "m:zz", "m:x")
+            with pytest.raises(ValueError):
+                m.alias(doc, "m:a", "m:a")
+            return {"entity": e, "doc": doc}
+
+        run_both_modes(call)
+
+    def test_frontier_events_diff_dedupe(self):
+        def call(m):
+            doc = _make_doc(m)
+            d2 = copy.deepcopy(doc)
+            m.dedupe_observations(d2)
+            return {
+                "frontier": m.impact_frontier(doc, ["c:api"], change_kind=None),
+                "frontier_impl": m.impact_frontier(doc, ["m:a"], change_kind=None),
+                "frontier_violation": m.impact_frontier(doc, ["c:api"], change_kind="contract_violation"),
+                "events": m.structural_events(doc),
+                "diff": m.diff(doc, copy.deepcopy(doc)),
+                "deduped": d2,
+                "task_entities": m.task_entities(doc, "T1"),
+                "health": m.authority_health(doc),
+            }
+
+        run_both_modes(call)
+
+    def test_bind_task_and_observed_impact(self):
+        def call(m):
+            doc = _make_doc(m)
+            m.bind_task(doc, "T1", ["m:a", "m:b"], "ev-t")
+            r = m.record_observed_impact(
+                doc, "T2", [{"entity": "m:a", "source": "test", "detail": "x" * 400}]
+            )
+            return {"doc": doc, "observed": r}
+
+        run_both_modes(call)
+
+    def test_sorted_redact_and_authorities(self):
+        def call(m):
+            v = {"b": 2, "a": [{"z": 1, "y": 2}, "é", True, None], "token": "s3cret"}
+            doc = _make_doc(m)
+            m._doc_authorities(doc)["p1"] = {"generation": 1, "observed": []}
+            return {
+                "sorted": m._sorted(v),
+                "redacted": m._redact_secrets(v),
+                "doc": doc,
+            }
+
+        run_both_modes(call)
+
+    def test_promote_contract_unchanged(self):
+        def call(m):
+            doc = _make_doc(m)
+            e = m.promote(doc, "m:a", "REACHABLE", "ev-3")
+            with pytest.raises(ValueError):
+                m.promote(doc, "m:a", "WIRED", "")
+            with pytest.raises(ValueError):
+                m.promote(doc, "m:a", "BOGUS", "ev")
+            assert m.promote(doc, "m:zz", "PRESENT", "ev") is None
+            return {"entity": e, "doc": doc}
+
+        run_both_modes(call)
+
+
+def _make_doc(m):
+    d = m.new_doc()
+    m.add_entity(d, id="m:a", kind="module")
+    m.add_entity(d, id="m:b", kind="widget_component")
+    m.add_entity(d, id="c:api", kind="contract")
+    m.add_relationship(d, type="implements", from_id="m:a", to_id="c:api", evidence_refs=["ev"])
+    m.add_relationship(d, type="depends_on", from_id="m:b", to_id="m:a", evidence_refs=["ev"])
+    m.promote(d, "m:a", "WIRED", "ev-2")
+    return d