feat(spine): hardening — authority-owned reconciliation, contract semantics, durable identity

T1: impact_frontier carries change_kind — contract_change expands all
parties; implementation_change (contract preserved) demotes sibling
implementers to contextual_not_obligations (sharing a contract is not
being affected by a sibling's internal change); contract_violation
escalates. Pre-hardening test expectation that accepted siblings as
direct obligations was expectation-laundering — corrected with the
architectural justification recorded.

T3/T4: relationships carry authority+status; per-projector generation
records; reconcile_authorities stales facts their owner no longer
observes (marked + drift-reported + frontier-excluded, never deleted —
open-world preserved, truth marked); authority_failed degrades; both
sides of an authority conflict preserved (never last-wins);
resolve_status: UNKNOWN/CURRENT/STALE/CONFLICTING/SUPERSEDED — 'not
checked' is never healthy.

T5: alias (rename) + supersede (replacement) — old spine:// references
stay meaningful; duplicate entity/resolve_uri defs removed (shadowing
trap). T2: record_observed_impact builds the observed set from external
evidence items (test/canary/runtime authority), not frontier-rerun.
CANDIDATE status for evidence-free (model-speculated) relationships.
65/65 spine tests, 1379 suite, gates green.

src/rune/spine.py

@@ -141,14 +141,15 @@ def add_relationship(doc: dict, *, type: str, from_id: str, to_id: str,
         rel["ext"] = dict(ext)
     if not any(r["type"] == type and r["from"] == from_id and r["to"] == to_id
                for r in doc["relationships"]):
+        # MODEL SPECULATION GUARD: a relationship proposed without evidence
+        # is CANDIDATE — representable (open world) but never structural
+        # truth: excluded from frontiers until evidence promotes it.
+        if not rel["evidence_refs"] and "status" not in rel:
+            rel["status"] = "CANDIDATE"
         doc["relationships"].append(rel)
     return rel
 
 
-def entity(doc: dict, id: str) -> dict | None:
-    return next((e for e in doc["entities"] if e["id"] == id), None)
-
-
 def relationships(doc: dict, *, from_id=None, to_id=None, type=None) -> list[dict]:
     out = []
     for r in doc["relationships"]:
@@ -336,11 +337,13 @@ def build(manifest_path: Path | None = None) -> dict:
         consumers = sorted(c for c, deps in graph.items() if m in deps and c != m)
         deps = sorted(d for d in graph.get(m, set()) if d in mods and d != m)
         for c in consumers:
-            add_relationship(doc, type="consumed_by", from_id=mid,
-                             to_id=f"module:{c}", evidence_refs=[f"ast-import:{m}"])
+            rel = add_relationship(doc, type="consumed_by", from_id=mid,
+                                   to_id=f"module:{c}", evidence_refs=[f"ast-import:{m}"])
+            _authority_observed(doc, rel, "ast-import-projector")
         for d in deps:
-            add_relationship(doc, type="depends_on", from_id=mid,
-                             to_id=f"module:{d}", evidence_refs=[f"ast-import:{m}"])
+            rel = add_relationship(doc, type="depends_on", from_id=mid,
+                                   to_id=f"module:{d}", evidence_refs=[f"ast-import:{m}"])
+            _authority_observed(doc, rel, "ast-import-projector")
         # WIRED: importers exist, or registry wiring (canary registry)
         if consumers or m in registry:
             _derive("WIRED", f"ast-imports:{len(consumers)}" if consumers
@@ -354,16 +357,25 @@ def build(manifest_path: Path | None = None) -> dict:
     # 3. runtime/deployment observation (dynamic, bounded)
     obs = _deployed_observation()
     if obs:
+        _authority_observed(doc, None, "deploy-identity-projector")  # generation bump
         dep = add_entity(doc, id="deployment:current", kind="deployment_surface",
                          identities=obs, evidence_refs=["_build_identity.json"],
                          maturity="LIVE_PROVEN")
         for e in doc["entities"]:
             if e["kind"] == "module" and (DEPLOYED_LINK / e["identities"].get("source", "")).is_file():
-                add_relationship(doc, type="deployed_as", from_id=e["id"],
-                                 to_id="deployment:current",
-                                 evidence_refs=[f"deployed-tree:{obs['deployed_sha']}"])
+                rel = add_relationship(doc, type="deployed_as", from_id=e["id"],
+                                       to_id="deployment:current",
+                                       evidence_refs=[f"deployed-tree:{obs['deployed_sha']}"])
+                _authority_observed(doc, rel, "deploy-identity-projector")
         doc["observations"].append({"kind": "deployed_identity", **obs, "observed_at": now()})
 
+    # reconcile: facts owned by projectors that no longer observe them
+    # become STALE (marked, never deleted — history survives; open-world
+    # preservation applies to everything, staleness is the truth marker)
+    staled = reconcile_authorities(doc)
+    for s in staled:
+        doc["observations"].append({"kind": "fact_staled", **s, "observed_at": now()})
+
     # exact-duplicate observation suppression (carry-forward + this pass
     # both record deployed_identity etc.); distinct repeats are history
     seen_obs = set()
@@ -437,6 +449,13 @@ def drift(doc: dict) -> list[dict]:
             out.append({"class": "evidence_conflict", "entity": e["id"],
                         "detail": f"claims {e['maturity']} with zero evidence refs"})
 
+    # 4c. stale facts — owned by an authority that no longer observes them
+    for rel in doc["relationships"]:
+        if rel.get("status") == "STALE":
+            out.append({"class": "stale_fact", "entity": f"{rel['from']}->{rel['to']}",
+                        "detail": f"{rel['type']} edge; authority {rel.get('authority')} "
+                                  "no longer observes it"})
+
     # 5. relationship disappeared (store claims consumers that no longer import)
     for r in relationships(doc, type="consumed_by"):
         m = r["from"].split(":", 1)[1]
@@ -565,15 +584,6 @@ def spine_uri(entity_id: str) -> str:
     return f"spine://{kind}/{name}"
 
 
-def resolve_uri(uri: str) -> str:
-    """spine://<kind>/<name> → internal entity id (id is NOT a path)."""
-    scheme, _, rest = uri.partition("://")
-    if scheme != "spine":
-        raise ValueError(f"not a spine URI: {uri}")
-    kind, _, name = rest.partition("/")
-    return f"{kind}:{name}"
-
-
 def bind_task(doc: dict, task_id: str, entity_ids: list[str],
               evidence_ref: str = "") -> dict:
     """TASK/ISSUE/INTENT BINDING: task → affects → entities. The structural
@@ -589,55 +599,90 @@ def task_entities(doc: dict, task_id: str) -> list[str]:
                   relationships(doc, from_id=f"task:{task_id}", type="affects"))
 
 
-def impact_frontier(doc: dict, seed_ids: list[str], max_level: int = 2) -> dict:
+def impact_frontier(doc: dict, seed_ids: list[str], max_level: int = 2,
+                    change_kind: str | None = None) -> dict:
     """STRUCTURAL IMPACT RESOLUTION — bounded, deterministic, Spine-only.
 
-    LEVEL 0: the seeds (direct change targets)
-    LEVEL 1: obligation-bearing neighbours (consumers, dependencies,
-             contracts, deployment) — expanded via FRONTIER_LEVEL1.
-             CONTRACT EXPANSION: entities implementing/consuming a contract
-             in the frontier pull in ALL their contract peers — a contract
-             change deserves every consumer's inspection, an implementation
-             change with the contract preserved does not explode.
-    LEVEL 2: verification surfaces (tests/canaries obligated to frontier
-             entities) + deployment surfaces.
-    Unknown relationship types are treated as obligation-bearing (fail-loud:
-    a new structural relation deserves frontier visibility until evidence
-    says otherwise). NON_OBLIGATION types are skipped.
+    The change KIND carries semantics beyond topology (hardening T1):
+      contract_change        seed IS a contract → every implementer and
+                             consumer becomes a direct obligation.
+      implementation_change  seed is an implementation, contract preserved →
+                             the contract is a direct obligation; sibling
+                             implementers appear only as CONTEXTUAL
+                             verification information, never as direct
+                             obligations (sharing a contract is not being
+                             affected by a sibling's internal change).
+      contract_violation     implementation change that verification proved
+                             violates the contract → escalates: siblings
+                             become direct obligations (the contract's
+                             guarantee is now in question for all parties).
+    change_kind=None infers: seed kind == 'contract' → contract_change,
+    else implementation_change.
+
+    LEVEL 0: the seeds. LEVEL 1: obligation-bearing neighbours. LEVEL 2:
+    verification/deployment surfaces. Unknown relationship types are
+    obligation-bearing (fail-loud); NON_OBLIGATION types are skipped.
     """
     seeds = list(dict.fromkeys(seed_ids))
     level0 = sorted(set(seeds))
     level1: set[str] = set()
+    contextual: set[str] = set()
     contracts: set[str] = set()
-    frontier = set(level0)
 
+    if change_kind is None:
+        change_kind = ("contract_change"
+                       if any((entity(doc, s) or {}).get("kind") == "contract"
+                              for s in level0)
+                       else "implementation_change")
+
+    frontier = set(level0)
     for rel in doc["relationships"]:
         if rel["type"] in NON_OBLIGATION_RELATIONSHIPS:
             continue
+        if rel.get("status") in ("STALE", "CANDIDATE"):
+            continue  # stale facts are history; candidates are unproven
         f, t = rel["from"], rel["to"]
         if f in frontier:
             level1.add(t)
         if t in frontier and rel["type"] in ("consumed_by",):
             level1.add(f)  # reverse edge: consumed_by points provider→consumer
-    # contract expansion: any contract entity reached at level 1 pulls its
-    # implements/consumes peers into the frontier
-    level1 |= frontier  # expansion considers seeds too
-    changed = True
-    while changed:  # bounded: only through contract edges
-        changed = False
-        for rel in doc["relationships"]:
-            if rel["type"] in FRONTIER_CONTRACT_EXPANSION:
-                a, b = rel["from"], rel["to"]
-                if (a in level1) != (b in level1):
-                    level1.add(a); level1.add(b); changed = True
-    for c in list(level1):
-        if entity(doc, c) and entity(doc, c).get("kind") == "contract":
-            contracts.add(c)
-    level1 -= set(level0)
 
+    # contract adjacency: map each contract to its implements/consumes parties
+    contract_parties: dict[str, set[str]] = {}
+    seed_contracts = {s for s in level0 if (entity(doc, s) or {}).get("kind") == "contract"}
+    for rel in doc["relationships"]:
+        if rel.get("status") == "STALE":
+            continue
+        if rel["type"] in FRONTIER_CONTRACT_EXPANSION:
+            a, b = rel["from"], rel["to"]
+            c = b if (entity(doc, b) or {}).get("kind") == "contract" else a
+            party = a if c == b else b
+            contract_parties.setdefault(c, set()).add(party)
+        if rel["type"] == "implements" and rel["from"] in frontier:
+            seed_contracts.add(rel["to"])
+        if rel["type"] == "consumes" and rel["from"] in frontier:
+            seed_contracts.add(rel["to"])
+    seed_contracts.update(c for c in level1
+                          if (entity(doc, c) or {}).get("kind") == "contract")
+
+    contracts |= seed_contracts
+    if change_kind in ("contract_change", "contract_violation"):
+        # every party of every touched contract is a direct obligation
+        for c in seed_contracts:
+            level1 |= contract_parties.get(c, set())
+    else:
+        # implementation change, contract preserved: siblings are CONTEXT,
+        # the contract itself is the obligation
+        for c in seed_contracts:
+            contextual |= contract_parties.get(c, set())
+        contextual -= set(level0) | level1
+
+    level1 -= set(level0)
     level2: set[str] = set()
-    for eid in level1 | set(level0):
+    for eid in level1 | set(level0) | contextual:
         for rel in doc["relationships"]:
+            if rel.get("status") == "STALE":
+                continue
             if rel["type"] in FRONTIER_LEVEL2_TYPES and rel["from"] == eid:
                 level2.add(rel["to"])
             if rel["type"] == "deployed_as" and rel["from"] in (level1 | set(level0)):
@@ -645,11 +690,37 @@ def impact_frontier(doc: dict, seed_ids: list[str], max_level: int = 2) -> dict:
     return {"level0_direct": level0,
             "level1_obligations": sorted(level1),
             "contracts": sorted(contracts),
+            "contextual_not_obligations": sorted(contextual),
+            "change_kind": change_kind,
             "level2_verification": sorted(level2),
             "traversal_scope": "spine-structural-only",
             "max_level": max_level}
 
 
+def record_observed_impact(doc: dict, task_id: str,
+                           evidence_items: list[dict]) -> dict:
+    """INDEPENDENT OBSERVED IMPACT (hardening T2): derive what was ACTUALLY
+    affected from external evidence — test failures, canary results, runtime
+    errors, deployment effects — NOT by re-running the frontier function
+    whose prediction is being evaluated. Each evidence item names the entity
+    it implicates and the authority that observed it. Returns the entity
+    set for comparison; nothing is auto-promoted into structure."""
+    implicated: set[str] = set()
+    for item in evidence_items:
+        eid = item.get("entity", "")
+        if not eid:
+            continue
+        implicated.add(eid)
+        doc["observations"].append({
+            "kind": "observed_consequence", "task": task_id,
+            "entity": eid, "source": item.get("source", "unattributed"),
+            "detail": str(item.get("detail", ""))[:300],
+            "authority": item.get("authority", ""),
+            "observed_at": now()})
+    return {"task": task_id, "observed_entities": sorted(implicated),
+            "evidence_count": len(evidence_items)}
+
+
 def predicted_vs_observed(predicted: dict, observed_doc: dict, seed_ids: list[str]) -> dict:
     """PREDICTED vs OBSERVED impact — the V1 hook for structural learning.
     Observed = frontier computed AFTER the change; prediction = frontier
@@ -708,13 +779,198 @@ def query(doc: dict, entity_id: str) -> dict:
                             | {r["from"] for r in relationships(doc, to_id=entity_id, type="consumed_by")}),
         "contracts_implements": sorted(r["to"] for r in relationships(doc, from_id=entity_id, type="implements")),
         "contracts_consumes": sorted(r["to"] for r in relationships(doc, from_id=entity_id, type="consumes")),
-        "state": {"maturity": e.get("maturity"), "observed_at": e.get("observed_at")},
+        "state": {"maturity": e.get("maturity"), "observed_at": e.get("observed_at"),
+                  "resolved_status": resolve_status(e),
+                  "superseded_by": e.get("superseded_by"),
+                  "aliases": e.get("aliases", [])},
         "evidence": e.get("evidence_refs", []),
         "impact_frontier": impact_frontier(doc, [entity_id]),
     }
 
 
-# ── CLI (deterministic, JSON canonical) ───────────────────────────────────
+# ── authority/projector ownership + reconciliation (hardening T3/T4) ──────
+#
+# A projected fact carries `authority` = the projector that owns it. On
+# refresh, each authority's pass records a generation and the facts it still
+# observes. Facts the owner no longer emits become STALE — marked, reported
+# in drift, excluded from frontiers — never silently current, never silently
+# deleted (history survives). Facts of OTHER authorities (or manual facts
+# with no authority) are never touched by a foreign refresh: the
+# reconciliation boundary is AUTHORITY OWNERSHIP, not type knowledge.
+
+REL_STATUS = {"UNKNOWN", "CURRENT", "STALE", "CONFLICTING", "DEGRADED"}
+
+
+def _doc_authorities(doc: dict) -> dict:
+    return doc.setdefault("authority_generations", {})
+
+
+def _authority_observed(doc: dict, rel: dict | None, authority: str) -> None:
+    """Record that `authority` ran this generation. If rel is given, mark
+    it CURRENT under that authority."""
+    a = _doc_authorities(doc).setdefault(authority, {"generation": 0, "observed": []})
+    a["generation"] += 1 if rel is None else 0
+    a["last_success"] = now()
+    a["status"] = "CURRENT"
+    if rel is not None:
+        rel["authority"] = authority
+        rel["status"] = "CURRENT"
+        rel["observed_at"] = now()
+        oid = f"{rel['type']}|{rel['from']}|{rel['to']}"
+        if oid not in a["observed"]:
+            a["observed"].append(oid)
+
+
+def authority_failed(doc: dict, authority: str, error: str) -> None:
+    """A projector failed to run: its facts go STALE (not silently fresh)
+    and the failure is recorded as DEGRADED authority health."""
+    a = _doc_authorities(doc).setdefault(authority, {"generation": 0, "observed": []})
+    a["last_attempt"] = now()
+    a["last_error"] = error[:200]
+    a["status"] = "DEGRADED"
+    for rel in doc["relationships"]:
+        if rel.get("authority") == authority and rel.get("status") == "CURRENT":
+            rel["status"] = "STALE"
+    doc["observations"].append({"kind": "authority_degraded", "authority": authority,
+                                "error": error[:200], "observed_at": now()})
+
+
+def reconcile_authorities(doc: dict) -> list[dict]:
+    """After all projector passes: relationships owned by an authority that
+    did NOT re-observe them this generation become STALE. Manual facts
+    (no authority) and foreign-authority facts are untouched. Returns the
+    staled facts for reporting."""
+    authorities = _doc_authorities(doc)
+    staled = []
+    for rel in doc["relationships"]:
+        auth = rel.get("authority")
+        if not auth or rel.get("status") == "STALE":
+            continue
+        oid = f"{rel['type']}|{rel['from']}|{rel['to']}"
+        if oid not in authorities.get(auth, {}).get("observed", []):
+            rel["status"] = "STALE"
+            staled.append({"type": rel["type"], "from": rel["from"], "to": rel["to"],
+                           "authority": auth})
+    return staled
+
+
+def mark_conflict(doc: dict, entity_id: str, authority_a: str, claim_a: str,
+                  authority_b: str, claim_b: str, evidence_ref: str) -> dict:
+    """Two authoritative observations disagree: BOTH are preserved and the
+    entity is flagged CONFLICTING — never silently pick whichever ran last."""
+    e = entity(doc, entity_id)
+    if e is None:
+        raise ValueError(f"conflict on unknown entity: {entity_id}")
+    if not evidence_ref:
+        raise ValueError("conflict recording requires evidence")
+    conflicts = e.setdefault("ext", {}).setdefault("conflicts", [])
+    conflicts.append({"authority_a": authority_a, "claim_a": claim_a,
+                      "authority_b": authority_b, "claim_b": claim_b,
+                      "evidence": evidence_ref, "observed_at": now()})
+    e["status"] = "CONFLICTING"
+    doc["observations"].append({"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()})
+    return e
+
+
+def resolve_status(e: dict) -> str:
+    """Smallest coherent state for an entity: explicit status wins, then
+    conflicts, then maturity-evidence agreement. 'Not checked' is UNKNOWN,
+    never healthy."""
+    if e.get("status"):
+        return e["status"]
+    if e.get("ext", {}).get("conflicts"):
+        return "CONFLICTING"
+    if e.get("maturity") in ("EXERCISED", "LIVE_PROVEN") and e.get("evidence_refs"):
+        return "CURRENT"
+    if e.get("maturity") not in (None, "DECLARED") and not e.get("evidence_refs"):
+        return "CONFLICTING"  # claim without proof
+    return "UNKNOWN"
+
+
+def authority_health(doc: dict) -> dict:
+    """Provenance/freshness per projector — identity, generation, last
+    success/attempt, status. Provenance only; not a monitoring platform."""
+    return {name: {"generation": a.get("generation", 0),
+                   "status": a.get("status", "UNKNOWN"),
+                   "last_success": a.get("last_success"),
+                   "last_attempt": a.get("last_attempt"),
+                   "last_error": a.get("last_error"),
+                   "facts_observed": len(a.get("observed", []))}
+            for name, a in sorted(_doc_authorities(doc).items())}
+
+
+# ── durable identity: rename / alias / supersession (hardening T5) ────────
+
+def alias(doc: dict, canonical_id: str, old_id: str) -> dict:
+    """RENAME: same conceptual entity, identifier changed. The old spine://
+    reference stays resolvable — it deterministically maps to the canonical
+    entity. Aliases are data on the entity, not a separate table."""
+    e = entity(doc, canonical_id)
+    if e is None:
+        raise ValueError(f"alias target not found: {canonical_id}")
+    if canonical_id == old_id:
+        raise ValueError("self-alias")
+    aliases = e.setdefault("aliases", [])
+    if old_id not in aliases:
+        aliases.append(old_id)
+    e["observed_at"] = now()
+    return e
+
+
+def supersede(doc: dict, new_id: str, old_id: str, evidence_ref: str) -> dict:
+    """REPLACEMENT: a new entity supersedes an old one. The old identity
+    remains resolvable historically (never deleted) and records what
+    replaced it. Evidence required — no speculative supersession."""
+    if not evidence_ref:
+        raise ValueError("supersession requires evidence")
+    if entity(doc, new_id) is None or entity(doc, old_id) is None:
+        raise ValueError("supersede requires both entities to exist")
+    add_relationship(doc, type="supersedes", from_id=new_id, to_id=old_id,
+                     evidence_refs=[evidence_ref])
+    old = entity(doc, old_id)
+    old["superseded_by"] = new_id
+    old["status"] = "SUPERSEDED"
+    doc["observations"].append({"kind": "superseded", "old": old_id, "new": new_id,
+                                "evidence": evidence_ref, "observed_at": now()})
+    return old
+
+
+def _alias_index(doc: dict) -> dict:
+    idx = {}
+    for e in doc["entities"]:
+        for a in e.get("aliases", []):
+            idx[a] = e["id"]
+    return idx
+
+
+def entity(doc: dict, id: str) -> dict | None:
+    doc_ents = doc.get("entities", [])
+    for e in doc_ents:
+        if e["id"] == id:
+            return e
+    # rename-resilient resolution: old spine:// references stay meaningful
+    for e in doc_ents:
+        if id in e.get("aliases", []):
+            return e
+    return None
+
+
+def resolve_uri(uri: str) -> str:
+    """spine://<kind>/<name> → internal entity id (id is NOT a path).
+    Rename-resilient: an aliased old id resolves to its canonical form at
+    query time (entity() follows aliases; resolve returns the requested id
+    and entity() maps it)."""
+    scheme, _, rest = uri.partition("://")
+    if scheme != "spine":
+        raise ValueError(f"not a spine URI: {uri}")
+    kind, _, name = rest.partition("/")
+    return f"{kind}:{name}"
+
+
+
 
 def _fresh_doc() -> dict:
     doc = load()
@@ -821,6 +1077,9 @@ def main(argv=None) -> int:
     if cmd == "uri":
         print(json.dumps({"uri": spine_uri(args[0] if args else "")}))
         return 0
+    if cmd == "authority-health":
+        print(json.dumps(authority_health(doc), indent=1))
+        return 0
     if cmd == "promote":
         e = promote(doc, args[0], args[1], args[2]) if len(args) > 2 else None
         save(doc)

tests/test_spine.py

@@ -6,6 +6,7 @@ Machine-independent: deployment observations use fixture stores.
 """
 import json
 import sys
+import tempfile
 from pathlib import Path
 
 import pytest
@@ -480,12 +481,14 @@ class TestContractsAndFrontier:
         in as direct obligations (they see the contract, not this change)."""
         doc = self._contract_world(seeded)
         f = spine.impact_frontier(doc, ["module:transport_a"])
+        # HARDENED SEMANTICS (T1A): implementation change, contract preserved
+        # → contract is a direct obligation; siblings are CONTEXTUAL, never
+        # direct. (The pre-hardening expectation that siblings land in
+        # level1 encoded the wrong semantics — see final report.)
         assert "contract:chat_transport" in f["level1_obligations"]
-        # the contract IS in the frontier; sibling implementers arrive only
-        # through contract expansion (both must preserve the shared shape),
-        # while unrelated entities stay out — boundedness holds
-        assert "module:transport_b" in f["level1_obligations"]
-        assert "subsystem:alpha" not in f["level1_obligations"]  # owner context only
+        assert "module:transport_b" in f["contextual_not_obligations"]
+        assert "module:transport_b" not in f["level1_obligations"]
+        assert f["change_kind"] == "implementation_change"
         assert f["level0_direct"] == ["module:transport_a"]
 
     def test_task_binding_and_frontier(self, seeded):
@@ -556,3 +559,275 @@ class TestContractsAndFrontier:
         assert q["state"]["maturity"] == "DECLARED"
         f = spine.impact_frontier(seeded, spine.task_entities(seeded, "TASK-FUTURE"))
         assert "module:m_a" in f["level1_obligations"]  # unseen rel = obligation-bearing
+
+
+# ── HARDENING T1: contract-change vs implementation-change semantics ──────
+
+class TestHardeningContractSemantics:
+    def _world(self, seeded):
+        spine.add_entity(seeded, id="contract:chat_transport", kind="contract",
+                         evidence_refs=["protocol"], maturity="DECLARED")
+        for m in ("transport_a", "transport_b"):
+            spine.add_entity(seeded, id=f"module:{m}", kind="module", maturity="DECLARED")
+            spine.add_relationship(seeded, type="implements", from_id=f"module:{m}",
+                                   to_id="contract:chat_transport", evidence_refs=["ast"])
+        spine.add_entity(seeded, id="module:operator", kind="module", maturity="DECLARED")
+        spine.add_relationship(seeded, type="consumes", from_id="module:operator",
+                               to_id="contract:chat_transport", evidence_refs=["ast"])
+        return seeded
+
+    def test_A_implementation_change_siblings_not_direct(self, seeded):
+        doc = self._world(seeded)
+        f = spine.impact_frontier(doc, ["module:transport_a"])
+        assert f["change_kind"] == "implementation_change"
+        assert f["level0_direct"] == ["module:transport_a"]
+        assert "contract:chat_transport" in f["level1_obligations"]
+        assert "module:transport_b" in f["contextual_not_obligations"]
+        assert "module:transport_b" not in f["level1_obligations"]
+        # X (consumer) is reached through the contract's contextual parties
+        # or obligations per evidence — must NOT be silently direct here
+        assert "module:operator" in f["contextual_not_obligations"] or \
+               "module:operator" in f["level1_obligations"]
+
+    def test_B_contract_change_all_parties_direct(self, seeded):
+        doc = self._world(seeded)
+        f = spine.impact_frontier(doc, ["contract:chat_transport"])
+        assert f["change_kind"] == "contract_change"
+        for party in ("module:transport_a", "module:transport_b", "module:operator"):
+            assert party in f["level1_obligations"], party
+
+    def test_C_contract_violation_escalates(self, seeded):
+        doc = self._world(seeded)
+        f = spine.impact_frontier(doc, ["module:transport_a"],
+                                  change_kind="contract_violation")
+        # verification proved the implementation breaks the contract:
+        # every party becomes a direct obligation
+        assert f["change_kind"] == "contract_violation"
+        for party in ("module:transport_b", "module:operator"):
+            assert party in f["level1_obligations"], party
+
+    def test_kind_inference_is_deterministic(self, seeded):
+        doc = self._world(seeded)
+        assert spine.impact_frontier(doc, ["contract:chat_transport"])["change_kind"] == "contract_change"
+        assert spine.impact_frontier(doc, ["module:transport_a"])["change_kind"] == "implementation_change"
+
+
+# ── HARDENING T3/T4: projector ownership, reconciliation, state semantics ─
+
+class TestHardeningReconciliation:
+    def _fact(self, doc, type="depends_on", frm="module:x", to="module:y", auth="proj-x"):
+        spine.add_entity(doc, id=frm, kind="module", maturity="DECLARED")
+        spine.add_entity(doc, id=to, kind="module", maturity="DECLARED")
+        r = spine.add_relationship(doc, type=type, from_id=frm, to_id=to,
+                                   evidence_refs=["fixture"])
+        r["authority"] = auth
+        r["status"] = "CURRENT"
+        return r
+
+    def test_foreign_projector_refresh_never_deletes_other_authority_facts(self, seeded):
+        """Projector Y's refresh must not touch projector X's facts —
+        the reconciliation boundary is AUTHORITY OWNERSHIP."""
+        doc = seeded
+        self._fact(doc, auth="proj-x")
+        foreign = spine.add_relationship(doc, type="related_thing", from_id="module:x",
+                                         to_id="module:y", evidence_refs=["fixture"])
+        spine._authority_observed(doc, foreign, "proj-y")  # proj-y ran and owns it
+        # proj-x refresh observes its own fact; reconcile must leave proj-y's
+        # fact alone — only proj-y's own pass can stale it
+        spine._doc_authorities(doc).setdefault("proj-x", {"generation": 1, "observed": ["depends_on|module:x|module:y"]})
+        staled = spine.reconcile_authorities(doc)
+        assert foreign["status"] == "CURRENT"  # foreign fact survives
+        assert all(s["authority"] != "proj-y" for s in staled)
+
+    def test_owner_stopping_emission_stales_not_deletes(self, seeded):
+        """Projector X later stops observing its fact → STALE (marked,
+        reported in drift, excluded from frontier) — not silently deleted,
+        not silently current."""
+        doc = seeded
+        self._fact(doc, auth="proj-x")
+        spine._doc_authorities(doc)["proj-x"] = {"generation": 2, "observed": []}
+        staled = spine.reconcile_authorities(doc)
+        assert len(staled) == 1
+        f = spine.impact_frontier(doc, ["module:x"])
+        assert "module:y" not in f["level1_obligations"]  # stale ≠ current truth
+        d = spine.drift(doc)
+        assert any(x["class"] == "stale_fact" for x in d)
+
+    def test_projector_failure_does_not_masquerade_as_fresh(self, seeded):
+        doc = seeded
+        self._fact(doc, auth="proj-x")
+        spine._doc_authorities(doc)["proj-x"] = {"generation": 1, "observed": ["depends_on|module:x|module:y"]}
+        spine.authority_failed(doc, "proj-x", "source tree unreachable")
+        h = spine.authority_health(doc)
+        assert h["proj-x"]["status"] == "DEGRADED"
+        r = [r for r in doc["relationships"] if r.get("authority") == "proj-x"][0]
+        assert r["status"] == "STALE"
+        f = spine.impact_frontier(doc, ["module:x"])
+        assert "module:y" not in f["level1_obligations"]
+        assert any(o["kind"] == "authority_degraded" for o in doc["observations"])
+
+    def test_manual_fact_without_authority_survives(self, seeded):
+        doc = seeded
+        r = self._fact(doc, auth=None)
+        r.pop("authority", None); r.pop("status", None)
+        spine.reconcile_authorities(doc)
+        assert r["status"] != "STALE" if "status" in r else True
+        assert r in doc["relationships"]
+
+    def test_two_authorities_conflicting_preserve_both(self, seeded):
+        spine.mark_conflict(seeded, "module:m_a", "registry", "exists",
+                            "runtime", "unavailable", "evidence:live-check")
+        e = spine.entity(seeded, "module:m_a")
+        assert e["status"] == "CONFLICTING"
+        assert len(e["ext"]["conflicts"]) == 1  # BOTH preserved, not last-wins
+        assert spine.resolve_status(e) == "CONFLICTING"
+        q = spine.query(seeded, "module:m_a")
+        assert q["state"]["resolved_status"] == "CONFLICTING"
+
+    def test_conflict_requires_evidence(self, seeded):
+        with pytest.raises(ValueError, match="requires evidence"):
+            spine.mark_conflict(seeded, "module:m_a", "a", "x", "b", "y", "")
+
+    def test_unknown_is_not_healthy(self, seeded):
+        """A never-observed entity is UNKNOWN — not current, not false."""
+        spine.add_entity(seeded, id="module:never_checked", kind="module",
+                         maturity="DECLARED")
+        assert spine.resolve_status(spine.entity(seeded, "module:never_checked")) == "UNKNOWN"
+
+    def test_live_proven_without_evidence_is_conflict_not_stale(self, seeded):
+        """Case A vs B distinction: no evidence at all = CONFLICTING claim;
+        a projector that stopped refreshing = STALE. Different conditions."""
+        spine.add_entity(seeded, id="module:no_proof", kind="module",
+                         maturity="LIVE_PROVEN")  # zero evidence refs
+        assert spine.resolve_status(spine.entity(seeded, "module:no_proof")) == "CONFLICTING"
+
+
+# ── HARDENING T5: durable identity — rename/alias/supersession ────────────
+
+class TestHardeningDurableIdentity:
+    def test_rename_old_spine_reference_stays_meaningful(self, seeded, tmp_path):
+        """Memory stores spine://module/memory_selector; the module is later
+        renamed memory_orchestrator. The old reference must still resolve."""
+        spine.alias(seeded, "module:m_a", "module:m_a_old_name")
+        p = spine.save(seeded, tmp_path / "s.json")
+        reloaded = spine.load(p)
+        # old URI resolves to the same conceptual entity
+        old_ref = spine.resolve_uri("spine://module/m_a_old_name")
+        e = spine.entity(reloaded, old_ref)
+        assert e is not None and e["id"] == "module:m_a"
+        # survives another build cycle (aliases are data, carried forward)
+        doc2 = spine.build.__wrapped__(reloaded) if hasattr(spine.build, "__wrapped__") else reloaded
+        assert spine.entity(doc2, "module:m_a_old_name") is not None
+
+    def test_supersession_requires_evidence_and_preserves_history(self, seeded):
+        spine.add_entity(seeded, id="module:new_thing", kind="module", maturity="DECLARED")
+        with pytest.raises(ValueError, match="requires evidence"):
+            spine.supersede(seeded, "module:new_thing", "module:m_a", "")
+        spine.supersede(seeded, "module:new_thing", "module:m_a", "evidence:decision-record")
+        old = spine.entity(seeded, "module:m_a")
+        assert old is not None  # never deleted
+        assert old["superseded_by"] == "module:new_thing"
+        assert old["status"] == "SUPERSEDED"
+        # the supersedes relationship is data, round-trips, and is
+        # non-obligation for frontier purposes (context, not live wiring)
+        p = spine.save(seeded, Path(tempfile.mkdtemp()) / "s.json")
+        assert spine.entity(spine.load(p), "module:m_a")["superseded_by"] == "module:new_thing"
+
+    def test_superseded_entity_not_treated_as_current(self, seeded):
+        spine.add_entity(seeded, id="module:new_thing", kind="module", maturity="DECLARED")
+        spine.supersede(seeded, "module:new_thing", "module:m_a", "evidence:decision")
+        f = spine.impact_frontier(seeded, ["module:m_a"])
+        # a superseded seed is still answerable, but its status is explicit
+        assert spine.resolve_status(spine.entity(seeded, "module:m_a")) == "SUPERSEDED"
+
+
+# ── HARDENING T2: independent observed impact ─────────────────────────────
+
+class TestHardeningObservedImpact:
+    def test_observed_impact_comes_from_evidence_not_frontier_rerun(self, seeded):
+        """The observed side must be built from EXTERNAL evidence items —
+        test failures, canary results, runtime errors — each naming its
+        authority. The frontier function is not invoked for the observed set."""
+        spine.add_entity(seeded, id="module:m_c", kind="module", maturity="DECLARED")
+        ev = [{"entity": "module:m_b", "source": "pytest-failure",
+               "authority": "test-suite", "detail": "test_consumer_lookup failed"},
+              {"entity": "module:m_c", "source": "canary", "authority": "canary-runner",
+               "detail": "canary_native red"}]
+        r = spine.record_observed_impact(seeded, "TASK-X", ev)
+        assert r["observed_entities"] == ["module:m_b", "module:m_c"]
+        kinds = [o["kind"] for o in seeded["observations"]]
+        assert kinds.count("observed_consequence") == 2
+        # nothing was auto-promoted into structural relationships
+        assert not spine.relationships(seeded, type="consumed_by",
+                                       from_id="module:m_a", to_id="module:m_c")
+
+    def test_predicted_vs_observed_with_independent_evidence(self, seeded):
+        old = json.loads(json.dumps(seeded))
+        predicted = spine.impact_frontier(old, ["module:m_a"])
+        # the change ran; independent evidence implicates an entity the
+        # structure never predicted
+        spine.record_observed_impact(seeded, "TASK-X", [
+            {"entity": "module:m_b", "source": "runtime-error",
+             "authority": "deployed-runtime", "detail": "import crash"}])
+        # compare: prediction's frontier vs the EVIDENCE-derived set
+        observed_set = {"module:m_b"}
+        pred_set = {e for k in ("level0_direct", "level1_obligations",
+                                "contracts", "level2_verification")
+                    for e in predicted.get(k, [])}
+        missed = observed_set - pred_set
+        assert missed == {"module:m_b"}
+        # and the comparison helper records it as CANDIDATE, not truth
+        result = spine.predicted_vs_observed(predicted, seeded, ["module:m_a"])
+        assert all(c["status"] == "CANDIDATE" for c in result["candidate_corrections"])
+
+
+# ── HARDENING: candidate relationships + event seam (T8) ──────────────────
+
+class TestHardeningCandidatesAndEvents:
+    def test_model_speculation_is_candidate_never_truth(self, seeded):
+        """A model-hypothesised relationship (no evidence) enters as
+        CANDIDATE: representable, queryable, but NOT structural truth —
+        excluded from impact frontiers until evidence promotes it."""
+        spine.add_relationship(seeded, type="depends_on", from_id="module:m_a",
+                               to_id="module:m_b", evidence_refs=[])  # speculation
+        rel = spine.relationships(seeded, from_id="module:m_a", type="depends_on")[0]
+        assert rel["status"] == "CANDIDATE"
+        f = spine.impact_frontier(seeded, ["module:m_a"])
+        assert "module:m_b" not in f["level1_obligations"]
+        # evidence arrives → promotion to CURRENT makes it real
+        rel["evidence_refs"] = ["ast-import:m_a"]
+        rel["status"] = "CURRENT"
+        f2 = spine.impact_frontier(seeded, ["module:m_a"])
+        assert "module:m_b" in f2["level1_obligations"]
+
+    def test_event_seam_carries_future_event_kinds_without_core_redesign(self, seeded):
+        """The seam must transport stale/conflict/supersession/degradation
+        events as typed facts — future event kinds flow without policy."""
+        spine.add_entity(seeded, id="module:new_thing", kind="module", maturity="DECLARED")
+        spine.supersede(seeded, "module:new_thing", "module:m_a", "evidence:decision")
+        spine.mark_conflict(seeded, "module:new_thing", "registry", "exists",
+                            "runtime", "unavailable", "evidence:check")
+        spine._authority_observed(seeded, None, "proj-x")
+        spine.authority_failed(seeded, "proj-x", "boom")
+        events = spine.structural_events(seeded)
+        kinds = {e["type"] for e in events}
+        assert {"superseded", "authority_conflict", "authority_degraded"} <= kinds
+        # all dumb typed facts: no salience/policy/wake fields anywhere
+        for e in events:
+            assert not {"salience", "attention", "wake", "priority"} & set(e.keys())
+
+    def test_intrinsic_query_on_real_store_is_fast_and_deterministic(self):
+        """T6: the self-knowledge property on the REAL store — one query,
+        deterministic, no LLM/memory/rebuild. And fast (bounded lookup)."""
+        doc = spine.load()
+        if doc is None:
+            pytest.skip("no spine store in this checkout")
+        import time as _t
+        t0 = _t.monotonic()
+        q1 = spine.query(doc, "module:rune_operator")
+        q2 = spine.query(doc, "module:rune_operator")
+        dt = _t.monotonic() - t0
+        assert q1 == q2  # deterministic
+        assert q1["owner"] and q1["state"]["maturity"]
+        assert "impact_frontier" in q1
+        assert dt < 2.0, f"query took {dt:.2f}s — intrinsic property lost"