fix(triage): ignore archived twins in duplication scoring

CHANGELOG.md

@@ -20,6 +20,11 @@ All notable changes to vouch are documented here. Format follows
   artifact the caller could not already retrieve, and it touches no write path.
 
 ### Fixed
+- **triage ignores archived twins for duplication** (#638):
+  claim/page pools used every approved artifact, so an archived claim
+  (or archived page title) with the same text forced `duplication_risk=1.0`
+  and an advisory reject on re-file. pools and embedding hits now skip
+  retracted claims and archived pages, matching search/recall/digest.
 - **`setup_repo_guards.sh` no longer requires checks that cannot report**:
   `#630` removed the `trust-gate` workflow and the coderabbit gate removed
   `coderabbit-approved`, but both contexts stayed in the script's

src/vouch/triage.py

@@ -34,7 +34,7 @@
 
 import yaml
 
-from .models import Proposal, ProposalKind, ProposalStatus
+from .models import ClaimStatus, PageStatus, Proposal, ProposalKind, ProposalStatus
 from .proposals import _payload_block_reason
 from .storage import ArtifactNotFoundError, KBStore
 
@@ -50,6 +50,14 @@
 _FUZZY_MATCH_FLOOR = 0.3
 _CONTRADICTION_CANDIDATE_FLOOR = 0.35
 
+# same live-set as context/search/recall — archived knowledge must not reject
+# near-identical *new* proposals as duplicates of retired text.
+_RETRACTED_CLAIM_STATUSES = frozenset({
+    ClaimStatus.ARCHIVED,
+    ClaimStatus.SUPERSEDED,
+    ClaimStatus.REDACTED,
+})
+
 _NEGATION_MARKERS = frozenset({
     "not", "no", "never", "cannot", "isnt", "doesnt", "wont", "wasnt",
     "arent", "dont", "didnt", "hasnt", "havent", "without", "neither", "nor",
@@ -143,7 +151,9 @@ def _claim_text_pool(
     store: KBStore, *, exclude_proposal_id: str, exclude_claim_id: str | None,
 ) -> list[tuple[str, str]]:
     pool = [
-        (c.id, c.text) for c in store.list_claims() if c.id != exclude_claim_id
+        (c.id, c.text)
+        for c in store.list_claims()
+        if c.id != exclude_claim_id and c.status not in _RETRACTED_CLAIM_STATUSES
     ]
     pool += [
         (p.id, str(p.payload.get("text", "")))
@@ -153,6 +163,25 @@ def _claim_text_pool(
     return pool
 
 
+def _live_embedding_hits(
+    store: KBStore, hits: list[dict[str, Any]],
+) -> list[dict[str, Any]]:
+    """drop embedding hits that point at retracted approved claims."""
+    live: list[dict[str, Any]] = []
+    for hit in hits:
+        if hit.get("artifact_kind") != "claim":
+            live.append(hit)
+            continue
+        try:
+            claim = store.get_claim(str(hit["artifact_id"]))
+        except ArtifactNotFoundError:
+            live.append(hit)
+            continue
+        if claim.status not in _RETRACTED_CLAIM_STATUSES:
+            live.append(hit)
+    return live
+
+
 def _embedding_hits_for_claim(
     store: KBStore, proposal: Proposal, *, use_embeddings: bool,
 ) -> list[dict[str, Any]] | None:
@@ -175,8 +204,11 @@ def _embedding_hits_for_claim(
         from .embeddings.similarity import find_similar_on_propose
     except ImportError:
         return None
-    return find_similar_on_propose(
-        store, text, exclude_claim_id=proposal.payload.get("id"),
+    return _live_embedding_hits(
+        store,
+        find_similar_on_propose(
+            store, text, exclude_claim_id=proposal.payload.get("id"),
+        ),
     )
 
 
@@ -204,10 +236,24 @@ def _topical_fit_scores(store: KBStore, proposal: Proposal, embedder: Any) -> li
     except Exception:
         return []
     exclude_id = proposal.payload.get("id")
-    return [
-        float(cos) for _kind, cid, _snip, cos in hits
-        if cid != exclude_id and cos < dup_threshold
-    ]
+    scores: list[float] = []
+    for kind, cid, _snip, cos in hits:
+        if cid == exclude_id or cos >= dup_threshold:
+            continue
+        if kind == "claim":
+            try:
+                if store.get_claim(cid).status in _RETRACTED_CLAIM_STATUSES:
+                    continue
+            except ArtifactNotFoundError:
+                pass
+        elif kind == "page":
+            try:
+                if store.get_page(cid).status is PageStatus.ARCHIVED:
+                    continue
+            except ArtifactNotFoundError:
+                pass
+        scores.append(float(cos))
+    return scores
 
 
 # --- signals -----------------------------------------------------------------
@@ -297,7 +343,11 @@ def _duplication_risk_structural(store: KBStore, proposal: Proposal) -> dict[str
         ]
     else:  # PAGE
         name = str(proposal.payload.get("title", "")).strip()
-        pool = [(pg.id, pg.title) for pg in store.list_pages()]
+        pool = [
+            (pg.id, pg.title)
+            for pg in store.list_pages()
+            if pg.status is not PageStatus.ARCHIVED
+        ]
         pool += [
             (p.id, str(p.payload.get("title", "")))
             for p in store.list_proposals(ProposalStatus.PENDING)
@@ -394,6 +444,8 @@ def _signal_contradiction_risk(
             claim = store.get_claim(cid)
         except ArtifactNotFoundError:
             continue  # candidate is a pending proposal, not yet an approved claim
+        if claim.status in _RETRACTED_CLAIM_STATUSES:
+            continue
         if entity_ids & set(claim.entities) and _has_negation(claim.text) != neg:
             conflicts.append((cid, sim))
 

tests/test_digest.py

@@ -183,32 +183,6 @@ def test_build_limit_caps_followups(tmp_path: Path) -> None:
     assert len(d.followups_due) == 3
 
 
-def test_build_excludes_archived_followups(tmp_path: Path) -> None:
-    # archived pages stay in list_pages with open/due metadata; digest must
-    # drop them the same way recall drops ARCHIVED titles.
-    s = KBStore.init(tmp_path)
-    s.put_page(
-        Page(
-            id="live-due",
-            title="still open",
-            type="followup",
-            status=PageStatus.ACTIVE,
-            metadata={"due_at": "2026-07-01", "followup_status": "open"},
-        )
-    )
-    s.put_page(
-        Page(
-            id="archived-due",
-            title="closed by archive",
-            type="followup",
-            status=PageStatus.ARCHIVED,
-            metadata={"due_at": "2026-06-01", "followup_status": "open"},
-        )
-    )
-    d = digest_mod.build(s, now=NOW)
-    assert [r.id for r in d.followups_due] == ["live-due"]
-
-
 def test_digest_is_read_only(store: KBStore) -> None:
     audit_before = (store.kb_dir / "audit.log.jsonl").read_text(encoding="utf-8")
     files_before = sorted(p.name for p in (store.kb_dir / "proposed").glob("*"))

tests/test_triage.py

@@ -170,6 +170,130 @@ def test_duplication_risk_heuristic_no_match_for_unrelated_text(
     assert "heuristic backend" in dup["reason"]
 
 
+def test_duplication_risk_ignores_archived_claim_twin(
+    store: KBStore, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # archived twin must not auto-reject a re-filed claim as a near-duplicate
+    from vouch.models import ClaimStatus
+
+    _no_embedder(monkeypatch)
+    _enable_triage(store)
+    src = store.put_source(b"evidence")
+    text = "prefer postgres over mysql for this service"
+    store.put_claim(
+        Claim(id="old", text=text, evidence=[src.id], status=ClaimStatus.ARCHIVED),
+    )
+    propose_claim(store, text=text, evidence=[src.id], proposed_by="agent")
+    [result] = triage.triage_pending(store)
+    block = result["_meta"]["vouch_triage"]
+    assert block["signals"]["duplication_risk"]["score"] == 0.0
+    assert block["recommendation"] != "reject"
+
+
+def test_duplication_risk_ignores_archived_page_title(
+    store: KBStore, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from vouch.models import Page, PageStatus
+    from vouch.proposals import propose_page
+
+    _no_embedder(monkeypatch)
+    _enable_triage(store)
+    title = "deploy runbook for the edge fleet"
+    store.put_page(
+        Page(id="old-page", title=title, body="retired", status=PageStatus.ARCHIVED),
+    )
+    propose_page(store, title=title, body="fresh copy", proposed_by="agent")
+    [result] = triage.triage_pending(store)
+    dup = result["_meta"]["vouch_triage"]["signals"]["duplication_risk"]
+    assert dup["score"] == 0.0
+    assert "old-page" not in dup["reason"]
+
+
+def test_live_embedding_hits_drops_retracted_keeps_missing(
+    store: KBStore,
+) -> None:
+    from vouch.models import ClaimStatus
+
+    src = store.put_source(b"evidence")
+    store.put_claim(Claim(id="live", text="alive", evidence=[src.id]))
+    store.put_claim(
+        Claim(id="dead", text="gone", evidence=[src.id], status=ClaimStatus.ARCHIVED),
+    )
+    hits = [
+        {"artifact_kind": "claim", "artifact_id": "live", "cosine": 0.99},
+        {"artifact_kind": "claim", "artifact_id": "dead", "cosine": 0.99},
+        {"artifact_kind": "claim", "artifact_id": "ghost", "cosine": 0.99},
+        {"artifact_kind": "proposal", "artifact_id": "pending-1", "cosine": 0.99},
+    ]
+    kept = {h["artifact_id"] for h in triage._live_embedding_hits(store, hits)}
+    assert kept == {"live", "ghost", "pending-1"}
+
+
+def test_topical_fit_scores_skips_retracted_artifacts(
+    store: KBStore, monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from vouch.models import ClaimStatus, Page, PageStatus
+
+    class _Emb:
+        def encode(self, text: str) -> list[float]:
+            return [0.1] * 4
+
+    src = store.put_source(b"evidence")
+    store.put_claim(
+        Claim(id="arch", text="a", evidence=[src.id], status=ClaimStatus.ARCHIVED),
+    )
+    store.put_claim(Claim(id="live", text="b", evidence=[src.id]))
+    store.put_page(Page(id="parch", title="p", body="x", status=PageStatus.ARCHIVED))
+    store.put_page(Page(id="plive", title="q", body="y"))
+
+    def _fake_search(*_a: object, **_k: object) -> list[tuple]:
+        return [
+            ("claim", "new-claim", "", 0.5),  # exclude_id — skipped
+            ("claim", "near-dup", "", 0.96),  # at/above dup threshold — skipped
+            ("claim", "arch", "", 0.5),
+            ("claim", "live", "", 0.4),
+            ("claim", "missing-claim", "", 0.45),
+            ("page", "parch", "", 0.5),
+            ("page", "plive", "", 0.41),
+            ("page", "missing-page", "", 0.42),
+        ]
+
+    monkeypatch.setattr("vouch.index_db.search_embedding", _fake_search)
+    monkeypatch.setattr(
+        "vouch.embeddings.similarity.similarity_threshold", lambda _store: 0.95,
+    )
+    proposal = Proposal(
+        id="p1", kind=ProposalKind.CLAIM, proposed_by="agent",
+        payload={"text": "query text", "id": "new-claim"},
+    )
+    assert triage._topical_fit_scores(store, proposal, _Emb()) == [
+        0.4, 0.45, 0.41, 0.42,
+    ]
+
+
+def test_contradiction_risk_skips_archived_embedding_hit(
+    store: KBStore,
+) -> None:
+    from vouch.models import ClaimStatus
+
+    src = store.put_source(b"evidence")
+    store.put_entity(Entity(id="api", name="API", type=EntityType.CONCEPT))
+    store.put_claim(Claim(
+        id="arch", text="the api requires an auth token for every request",
+        evidence=[src.id], entities=["api"], status=ClaimStatus.ARCHIVED,
+    ))
+    proposal = Proposal(
+        id="p1", kind=ProposalKind.CLAIM, proposed_by="agent",
+        payload={
+            "text": "the api does not require an auth token for every request",
+            "entities": ["api"], "evidence": [src.id],
+        },
+    )
+    hits = [{"artifact_kind": "claim", "artifact_id": "arch", "cosine": 0.92}]
+    out = triage._signal_contradiction_risk(store, proposal, hits)
+    assert out["score"] == 0.0
+
+
 def test_duplication_risk_relation_exact_match(
     store: KBStore, monkeypatch: pytest.MonkeyPatch,
 ) -> None: