fix(storage): reject dangling Claim graph references on every write path

CHANGELOG.md

@@ -65,6 +65,16 @@ All notable changes to vouch are documented here. Format follows
 - Performance benchmark suite in `benchmarks/` covering search latency, proposal write throughput, bundle export/import/verify round-trips, and index rebuild time at 1k/10k claim sizes. Run with `pytest benchmarks/ --benchmark-only`.
 
 ### Fixed
+- `put_claim` / `update_claim` now reject a Claim whose `entities`,
+  `supersedes`, `superseded_by`, or `contradicts` reference an artifact that
+  is not in the KB, via a new `KBStore._validate_claim_refs`. `bundle.import_check`
+  gains the matching check so a bundle can no longer land a claim with dangling
+  graph refs through `import_apply`'s direct write. Previously only `claim.evidence`
+  was checked: the graph-integrity fix for Relations/Pages (#124) skipped the
+  Claim model's own four reference fields, even though `fsck` already declared
+  `dangling_supersedes` / `dangling_superseded_by` / `dangling_contradicts` as
+  error-severity findings — the invariant was articulated but enforced by no
+  writer. Same model-layer/storage pattern as #81 / #123. Closes #196.
 - `discover_root()` now honours `VOUCH_KB_PATH=/abs/path/.vouch` and returns the parent root, instead of always walking up from cwd. The env var was already documented in `adapters/generic-mcp/README.md` but wasn't wired into the code — closing the doc-vs-code drift removes the `"cwd": "..."` ceremony hosts like Claude Desktop need today to point at a specific KB.
 
 ## [0.1.0] — 2026-05-26

src/vouch/bundle.py

@@ -422,6 +422,28 @@ def _check_graph_integrity(
                             f"dangling reference: {path}: claim citation "
                             f"{ref!r} not in bundle or destination"
                         )
+                # The Claim's own graph refs mirror the page checks above:
+                # entities -> entity ids, supersedes/contradicts/superseded_by
+                # -> claim ids. Without this, import_apply writes the claim
+                # YAML straight to disk (no put_claim guard) carrying links to
+                # artifacts that exist in neither the bundle nor the
+                # destination — the dangling_* errors fsck reports after the
+                # fact (see storage._validate_claim_refs).
+                for eid in claim.entities:
+                    if eid not in ids["entity"]:
+                        issues.append(
+                            f"dangling reference: {path}: claim entity "
+                            f"{eid!r} not in bundle or destination"
+                        )
+                claim_refs = [*claim.supersedes, *claim.contradicts]
+                if claim.superseded_by is not None:
+                    claim_refs.append(claim.superseded_by)
+                for cid in claim_refs:
+                    if cid not in ids["claim"]:
+                        issues.append(
+                            f"dangling reference: {path}: claim graph ref "
+                            f"{cid!r} not in bundle or destination"
+                        )
         except Exception:
             # Schema validation already ran on `body` in `_validate_content`
             # and recorded any structural issue. Swallow here so a single

src/vouch/storage.py

@@ -354,6 +354,36 @@ def _evidence_ref_exists(self, ref_id: str) -> bool:
 
     # --- claims ------------------------------------------------------------
 
+    def _validate_claim_refs(self, claim: Claim) -> None:
+        """Reject dangling graph references on a Claim before it lands.
+
+        The #124 graph-integrity fix closed `Relation.source/target/evidence`
+        and `Page.entities/sources` (see the note above `_node_exists`) but
+        left the Claim's own four reference fields — `entities`,
+        `supersedes`, `superseded_by`, `contradicts` — unchecked on every
+        write path. `fsck._check_lifecycle_chains` already declares three of
+        them as `error`-severity findings (`dangling_supersedes`,
+        `dangling_superseded_by`, `dangling_contradicts`), so the invariant
+        was articulated but enforced by no writer. Enforce it here, the same
+        way `_validate_relation_refs` guards relation endpoints.
+
+        `evidence` is validated separately by `put_claim` (it accepts either
+        a Source id or an Evidence id, a different resolution surface).
+        """
+        for eid in claim.entities:
+            if not self._entity_path(eid).exists():
+                raise ValueError(
+                    f"claim {claim.id} references unknown entity {eid!r}"
+                )
+        claim_refs = [*claim.supersedes, *claim.contradicts]
+        if claim.superseded_by is not None:
+            claim_refs.append(claim.superseded_by)
+        for cid in claim_refs:
+            if not self._claim_path(cid).exists():
+                raise ValueError(
+                    f"claim {claim.id} references unknown claim {cid!r}"
+                )
+
     def put_claim(self, claim: Claim) -> Claim:
         # Evidence entries can be Source IDs or Evidence IDs -- accept either.
         for cid_or_sid in claim.evidence:
@@ -364,6 +394,7 @@ def put_claim(self, claim: Claim) -> Claim:
             raise ValueError(
                 f"claim {claim.id} cites unknown source/evidence {cid_or_sid}"
             )
+        self._validate_claim_refs(claim)
         try:
             with self._claim_path(claim.id).open("x") as f:
                 f.write(_yaml_dump(claim.model_dump(mode="json")))
@@ -398,6 +429,11 @@ def update_claim(self, claim: Claim) -> Claim:
         # The Claim model's field validators only run at construction
         # time; mutation alone bypasses them unless we round-trip.
         Claim.model_validate(claim.model_dump(mode="json"))
+        # Re-check graph references too: an in-place mutation can introduce a
+        # dangling entities/supersedes/superseded_by/contradicts link that the
+        # model validator can't catch (it has no KB access). Mirrors the
+        # put_claim guard so the update path can't reintroduce the gap.
+        self._validate_claim_refs(claim)
         self._claim_path(claim.id).write_text(_yaml_dump(claim.model_dump(mode="json")))
         self._embed_and_store(kind="claim", id=claim.id, text=claim.text)
         # Keep the FTS5 row in sync with the on-disk claim so lifecycle

tests/test_bundle.py

@@ -602,6 +602,28 @@ def _page_md(pid: str, entities: list[str], sources: list[str]) -> bytes:
     return f"---\n{_yaml.safe_dump(meta, sort_keys=False)}---\nbody".encode()
 
 
+def _claim_yaml(
+    cid: str,
+    evidence: list[str],
+    *,
+    entities: list[str] | None = None,
+    supersedes: list[str] | None = None,
+    contradicts: list[str] | None = None,
+    superseded_by: str | None = None,
+) -> bytes:
+    import yaml as _yaml
+    return _yaml.safe_dump({
+        "id": cid, "text": "t", "type": "observation", "status": "working",
+        "confidence": 0.7, "evidence": evidence,
+        "entities": entities or [], "supersedes": supersedes or [],
+        "superseded_by": superseded_by, "contradicts": contradicts or [],
+        "scope": "project", "tags": [],
+        "created_at": "2026-05-27T00:00:00+00:00",
+        "updated_at": "2026-05-27T00:00:00+00:00",
+        "last_confirmed_at": None, "approved_by": None,
+    }, sort_keys=False).encode()
+
+
 def test_import_check_rejects_relation_with_dangling_endpoints(
     store: KBStore, tmp_path: Path
 ) -> None:
@@ -659,6 +681,62 @@ def test_import_check_rejects_page_with_dangling_refs(
     assert not (store.kb_dir / "pages" / "evil-page.md").exists()
 
 
+def test_import_check_rejects_claim_with_dangling_graph_refs(
+    store: KBStore, tmp_path: Path
+) -> None:
+    """A bundle claim whose `entities` / `contradicts` (and siblings) point
+    at artifacts absent from both bundle and destination is rejected — the
+    Claim counterpart of the relation / page graph-integrity checks. Without
+    it, import_apply writes the claim YAML straight to disk carrying links
+    that fsck only flags after the fact (#196)."""
+    src = store.put_source(b"e")  # destination source the claim can cite
+    bundle_path = tmp_path / "evil-claim.tar.gz"
+    _write_multi_member_bundle(bundle_path, {
+        "claims/c-ent.yaml": _claim_yaml(
+            "c-ent", [src.id], entities=["ghost-entity"],
+        ),
+        "claims/c-graph.yaml": _claim_yaml(
+            "c-graph", [src.id], contradicts=["ghost-claim"],
+        ),
+    })
+
+    diff = bundle.import_check(store.kb_dir, bundle_path)
+    assert not diff.ok
+    assert any("dangling reference" in i and "claim entity" in i
+               for i in diff.issues), diff.issues
+    assert any("dangling reference" in i and "claim graph ref" in i
+               for i in diff.issues), diff.issues
+
+    with pytest.raises(RuntimeError, match="dangling reference"):
+        bundle.import_apply(store.kb_dir, bundle_path)
+    assert not (store.kb_dir / "claims" / "c-ent.yaml").exists()
+
+
+def test_import_check_accepts_claim_with_resolvable_graph_refs(
+    store: KBStore, tmp_path: Path
+) -> None:
+    """The honest round-trip guard: a claim whose graph refs all resolve
+    (here, to an entity shipped in the same bundle) imports cleanly."""
+    src = store.put_source(b"e")
+    import yaml as _yaml
+    ent_yaml = _yaml.safe_dump({
+        "id": "ent-x", "name": "X", "type": "project",
+        "aliases": [], "description": None, "page": None,
+        "created_at": "2026-05-27T00:00:00+00:00",
+        "updated_at": "2026-05-27T00:00:00+00:00",
+    }, sort_keys=False).encode()
+    bundle_path = tmp_path / "good-claim.tar.gz"
+    _write_multi_member_bundle(bundle_path, {
+        "entities/ent-x.yaml": ent_yaml,
+        "claims/c-ok.yaml": _claim_yaml("c-ok", [src.id], entities=["ent-x"]),
+    })
+
+    diff = bundle.import_check(store.kb_dir, bundle_path)
+    assert diff.ok, diff.issues
+    bundle.import_apply(store.kb_dir, bundle_path)
+    assert (store.kb_dir / "claims" / "c-ok.yaml").exists()
+
+
 def test_import_check_accepts_self_contained_bundle(
     store: KBStore, tmp_path: Path
 ) -> None:

tests/test_cli.py

@@ -93,15 +93,15 @@ def test_fsck_clean_kb_prints_clean_and_exits_zero(store: KBStore) -> None:
 def test_fsck_reports_dangling_chain_and_exits_nonzero(store: KBStore) -> None:
     """`vouch fsck` exits 1 on error findings and prints affected object ids."""
     from vouch.models import Claim
+    from vouch.storage import _yaml_dump
 
     src = store.put_source(b"e")
-    store.put_claim(
-        Claim(
-            id="c1",
-            text="t",
-            evidence=[src.id],
-            supersedes=["ghost"],
-        )
+    # Write straight to disk: put_claim now rejects dangling graph refs
+    # (_validate_claim_refs), so this reproduces the legacy/poisoned claim
+    # YAML that fsck must still surface after the write path is tightened.
+    bad = Claim(id="c1", text="t", evidence=[src.id], supersedes=["ghost"])
+    (store.kb_dir / "claims" / "c1.yaml").write_text(
+        _yaml_dump(bad.model_dump(mode="json"))
     )
     result = CliRunner().invoke(cli, ["fsck"])
     assert result.exit_code == 1, result.output

tests/test_health.py

@@ -8,7 +8,7 @@
 
 from vouch import health, index_db
 from vouch.models import Claim, ClaimStatus, Proposal, ProposalKind, ProposalStatus
-from vouch.storage import KBStore
+from vouch.storage import KBStore, _yaml_dump
 
 
 @pytest.fixture
@@ -122,6 +122,16 @@ def _index_claim(store: KBStore, claim: Claim) -> None:
         )
 
 
+def _write_claim_direct(store: KBStore, claim: Claim) -> None:
+    """Persist a claim straight to disk, bypassing put_claim's reference
+    guard (`_validate_claim_refs`). Simulates a poisoned / legacy claim YAML
+    that landed before the guard existed — exactly the on-disk state fsck's
+    dangling_* checks must still surface after the write path is tightened."""
+    (store.kb_dir / "claims" / f"{claim.id}.yaml").write_text(
+        _yaml_dump(claim.model_dump(mode="json"))
+    )
+
+
 def test_fsck_clean_kb_passes(store: KBStore) -> None:
     """A KB with one consistently-indexed claim is fsck-clean."""
     src = store.put_source(b"e")
@@ -134,10 +144,14 @@ def test_fsck_clean_kb_passes(store: KBStore) -> None:
 
 
 def test_fsck_flags_dangling_supersedes(store: KBStore) -> None:
-    """`claim.supersedes` pointing at a missing claim is an error."""
+    """`claim.supersedes` pointing at a missing claim is an error.
+
+    Written directly to disk: put_claim now rejects dangling graph refs
+    (`_validate_claim_refs`), so this reproduces the legacy/poisoned on-disk
+    YAML that fsck must still catch after the write path is tightened."""
     src = store.put_source(b"e")
-    store.put_claim(Claim(id="c1", text="t", evidence=[src.id],
-                          supersedes=["ghost"]))
+    _write_claim_direct(store, Claim(id="c1", text="t", evidence=[src.id],
+                                     supersedes=["ghost"]))
     report = health.fsck(store)
     codes = {f.code for f in report.findings}
     assert "dangling_supersedes" in codes
@@ -147,8 +161,8 @@ def test_fsck_flags_dangling_supersedes(store: KBStore) -> None:
 def test_fsck_flags_dangling_superseded_by(store: KBStore) -> None:
     """`claim.superseded_by` pointing at a missing claim is an error."""
     src = store.put_source(b"e")
-    store.put_claim(Claim(id="c1", text="t", evidence=[src.id],
-                          superseded_by="ghost"))
+    _write_claim_direct(store, Claim(id="c1", text="t", evidence=[src.id],
+                                     superseded_by="ghost"))
     report = health.fsck(store)
     codes = {f.code for f in report.findings}
     assert "dangling_superseded_by" in codes
@@ -158,8 +172,8 @@ def test_fsck_flags_dangling_superseded_by(store: KBStore) -> None:
 def test_fsck_flags_dangling_contradicts(store: KBStore) -> None:
     """`claim.contradicts` pointing at a missing claim is an error."""
     src = store.put_source(b"e")
-    store.put_claim(Claim(id="c1", text="t", evidence=[src.id],
-                          contradicts=["ghost"]))
+    _write_claim_direct(store, Claim(id="c1", text="t", evidence=[src.id],
+                                     contradicts=["ghost"]))
     report = health.fsck(store)
     codes = {f.code for f in report.findings}
     assert "dangling_contradicts" in codes
@@ -169,9 +183,11 @@ def test_fsck_flags_dangling_contradicts(store: KBStore) -> None:
 def test_fsck_flags_asymmetric_contradicts(store: KBStore) -> None:
     """A → B contradiction not mirrored by B → A is a warning, not silent."""
     src = store.put_source(b"e")
+    # c2 must exist before c1 cites it — put_claim now enforces resolvable
+    # graph refs, and an asymmetric (not dangling) link needs both ends real.
+    store.put_claim(Claim(id="c2", text="b", evidence=[src.id]))
     store.put_claim(Claim(id="c1", text="a", evidence=[src.id],
                           contradicts=["c2"]))
-    store.put_claim(Claim(id="c2", text="b", evidence=[src.id]))
     report = health.fsck(store)
     codes = {f.code for f in report.findings}
     assert "asymmetric_contradicts" in codes

tests/test_storage.py

@@ -343,6 +343,83 @@ def test_put_page_accepts_resolvable_entity_and_source_refs(
     assert store.get_page("p-ok").sources == [src.id]
 
 
+# --- claim graph references (#196) ---------------------------------------
+#
+# put_claim already rejects unresolvable `evidence`; these cover the Claim's
+# *other* four reference fields — entities / supersedes / superseded_by /
+# contradicts — which #124 left unchecked even though fsck declares dangling
+# supersedes/superseded_by/contradicts as error-severity findings.
+
+
+def test_put_claim_rejects_unknown_entity_ref(store: KBStore) -> None:
+    src = store.put_source(b"e")
+    bad = Claim(id="c-ent", text="t", evidence=[src.id], entities=["ghost"])
+    with pytest.raises(ValueError, match="unknown entity"):
+        store.put_claim(bad)
+    assert not (store.kb_dir / "claims" / "c-ent.yaml").exists()
+
+
+def test_put_claim_rejects_unknown_supersedes_ref(store: KBStore) -> None:
+    src = store.put_source(b"e")
+    bad = Claim(id="c-sup", text="t", evidence=[src.id], supersedes=["ghost"])
+    with pytest.raises(ValueError, match="unknown claim"):
+        store.put_claim(bad)
+    assert not (store.kb_dir / "claims" / "c-sup.yaml").exists()
+
+
+def test_put_claim_rejects_unknown_superseded_by_ref(store: KBStore) -> None:
+    src = store.put_source(b"e")
+    bad = Claim(id="c-sb", text="t", evidence=[src.id], superseded_by="ghost")
+    with pytest.raises(ValueError, match="unknown claim"):
+        store.put_claim(bad)
+    assert not (store.kb_dir / "claims" / "c-sb.yaml").exists()
+
+
+def test_put_claim_rejects_unknown_contradicts_ref(store: KBStore) -> None:
+    src = store.put_source(b"e")
+    bad = Claim(id="c-con", text="t", evidence=[src.id], contradicts=["ghost"])
+    with pytest.raises(ValueError, match="unknown claim"):
+        store.put_claim(bad)
+    assert not (store.kb_dir / "claims" / "c-con.yaml").exists()
+
+
+def test_put_claim_accepts_resolvable_graph_refs(store: KBStore) -> None:
+    src = store.put_source(b"e")
+    store.put_entity(Entity(id="ent1", name="E", type=EntityType.CONCEPT))
+    store.put_claim(Claim(id="base", text="b", evidence=[src.id]))
+    ok = Claim(
+        id="c-ok", text="t", evidence=[src.id],
+        entities=["ent1"], contradicts=["base"],
+    )
+    store.put_claim(ok)
+    assert store.get_claim("c-ok").entities == ["ent1"]
+    assert store.get_claim("c-ok").contradicts == ["base"]
+
+
+def test_update_claim_rejects_in_place_mutation_to_dangling_ref(
+    store: KBStore,
+) -> None:
+    src = store.put_source(b"e")
+    store.put_claim(Claim(id="c1", text="t", evidence=[src.id]))
+    c = store.get_claim("c1")
+    c.contradicts = ["ghost"]  # mutate after load — bypasses model validators
+    with pytest.raises(ValueError, match="unknown claim"):
+        store.update_claim(c)
+    # On-disk claim is untouched.
+    assert store.get_claim("c1").contradicts == []
+
+
+def test_lifecycle_contradict_round_trips_after_guard(store: KBStore) -> None:
+    """Honest lifecycle writes stay green: supersede/contradict load both
+    ends via get_claim, so their links always resolve."""
+    src = store.put_source(b"e")
+    store.put_claim(Claim(id="a", text="a", evidence=[src.id]))
+    store.put_claim(Claim(id="b", text="b", evidence=[src.id]))
+    lifecycle.contradict(store, claim_a="a", claim_b="b", actor="tester")
+    assert store.get_claim("a").contradicts == ["b"]
+    assert store.get_claim("b").contradicts == ["a"]
+
+
 # --- evidence -------------------------------------------------------------