feat(kb): strip dead claim references at approve time and kb-wide

src/vouch/capabilities.py

@@ -67,6 +67,7 @@
     "kb.archive",
     "kb.confirm",
     "kb.clear_claims",
+    "kb.wipe_dead_refs",
     "kb.cite",
     "kb.source_verify",
     "kb.session_start",

src/vouch/cli.py

@@ -73,6 +73,7 @@
     ProposalError,
     check_approvable,
     expire_pending,
+    missing_claim_refs,
     propose_claim,
     propose_delete,
     propose_entity,
@@ -1596,7 +1597,19 @@ def triage(proposal_ids: tuple[str, ...], as_json: bool, reverse: bool) -> None:
     help="Best-effort: approve every id that can be approved and report the "
     "rest, instead of the default all-or-nothing precheck.",
 )
-def approve(proposal_ids: tuple[str, ...], reason: str | None, keep_going: bool) -> None:
+@click.option(
+    "--drop-missing-claims",
+    is_flag=True,
+    help="Strip references to claims that no longer exist from page "
+    "proposals instead of refusing to approve them (dropped ids are "
+    "recorded in the audit event).",
+)
+def approve(
+    proposal_ids: tuple[str, ...],
+    reason: str | None,
+    keep_going: bool,
+    drop_missing_claims: bool,
+) -> None:
     """Approve one or more proposals — converts each into a durable artifact.
 
     Pass several ids to approve a batch in one call (useful for CI and
@@ -1611,16 +1624,46 @@ def approve(proposal_ids: tuple[str, ...], reason: str | None, keep_going: bool)
       aborts the whole batch and nothing is approved.
     - --keep-going (best-effort): approve each id independently, report the
       failures, and exit non-zero if any failed.
+
+    A page proposal citing claims that no longer exist blocks the batch;
+    interactively you are offered to strip the dead references, and
+    --drop-missing-claims does the same without the prompt.
     """
     store = _load_store()
     approver = _whoami()
 
     if not keep_going:
-        blocked = [
-            (pid, reason_blocked)
-            for pid in proposal_ids
-            if (reason_blocked := check_approvable(store, pid, approved_by=approver))
-        ]
+        blocked: list[tuple[str, str]] = []
+        dead_blocked: list[tuple[str, list[str]]] = []
+        for pid in proposal_ids:
+            why = check_approvable(store, pid, approved_by=approver)
+            if not why:
+                continue
+            # A dead claim reference is a decision, not a defect: offer the
+            # strip-and-approve path instead of a hard abort. Any other block
+            # (typo, already decided, self-approval) still aborts the batch.
+            if "references unknown claim" in why:
+                dead = missing_claim_refs(store, store.get_proposal(pid))
+                if dead:
+                    dead_blocked.append((pid, dead))
+                    continue
+            blocked.append((pid, why))
+        if dead_blocked and not drop_missing_claims:
+            for pid, dead in dead_blocked:
+                click.echo(
+                    f"! {pid}: cites missing claim(s): {', '.join(dead)}", err=True
+                )
+            if sys.stdin.isatty() and click.confirm(
+                f"strip the dead claim reference(s) from {len(dead_blocked)} "
+                "proposal(s) and approve?"
+            ):
+                drop_missing_claims = True
+            else:
+                blocked.extend(
+                    (pid, f"cites missing claim(s): {', '.join(dead)} "
+                     "(use --drop-missing-claims to strip them)")
+                    for pid, dead in dead_blocked
+                )
         if blocked:
             for pid, why in blocked:
                 click.echo(f"✗ {pid}: {why}", err=True)
@@ -1632,7 +1675,10 @@ def approve(proposal_ids: tuple[str, ...], reason: str | None, keep_going: bool)
     failures = 0
     for pid in proposal_ids:
         try:
-            artifact = do_approve(store, pid, approved_by=approver, reason=reason)
+            artifact = do_approve(
+                store, pid, approved_by=approver, reason=reason,
+                drop_missing_claims=drop_missing_claims,
+            )
         except (ArtifactNotFoundError, ValueError, ProposalError, LifecycleError) as e:
             failures += 1
             click.echo(f"✗ {pid}: {e}", err=True)
@@ -2594,6 +2640,53 @@ def claims_clear(auto_only: bool, before: str | None, confirm: bool, dry_run: bo
     click.echo(f"cleared {len(to_clear)} claims")
 
 
+@cli.command(name="wipe-dead-refs")
+@click.option("--dry-run", is_flag=True, default=False,
+              help="Preview what would be stripped without making changes")
+@click.option("--confirm", "skip_confirm", is_flag=True, default=False,
+              help="Skip confirmation prompt")
+def wipe_dead_refs(dry_run: bool, skip_confirm: bool) -> None:
+    """Strip references to claims that no longer exist, KB-wide.
+
+    Scans durable pages and pending page proposals for claim ids that
+    resolve to no claim file (archived claims still resolve) and removes
+    them — frontmatter list and inline [claim: …] markers both. One
+    audited bulk event records what was removed. Run after claims were
+    redacted or bulk-cleared and lint reports orphan_page_ref.
+    """
+    store = _load_store()
+    with _cli_errors():
+        preview = life.wipe_dead_claim_refs(store, actor=_whoami(), dry_run=True)
+
+    if not preview.pages and not preview.proposals:
+        click.echo("no dead claim references found")
+        return
+
+    click.echo(f"found {preview.dropped} dead claim reference(s):")
+    for page_id, dead in preview.pages.items():
+        click.echo(f"  page {page_id}: {', '.join(dead)}")
+    for pid, dead in preview.proposals.items():
+        click.echo(f"  proposal {pid}: {', '.join(dead)}")
+
+    if dry_run:
+        click.echo("(dry-run mode: no changes made)")
+        return
+
+    if not skip_confirm and not click.confirm(
+        f"\nStrip {preview.dropped} dead reference(s)?"
+    ):
+        click.echo("cancelled")
+        return
+
+    with _cli_errors():
+        result = life.wipe_dead_claim_refs(store, actor=_whoami(), dry_run=False)
+    click.echo(
+        f"stripped {result.dropped} dead reference(s) from "
+        f"{len(result.pages)} page(s) and {len(result.proposals)} "
+        "pending proposal(s)"
+    )
+
+
 @cli.command()
 @click.argument("claim_id")
 def confirm(claim_id: str) -> None:

src/vouch/hot_memory.py

@@ -164,6 +164,7 @@ def mark_volunteered(session_id: str, claim_id: str, *, pushed_at: float) -> Non
     "kb.compile": "write path — review gate (files page proposals via wiki-compiler)",
     "kb.summarize_session": "write path — review gate (files session-summary page proposals)",
     "kb.clear_claims": "lifecycle — mutates durable state",
+    "kb.wipe_dead_refs": "lifecycle — mutates durable state",
     "kb.approve": "lifecycle — mutates durable state",
     "kb.reject": "lifecycle — mutates durable state",
     "kb.reject_extracted": "lifecycle — mutates durable state",

src/vouch/jsonl_server.py

@@ -46,6 +46,7 @@
 from .page_filters import filter_pages
 from .proposals import (
     EXPIRE_ACTOR,
+    DeadClaimRefsError,
     ProposalError,
     approve,
     expire_pending,
@@ -508,7 +509,8 @@ def _h_propose_delete(p: dict) -> dict:
 
 def _h_approve(p: dict) -> dict:
     a = approve(_store(), p["proposal_id"], approved_by=_agent(),
-                reason=p.get("reason"))
+                reason=p.get("reason"),
+                drop_missing_claims=bool(p.get("drop_missing_claims", False)))
     return {"kind": type(a).__name__.lower(), "id": a.id}
 
 
@@ -560,6 +562,18 @@ def _h_archive(p: dict) -> dict:
     return {"id": c.id, "status": c.status.value}
 
 
+def _h_wipe_dead_refs(p: dict) -> dict:
+    r = life.wipe_dead_claim_refs(
+        _store(), actor=_agent(), dry_run=bool(p.get("dry_run", False)),
+    )
+    return {
+        "pages": r.pages,
+        "proposals": r.proposals,
+        "dropped": r.dropped,
+        "dry_run": r.dry_run,
+    }
+
+
 def _h_confirm(p: dict) -> dict:
     c = life.confirm(_store(), claim_id=p["claim_id"], actor=_agent())
     return {"id": c.id, "last_confirmed_at": c.last_confirmed_at.isoformat()
@@ -885,6 +899,7 @@ def _h_propose_theme(p: dict) -> dict:
     "kb.archive": _h_archive,
     "kb.confirm": _h_confirm,
     "kb.clear_claims": _h_clear_claims,
+    "kb.wipe_dead_refs": _h_wipe_dead_refs,
     "kb.cite": _h_cite,
     "kb.source_verify": _h_source_verify,
     "kb.session_start": _h_session_start,
@@ -941,6 +956,14 @@ def handle_request(envelope: dict) -> dict:
             "id": req_id, "ok": False,
             "error": {"code": "missing_param", "message": str(e)},
         }
+    except DeadClaimRefsError as e:
+        # Distinct code so interactive clients can offer "strip dead refs
+        # and approve" and retry with drop_missing_claims — a message-match
+        # on invalid_request would be a brittle contract.
+        return {
+            "id": req_id, "ok": False,
+            "error": {"code": "dead_claim_refs", "message": str(e)},
+        }
     except (ValueError, ProposalError, ArtifactNotFoundError) as e:
         return {
             "id": req_id, "ok": False,

src/vouch/lifecycle.py

@@ -11,10 +11,20 @@
 
 from __future__ import annotations
 
+from dataclasses import dataclass, field
 from datetime import UTC, datetime
 
-from . import audit
-from .models import Claim, ClaimStatus, Evidence, Relation, RelationType
+from . import audit, index_db
+from .models import (
+    Claim,
+    ClaimStatus,
+    Evidence,
+    ProposalKind,
+    ProposalStatus,
+    Relation,
+    RelationType,
+)
+from .proposals import strip_claim_markers
 from .storage import ArtifactNotFoundError, KBStore
 
 
@@ -219,6 +229,82 @@ def clear_claims(
     return to_clear
 
 
+@dataclass
+class DeadRefsWipeResult:
+    """Outcome of `wipe_dead_claim_refs` (dry-run or apply)."""
+
+    pages: dict[str, list[str]] = field(default_factory=dict)
+    proposals: dict[str, list[str]] = field(default_factory=dict)
+    dry_run: bool = False
+
+    @property
+    def dropped(self) -> int:
+        return sum(len(v) for v in self.pages.values()) + sum(
+            len(v) for v in self.proposals.values()
+        )
+
+
+def wipe_dead_claim_refs(
+    store: KBStore,
+    *,
+    actor: str,
+    dry_run: bool = False,
+) -> DeadRefsWipeResult:
+    """Strip references to claims that no longer exist, KB-wide.
+
+    Covers durable pages and pending PAGE proposals: the frontmatter claim
+    list and the inline `[claim: …]` body markers both lose the dead ids.
+    Claims themselves are untouched — an archived claim's file still exists,
+    so only ids that resolve to nothing count as dead. One audited bulk
+    event records exactly which ids were removed from where.
+    """
+    result = DeadRefsWipeResult(dry_run=dry_run)
+    for page in store.list_pages():
+        dead = [c for c in page.claims if not store._claim_path(c).exists()]
+        if not dead:
+            continue
+        result.pages[page.id] = dead
+        if dry_run:
+            continue
+        page.claims = [c for c in page.claims if c not in dead]
+        page.body = strip_claim_markers(page.body, dead)
+        page.updated_at = datetime.now(UTC)
+        store.update_page(page)
+        with index_db.open_db(store.kb_dir) as conn:
+            index_db.index_page(
+                conn, id=page.id, title=page.title, body=page.body,
+                type=page.type, tags=page.tags,
+            )
+    for prop in store.list_proposals(ProposalStatus.PENDING):
+        if prop.kind != ProposalKind.PAGE:
+            continue
+        refs = prop.payload.get("claims") or []
+        dead = [c for c in refs if not store._claim_path(c).exists()]
+        if not dead:
+            continue
+        result.proposals[prop.id] = dead
+        if dry_run:
+            continue
+        prop.payload["claims"] = [c for c in refs if c not in dead]
+        body = prop.payload.get("body")
+        if isinstance(body, str):
+            prop.payload["body"] = strip_claim_markers(body, dead)
+        store.update_proposal(prop)
+    if not dry_run and (result.pages or result.proposals):
+        audit.log_event(
+            store.kb_dir,
+            event="page.dead_refs_wipe",
+            actor=actor,
+            object_ids=[*result.pages, *result.proposals],
+            data={
+                "pages": result.pages,
+                "proposals": result.proposals,
+                "dropped": result.dropped,
+            },
+        )
+    return result
+
+
 def cite(store: KBStore, claim_id: str) -> list[Evidence | dict]:
     """Return resolved citations for a claim.