fix: reject dangling Relation/Page references at every write path

CHANGELOG.md

@@ -64,6 +64,7 @@ All notable changes to vouch are documented here. Format follows
   or dry-running pending proposals without bypassing the review gate.
 
 ### Fixed
+- `store.put_relation`, `store.put_relation_idempotent`, and `store.put_page` now reject artifacts whose foreign-id references don't resolve in the KB (relation `source` / `target` / `evidence`; page `entities` / `sources`). `proposals.propose_relation` and `proposals.propose_page` surface the same checks at proposal time as `ProposalError`. `bundle.import_check` and `sync.sync_check` run an equivalent cross-artifact pass against the post-merge id set so manifest-consistent bundles can't smuggle relations / pages whose references resolve to nothing — closes the write-time counterpart of the after-the-fact `dangling_relation` finding in `health.lint` (`src/vouch/health.py:135-145`).
 - `bundle.import_check` / `import_apply` and `sync.sync_check` / `sync_apply` now enforce the Source content-addressing invariant: a `sources/<sha>/content` member must hash to `<sha>`, and a `sources/<sha>/meta.yaml` must carry a matching `id`/`hash`. Previously the import side trusted the bundle's directory layout, so a manifest-consistent bundle could land a Source whose content did not match its claimed id — `verify_source` would report `stored_ok=False` only after the import had already succeeded with a clean `bundle.import` audit event. The per-file sha256 gate (#74) only proves bytes match the manifest; this closes the write-time counterpart of the `verify.verify_source` detection.
 - Add `put_relation_idempotent()` to `KBStore` and use it in `supersede()` and `contradict()` so retrying after a partial failure converges to a consistent state instead of raising `ValueError`.
 - Raise `ProposalError("forbidden_self_approval")` in `proposals.approve()` when `approved_by == proposal.proposed_by`, enforcing the review-gate guarantee documented in the README and CONTRIBUTING.

src/vouch/bundle.py

@@ -248,6 +248,157 @@ def _validate_content(path: str, data: bytes, issues: list[str]) -> None:
         issues.append(f"schema validation failed: {path}: {e}")
 
 
+def _artifact_id_from_path(path: str) -> tuple[str, str] | None:
+    """Return (kind, id) for a manifest path, or None for unrelated files.
+
+    Mirrors `sync._artifact_kind` semantics but only emits the entries the
+    referential pass needs. `sources/<sha>/...` collapses to ("source",
+    "<sha>") regardless of whether the path is meta.yaml or content.
+    """
+    parts = path.split("/")
+    if len(parts) < 2:
+        return None
+    top = parts[0]
+    if top == "sources" and len(parts) >= 3:
+        return "source", parts[1]
+    singular = {
+        "claims": "claim",
+        "pages": "page",
+        "entities": "entity",
+        "relations": "relation",
+        "evidence": "evidence",
+    }.get(top)
+    if singular is None:
+        return None
+    return singular, Path(parts[-1]).stem
+
+
+def _existing_ids(kb_dir: Path) -> dict[str, set[str]]:
+    """Snapshot the destination KB's artifact ids per kind.
+
+    Used by the post-merge referential pass: an incoming relation /
+    page reference is satisfied if the target id is either already on
+    disk OR is being delivered by this same bundle.
+    """
+    out: dict[str, set[str]] = {
+        "claim": set(), "page": set(), "entity": set(),
+        "source": set(), "evidence": set(),
+    }
+    for sub, kind, suffix in (
+        ("claims", "claim", ".yaml"),
+        ("entities", "entity", ".yaml"),
+        ("relations", "relation", ".yaml"),
+        ("evidence", "evidence", ".yaml"),
+    ):
+        d = kb_dir / sub
+        if not d.is_dir():
+            continue
+        for p in d.glob(f"*{suffix}"):
+            out[kind].add(p.stem)
+    pages_dir = kb_dir / "pages"
+    if pages_dir.is_dir():
+        for p in pages_dir.glob("*.md"):
+            out["page"].add(p.stem)
+    sources_dir = kb_dir / "sources"
+    if sources_dir.is_dir():
+        for sdir in sources_dir.iterdir():
+            if (sdir / "meta.yaml").exists():
+                out["source"].add(sdir.name)
+    return out
+
+
+def _check_graph_integrity(
+    kb_dir: Path,
+    incoming: dict[str, bytes],
+    issues: list[str],
+) -> None:
+    """Verify every Relation/Page/Claim in `incoming` resolves its refs.
+
+    Closes the bundle / sync equivalent of `put_relation` + `put_page`
+    validation. References are satisfied against the post-merge id set
+    (destination KB ids plus incoming bundle ids), so a self-contained
+    bundle that ships an entity alongside the relation that points at
+    it still imports cleanly. Skips files whose bytes already failed
+    schema validation upstream so a malformed file does not produce a
+    second cryptic error.
+    """
+    ids = _existing_ids(kb_dir)
+    incoming_meta: list[tuple[str, str, bytes]] = []
+    for path, body in incoming.items():
+        kind_id = _artifact_id_from_path(path)
+        if kind_id is None:
+            continue
+        kind, aid = kind_id
+        # Only artifacts that can satisfy a reference go into `ids`.
+        # `relation` is intentionally absent — relations aren't valid
+        # Relation endpoints. The incoming relation itself is still
+        # appended below so its refs get checked.
+        if kind in ids:
+            ids[kind].add(aid)
+        incoming_meta.append((path, kind, body))
+    node_kinds = ("claim", "page", "entity", "source")
+    evidence_kinds = ("source", "evidence")
+    failed_schema = {
+        i.split(": ", 2)[1] for i in issues
+        if i.startswith("schema validation failed: ")
+    }
+    for path, kind, body in incoming_meta:
+        if path in failed_schema:
+            continue
+        try:
+            if kind == "relation":
+                rel = Relation.model_validate(yaml.safe_load(body))
+                if not any(rel.source in ids[k] for k in node_kinds):
+                    issues.append(
+                        f"dangling reference: {path}: relation source "
+                        f"{rel.source!r} not in bundle or destination"
+                    )
+                if not any(rel.target in ids[k] for k in node_kinds):
+                    issues.append(
+                        f"dangling reference: {path}: relation target "
+                        f"{rel.target!r} not in bundle or destination"
+                    )
+                for eid in rel.evidence:
+                    if not any(eid in ids[k] for k in evidence_kinds):
+                        issues.append(
+                            f"dangling reference: {path}: relation "
+                            f"evidence {eid!r} not in bundle or destination"
+                        )
+            elif kind == "page":
+                page = _deserialize_page(body.decode())
+                for cid in page.claims:
+                    if cid not in ids["claim"]:
+                        issues.append(
+                            f"dangling reference: {path}: page claim "
+                            f"{cid!r} not in bundle or destination"
+                        )
+                for eid in page.entities:
+                    if eid not in ids["entity"]:
+                        issues.append(
+                            f"dangling reference: {path}: page entity "
+                            f"{eid!r} not in bundle or destination"
+                        )
+                for sid in page.sources:
+                    if sid not in ids["source"]:
+                        issues.append(
+                            f"dangling reference: {path}: page source "
+                            f"{sid!r} not in bundle or destination"
+                        )
+            elif kind == "claim":
+                claim = Claim.model_validate(yaml.safe_load(body))
+                for ref in claim.evidence:
+                    if not any(ref in ids[k] for k in evidence_kinds):
+                        issues.append(
+                            f"dangling reference: {path}: claim citation "
+                            f"{ref!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
+            # malformed file doesn't mask the more useful upstream error.
+            continue
+
+
 def _check_source_content_address(path: str, body: bytes, issues: list[str]) -> None:
     """Enforce the Source content-addressing invariant on import.
 
@@ -304,6 +455,7 @@ def import_check(kb_dir: Path, bundle_path: Path) -> ImportCheckResult:
     identical: list[str] = []
     issues: list[str] = []
     bundle_id = ""
+    incoming: dict[str, bytes] = {}
 
     with tarfile.open(bundle_path, "r:gz") as tar:
         try:
@@ -343,11 +495,20 @@ def import_check(kb_dir: Path, bundle_path: Path) -> ImportCheckResult:
                 continue
             _validate_content(member.name, body, issues)
             _check_source_content_address(member.name, body, issues)
+            incoming[member.name] = body
         for path in manifest_paths:
             try:
                 tar.getmember(path)
             except KeyError:
                 issues.append(f"manifest lists missing file: {path}")
+    # Graph-integrity pass: the storage layer's put_relation / put_page
+    # validators run on direct writes, but import_apply writes member
+    # bytes straight to disk, so the only place to enforce referential
+    # integrity for a bundle is here. Refs are resolved against the
+    # post-merge id set (destination KB plus incoming bundle), so a
+    # self-contained bundle that ships an entity alongside the relation
+    # pointing at it still imports cleanly.
+    _check_graph_integrity(kb_dir, incoming, issues)
 
     return ImportCheckResult(
         ok=not issues,

src/vouch/proposals.py

@@ -147,6 +147,25 @@ def propose_page(
 ) -> Proposal:
     if not title.strip():
         raise ProposalError("page title is empty")
+    # Mirror the existence check `propose_claim` already runs on evidence
+    # ids: a page that lists a claim / entity / source id but never had it
+    # resolved is exactly the dangling-reference shape `store.put_page`
+    # used to silently accept (issue: graph-integrity write gates).
+    for cid in claim_ids or []:
+        try:
+            store.get_claim(cid)
+        except ArtifactNotFoundError as e:
+            raise ProposalError(f"unknown claim id: {cid}") from e
+    for eid in entity_ids or []:
+        try:
+            store.get_entity(eid)
+        except ArtifactNotFoundError as e:
+            raise ProposalError(f"unknown entity id: {eid}") from e
+    for sid in source_ids or []:
+        try:
+            store.get_source(sid)
+        except ArtifactNotFoundError as e:
+            raise ProposalError(f"unknown source id: {sid}") from e
     payload = {
         "id": slug_hint or _slugify(title),
         "title": title.strip(),
@@ -208,6 +227,31 @@ def propose_relation(
 ) -> Proposal:
     if not src or not target or not relation:
         raise ProposalError("relation needs src, relation, target")
+    # Endpoint + evidence existence checks mirror the `propose_claim`
+    # citation loop. The corresponding write-time gate now lives in
+    # `store.put_relation` / `store.put_relation_idempotent`; surfacing
+    # the same error here means the agent sees a friendly `ProposalError`
+    # at proposal time instead of a downstream `ValueError` at approve.
+    if not _node_exists(store, src):
+        raise ProposalError(
+            f"unknown relation source endpoint: {src} (must be an existing "
+            f"claim, page, entity, or source id)"
+        )
+    if not _node_exists(store, target):
+        raise ProposalError(
+            f"unknown relation target endpoint: {target} (must be an "
+            f"existing claim, page, entity, or source id)"
+        )
+    for eid in evidence or []:
+        try:
+            store.get_source(eid)
+        except ArtifactNotFoundError:
+            try:
+                store.get_evidence(eid)
+            except ArtifactNotFoundError as e:
+                raise ProposalError(
+                    f"unknown source/evidence id: {eid}"
+                ) from e
     rid = f"{src}--{relation}--{target}"
     payload = {
         "id": _slugify(rid),
@@ -482,6 +526,29 @@ def _ensure_no_existing_artifact(
     )
 
 
+def _node_exists(store: KBStore, node_id: str) -> bool:
+    """True if `node_id` resolves to a Claim, Page, Entity, or Source.
+
+    The set of valid Relation endpoint kinds; mirrors
+    `KBStore._node_exists` (storage.py) so propose-time and write-time
+    rejection use the same definition.
+    """
+    if not node_id:
+        return False
+    for getter in (
+        store.get_claim,
+        store.get_page,
+        store.get_entity,
+        store.get_source,
+    ):
+        try:
+            getter(node_id)
+            return True
+        except ArtifactNotFoundError:
+            continue
+    return False
+
+
 def _slugify(text: str) -> str:
     out = []
     last_dash = False

src/vouch/storage.py

@@ -308,6 +308,40 @@ def list_sources(self) -> list[Source]:
                 out.append(Source.model_validate(_yaml_load(meta.read_text())))
         return out
 
+    # --- graph-integrity helpers ------------------------------------------
+
+    # Closes the structural counterpart of the #81 fix: every graph artifact
+    # (Relation source/target/evidence, Page entities/sources) must resolve
+    # to a known artifact in the KB before it lands on disk. The invariant
+    # was already articulated as a `dangling_relation` finding in
+    # `health.lint` (`src/vouch/health.py:135-145`) but no write path
+    # enforced it, so every approve / lifecycle / bundle / sync surface
+    # silently landed graph edges and pages pointing at nothing.
+
+    def _node_exists(self, node_id: str) -> bool:
+        """True if `node_id` resolves to a Claim, Page, Entity, or Source."""
+        if not node_id:
+            return False
+        if self._claim_path(node_id).exists():
+            return True
+        if self._page_path(node_id).exists():
+            return True
+        if self._entity_path(node_id).exists():
+            return True
+        return (self._source_dir(node_id) / "meta.yaml").exists()
+
+    def _evidence_ref_exists(self, ref_id: str) -> bool:
+        """True if `ref_id` resolves to a Source or Evidence record.
+
+        Matches the citation surface that `put_claim` already accepts:
+        either a content-hash Source id or an Evidence id.
+        """
+        if not ref_id:
+            return False
+        if (self._source_dir(ref_id) / "meta.yaml").exists():
+            return True
+        return self._evidence_path(ref_id).exists()
+
     # --- claims ------------------------------------------------------------
 
     def put_claim(self, claim: Claim) -> Claim:
@@ -382,6 +416,12 @@ def put_page(self, page: Page) -> Page:
         for cid in page.claims:
             if not self._claim_path(cid).exists():
                 raise ValueError(f"page {page.id} references unknown claim {cid}")
+        for eid in page.entities:
+            if not self._entity_path(eid).exists():
+                raise ValueError(f"page {page.id} references unknown entity {eid}")
+        for sid in page.sources:
+            if not (self._source_dir(sid) / "meta.yaml").exists():
+                raise ValueError(f"page {page.id} references unknown source {sid}")
         try:
             with self._page_path(page.id).open("x") as f:
                 f.write(_serialize_page(page))
@@ -435,7 +475,27 @@ def list_entities(self) -> list[Entity]:
 
     # --- relations ---------------------------------------------------------
 
+    def _validate_relation_refs(self, rel: Relation) -> None:
+        if not self._node_exists(rel.source):
+            raise ValueError(
+                f"relation {rel.id} references unknown source endpoint "
+                f"{rel.source!r} (must be an existing claim, page, entity, "
+                f"or source id)"
+            )
+        if not self._node_exists(rel.target):
+            raise ValueError(
+                f"relation {rel.id} references unknown target endpoint "
+                f"{rel.target!r} (must be an existing claim, page, entity, "
+                f"or source id)"
+            )
+        for eid in rel.evidence:
+            if not self._evidence_ref_exists(eid):
+                raise ValueError(
+                    f"relation {rel.id} cites unknown source/evidence {eid!r}"
+                )
+
     def put_relation(self, rel: Relation) -> Relation:
+        self._validate_relation_refs(rel)
         try:
             with self._relation_path(rel.id).open("x") as f:
                 f.write(_yaml_dump(rel.model_dump(mode="json")))
@@ -463,6 +523,12 @@ def put_relation_idempotent(self, rel: Relation) -> Relation:
                 text=f"{rel.source} {rel.relation.value} {rel.target}",
             )
             return rel
+        # Validate before the exclusive create. Skipping validation for the
+        # "already on disk" branch above is deliberate — a relation that's
+        # already durable was validated when it landed; re-checking would
+        # turn supersede/contradict retries into spurious failures whenever
+        # the linked claim was subsequently archived or retracted.
+        self._validate_relation_refs(rel)
         try:
             with path.open("x") as f:
                 f.write(_yaml_dump(rel.model_dump(mode="json")))