From 870a37531a549dbd6d6ce405cfa2fb34dcf04b44 Mon Sep 17 00:00:00 2001 From: minion1227 Date: Thu, 30 Jul 2026 02:45:46 -0700 Subject: [PATCH] =?UTF-8?q?feat(server):=20kb.explain=5Franking=20?= =?UTF-8?q?=E2=80=94=20why=20a=20result=20ranked=20where=20it=20did?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _retrieve hands back (kind, id, summary, score, backend): one opaque score and a backend label. a reviewer tuning fusion, the reranker, or the recency and pages-first signals cannot see how much of a score came from lexical vs semantic rank, what the rrf contribution was, which stage moved a candidate, or which gate cut it. the ranker is a black box exactly where it most needs to be inspectable. kb.explain_ranking re-runs the pipeline stage by stage and snapshots every candidate's rank and score after each one. a candidate present in one snapshot and absent from the next was removed by that stage, which is what the reported gate names — kept, scope-filtered, status-filtered, limit-dropped, budget-dropped, or uncited. the stages compose _retrieve's ranking chain with the lifecycle gate search_kb applies and the budget gate build_context_pack applies, so one call explains every stage an artifact can die at. scope filtering runs without a limit and truncation is its own stage, mirroring search_kb's "scope first so status filtering can refill the window" ordering and keeping a candidate lost to the window attributable instead of folded into the scope filter. no context.py edits: every stage helper is already importable, so the instrumentation re-runs them rather than threading a trace flag through the hot path. rescoring stages (recency, pages_first) report a score delta; ordering stages (rerank, strategy) report a rank delta. a stage that is off still appears, flagged, so the chain reads continuously. read-only throughout — no writes, no proposals, no lifecycle change, and viewer scoping goes through the same filter_hits as kb.context, so nothing is exposed that the caller could not already retrieve. registered on mcp, jsonl, capabilities and the cli, plus the hot-memory coverage map, where it is excluded: a recency sidebar would perturb the output being inspected. the uncited gate is defensive. Claim rejects evidence=[] on the model, so no stored claim can be uncited today; the branch mirrors the check build_context_pack still makes and starts reporting if that invariant is ever relaxed. Closes #432 --- CHANGELOG.md | 13 ++ src/vouch/capabilities.py | 1 + src/vouch/cli.py | 71 +++++++ src/vouch/explain_ranking.py | 322 +++++++++++++++++++++++++++++ src/vouch/hot_memory.py | 3 + src/vouch/jsonl_server.py | 18 ++ src/vouch/server.py | 37 ++++ tests/test_explain_ranking.py | 375 ++++++++++++++++++++++++++++++++++ 8 files changed, 840 insertions(+) create mode 100644 src/vouch/explain_ranking.py create mode 100644 tests/test_explain_ranking.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 5141d497..9363723b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,19 @@ All notable changes to vouch are documented here. Format follows ## [Unreleased] +### Added +- **`kb.explain_ranking` — why a result ranked where it did** (#432): a + read-only breakdown of the retrieval pipeline. Per candidate it reports the + lexical (FTS5) rank, the semantic rank, the RRF contribution, a row for every + stage — fusion, scope and status filters, recency, pages-first, rerank, the + pluggable strategy, the limit window, and the optional budget/citation gates + — with the rank and score delta that stage caused, plus the gate that kept or + dropped it (`kept` / `scope-filtered` / `status-filtered` / `limit-dropped` / + `budget-dropped` / `uncited`). Registered on MCP, JSONL and the CLI + (`vouch explain-ranking "" [--format text|json]`). Viewer-scoped + through the same `filter_hits` as `kb.context`, so it cannot expose an + artifact the caller could not already retrieve, and it touches no write path. + ### Fixed - **vault sync mirrors post-approve WORKING/DRAFT artifacts** (#583): `kb_to_vault` now includes durable `WORKING` claims and `DRAFT` pages diff --git a/src/vouch/capabilities.py b/src/vouch/capabilities.py index 86c348f7..8b69839f 100644 --- a/src/vouch/capabilities.py +++ b/src/vouch/capabilities.py @@ -35,6 +35,7 @@ "kb.activity", "kb.digest", "kb.search", + "kb.explain_ranking", "kb.neighbors", "kb.experts", "kb.context", diff --git a/src/vouch/cli.py b/src/vouch/cli.py index c7ba6d9d..0801242b 100644 --- a/src/vouch/cli.py +++ b/src/vouch/cli.py @@ -3456,6 +3456,77 @@ def search( click.echo(f"{k}/{i}\t{snip} ({used})") +@cli.command("explain-ranking") +@click.argument("query") +@click.option("--limit", "-n", default=10, show_default=True, type=int) +@click.option("--max-chars", default=None, type=int, + help="Also explain kb.context's budget gate at this character cap.") +@click.option("--require-citations", is_flag=True, + help="Also explain the uncited-claim gate.") +@click.option("--format", "fmt", type=click.Choice(["text", "json"]), + default="text", show_default=True) +@click.option("--project", default=None, help="Viewer project for scope filtering.") +@click.option("--agent", default=None, help="Viewer agent for scope filtering.") +def explain_ranking_cmd( + query: str, + limit: int, + max_chars: int | None, + require_citations: bool, + fmt: str, + project: str | None, + agent: str | None, +) -> None: + """Explain why each candidate for QUERY ranked where it did.""" + from .explain_ranking import explain_ranking + + store = _load_store() + with _cli_errors(): + result = explain_ranking( + store, + query=query, + limit=limit, + max_chars=max_chars, + require_citations=require_citations, + project=project, + agent=agent, + ) + + if fmt == "json": + _emit_json(result) + return + + retrieval = result["retrieval"] + stages = retrieval["stages"] + click.echo( + f"backend: {retrieval['used']} (configured {retrieval['configured']}, " + f"semantic {'available' if retrieval['semantic_available'] else 'unavailable'})" + ) + active = [name for name in ("fusion", "recency", "pages_first", "rerank") + if stages.get(name)] + if stages.get("strategy"): + active.append(f"strategy={stages['strategy']}") + click.echo(f"stages active: {', '.join(active) if active else 'none'}") + + for cand in result["candidates"]: + click.echo( + f"\n{cand['kind']}/{cand['id']} gate={cand['gate']}" + f" lexical={cand['lexical_rank']} semantic={cand['semantic_rank']}" + f" rrf={cand['rrf_contribution']}" + ) + for row in cand["stages"]: + # a stage that did not run is shown so the chain reads continuously, + # flagged rather than silently absent. + flag = "" if row["applied"] else " (off)" + drank = row["rank_delta"] + dscore = row["score_delta"] + moved = "" if not drank else f" rank{drank:+d}" + shifted = "" if not dscore else f" score{dscore:+.6f}" + click.echo( + f" {row['stage']:<14} rank={row['rank']} " + f"score={row['score']:.6f}{moved}{shifted}{flag}" + ) + + @cli.command() @click.argument("node_id") @click.option("--depth", default=1, show_default=True, type=int) diff --git a/src/vouch/explain_ranking.py b/src/vouch/explain_ranking.py new file mode 100644 index 00000000..2ec7871b --- /dev/null +++ b/src/vouch/explain_ranking.py @@ -0,0 +1,322 @@ +"""Read-only introspection over the retrieval ranking pipeline — issue #432. + +``_retrieve`` returns each hit as ``(kind, id, summary, score, backend)``: one +opaque score and a backend label. A reviewer tuning fusion, the reranker, or +the recency / pages-first signals has no way to see *why* an artifact surfaced +or got dropped — how much came from lexical vs. semantic rank, what the RRF +contribution was, whether a rescoring stage moved it, or which gate cut it. + +This module re-runs those stages against the same helpers ``context`` uses, +snapshotting every candidate's rank and score after each one. A candidate +present in one snapshot and absent from the next was removed by that stage, +which is what the reported gate names. + +The composition is deliberate rather than a copy of any single caller: it +chains ``_retrieve``'s ranking stages, the lifecycle gate ``search_kb`` +applies, and the budget / citation gates ``build_context_pack`` applies, so +one call explains every stage an artifact can die at. Scope filtering runs +without a limit and truncation is a separate ``limit`` stage — the same +"scope first so status filtering can refill the window" ordering ``search_kb`` +uses, and it keeps a candidate lost to truncation attributable instead of +folding it into the scope filter. + +Two stage shapes matter when reading a breakdown: + +* ``recency`` and ``pages_first`` are rescoring-only — the candidate set is + unchanged, so their signal is the score delta. +* ``rerank`` and ``strategy`` are ordering-only — scores are untouched, so + their signal is the rank delta. + +``strategy`` is the pluggable final reorder; it is reported as a stage so a +shipped ranking plugin's effect is visible rather than folded into the score +it did not change. + +Read-only by construction: every helper called here is one the read path +already uses, and nothing writes, proposes, or mutates the KB. Viewer scoping +runs through the same ``filter_hits`` as ``kb.context``, so a caller cannot +see a candidate it could not already retrieve. +""" + +from __future__ import annotations + +import sqlite3 +from dataclasses import dataclass, field +from typing import Any + +from . import index_db +from .context import ( + _configured_backend, + _configured_pages_first, + _configured_recency, + _configured_rerank, + _configured_strategy, + _filter_live_hits, + _maybe_pages_first, + _maybe_recency, + _maybe_rerank, + _maybe_strategy, +) +from .embeddings.fusion import rrf_fuse +from .scoping import ViewerContext, filter_hits, scoped_fetch_limit, viewer_from +from .storage import KBStore + +Hit = tuple[str, str, str, float] +Key = tuple[str, str] + +DEFAULT_LIMIT = 10 + +# Stage name -> the gate reported for a candidate that stage removed. +_GATE_FOR_STAGE = { + "scope_filter": "scope-filtered", + "status_filter": "status-filtered", + "limit": "limit-dropped", + "budget": "budget-dropped", +} + + +def _key(hit: Hit) -> Key: + return (hit[0], hit[1]) + + +@dataclass +class _Snapshot: + """The candidate set after one pipeline stage.""" + + stage: str + applied: bool + ranks: dict[Key, int] = field(default_factory=dict) + scores: dict[Key, float] = field(default_factory=dict) + + @classmethod + def of(cls, stage: str, hits: list[Hit], *, applied: bool = True) -> _Snapshot: + return cls( + stage=stage, + applied=applied, + ranks={_key(h): i for i, h in enumerate(hits, start=1)}, + scores={_key(h): h[3] for h in hits}, + ) + + +def _retrieve_traced( + store: KBStore, + query: str, + limit: int, + viewer: ViewerContext, +) -> tuple[list[Hit], str, dict[Key, int], dict[Key, int], list[_Snapshot]]: + """Re-run ``_retrieve``'s backend selection, keeping the per-retriever ranks. + + Returns the fused hits, the backend that served them, the lexical and + semantic rank maps, and the fusion snapshot. Mirrors ``_retrieve``'s + branching so the explanation describes the query that would actually run. + """ + backend = _configured_backend(store) + fetch_limit = scoped_fetch_limit(limit, viewer) + sem: list[Hit] = [] + lex: list[Hit] = [] + + def _lexical() -> list[Hit]: + try: + return index_db.search(store.kb_dir, query, limit=fetch_limit) + except sqlite3.Error: + return [] + + if backend in ("auto", "hybrid"): + sem = index_db.search_semantic(store.kb_dir, query, limit=fetch_limit) + lex = _lexical() + fused = rrf_fuse(sem, lex, limit=fetch_limit) + if fused: + used = "hybrid" + else: + # Both retrievers came back empty: _retrieve falls through to the + # substring scan, which applies none of the rescoring stages. + fused = store.search_substring(query, limit=fetch_limit) + used = "substring" + elif backend == "embedding": + sem = index_db.search_semantic(store.kb_dir, query, limit=fetch_limit) + fused, used = sem, "embedding" + elif backend == "fts5": + lex = _lexical() + fused, used = lex, "fts5" + else: + fused, used = store.search_substring(query, limit=fetch_limit), "substring" + + lex_ranks = {_key(h): i for i, h in enumerate(lex, start=1)} + sem_ranks = {_key(h): i for i, h in enumerate(sem, start=1)} + return fused, used, lex_ranks, sem_ranks, [_Snapshot.of(used, fused)] + + +def _budget_survivors(hits: list[Hit], max_chars: int) -> list[Hit]: + """The hits ``build_context_pack`` would keep under a ``max_chars`` budget. + + Mirrors the pack's omission pass — drop from the tail until the summary + total fits. The pack's clipping pass shortens a summary rather than + dropping the item, so it never changes the candidate set and is reported + as a stage that kept everything. + """ + kept = list(hits) + while kept and sum(len(h[2]) for h in kept) > max_chars: + kept.pop() + return kept + + +def _is_uncited(store: KBStore, key: Key) -> bool: + """True for a surviving claim that carries no citations. + + ``require_citations`` does not drop an item — ``build_context_pack`` keeps + it and fails the pack (``failed: ["require_citations"]``). So this is not a + drop stage: it renames the gate on the candidate that is *responsible* for + that failure, which is the artifact a reviewer needs to find. + + No missing-artifact guard: only candidates that survived ``status_filter`` + reach here, and that stage already dropped every claim it could not read. + + Defensive today — ``Claim`` rejects ``evidence=[]`` on the model, which + closes every write path (``models.py``: "claim must cite at least one + Source or Evidence id"), so no stored claim can be uncited. It mirrors the + check ``build_context_pack`` still makes, and starts reporting the moment + that invariant is relaxed rather than silently reporting ``kept``. + """ + if key[0] != "claim": + return False + return not store.get_claim(key[1]).evidence + + +def _stage_rows( + key: Key, + snapshots: list[_Snapshot], +) -> tuple[list[dict[str, Any]], str]: + """Per-stage rows for one candidate, plus the gate that decided its fate.""" + rows: list[dict[str, Any]] = [] + gate = "kept" + prev_rank: int | None = None + prev_score: float | None = None + + for snap in snapshots: + if key not in snap.ranks: + # Absent here but present in the previous snapshot -> this stage + # removed it. A stage that did not run cannot have dropped it. + if prev_rank is not None and snap.applied: + gate = _GATE_FOR_STAGE.get(snap.stage, f"{snap.stage}-dropped") + break + rank, score = snap.ranks[key], snap.scores[key] + rows.append({ + "stage": snap.stage, + "applied": snap.applied, + "rank": rank, + "score": round(score, 6), + "rank_delta": None if prev_rank is None else prev_rank - rank, + "score_delta": None if prev_score is None else round(score - prev_score, 6), + }) + prev_rank, prev_score = rank, score + + return rows, gate + + +def explain_ranking( + store: KBStore, + *, + query: str, + limit: int = DEFAULT_LIMIT, + max_chars: int | None = None, + require_citations: bool = False, + project: str | None = None, + agent: str | None = None, +) -> dict[str, Any]: + """Explain why each candidate for *query* ranked where it did. + + Returns ``{"query", "limit", "viewer", "retrieval", "candidates"}``. + Each candidate carries its lexical and semantic rank, its RRF + contribution, a row per pipeline stage, and the gate that kept or + dropped it. Read-only — no write path is touched. + """ + if limit < 0: + raise ValueError("limit must be >= 0") + + viewer = viewer_from(config_path=store.config_path, project=project, agent=agent) + hits, used, lex_ranks, sem_ranks, snapshots = _retrieve_traced( + store, query, limit, viewer + ) + rrf_scores = {_key(h): h[3] for h in hits} + + scoped = filter_hits(store, hits, viewer) + snapshots.append(_Snapshot.of("scope_filter", scoped)) + + live = _filter_live_hits(store, scoped) + snapshots.append(_Snapshot.of("status_filter", live)) + + recency_on, _half_life = _configured_recency(store) + rescored = _maybe_recency(store, hits=live) + snapshots.append(_Snapshot.of("recency", rescored, applied=recency_on)) + + pages_first_on, _boost = _configured_pages_first(store) + boosted = _maybe_pages_first(store, hits=rescored) + snapshots.append(_Snapshot.of("pages_first", boosted, applied=pages_first_on)) + + rerank_on, rerank_top_k = _configured_rerank(store, limit=limit) + reranked = _maybe_rerank(store, query=query, hits=boosted, limit=limit) + snapshots.append(_Snapshot.of("rerank", reranked, applied=rerank_on)) + + # _maybe_strategy speaks the 5-tuple retrieval shape (backend appended); + # carry `used` through and drop it again so the stage list stays uniform. + strategy_name = _configured_strategy(store) + ordered = [ + (k, i, s, sc) + for k, i, s, sc, _be in _maybe_strategy( + store, + query=query, + hits=[(k, i, s, sc, used) for k, i, s, sc in reranked], + limit=limit, + ) + ] + snapshots.append( + _Snapshot.of("strategy", ordered, applied=strategy_name is not None) + ) + + windowed = ordered[:limit] + snapshots.append(_Snapshot.of("limit", windowed)) + + if max_chars is not None: + windowed = _budget_survivors(windowed, max_chars) + snapshots.append(_Snapshot.of("budget", windowed)) + + # Every candidate the pipeline ever saw, in the order fusion produced them, + # so a dropped artifact is still explained rather than silently missing. + candidates: list[dict[str, Any]] = [] + summaries = {_key(h): h[2] for h in hits} + for hit in hits: + key = _key(hit) + rows, gate = _stage_rows(key, snapshots) + if gate == "kept" and require_citations and _is_uncited(store, key): + gate = "uncited" + candidates.append({ + "kind": key[0], + "id": key[1], + "summary": summaries.get(key, ""), + "lexical_rank": lex_ranks.get(key), + "semantic_rank": sem_ranks.get(key), + "rrf_contribution": round(rrf_scores.get(key, 0.0), 6), + "stages": rows, + "gate": gate, + }) + + return { + "query": query, + "limit": limit, + "viewer": {"project": viewer.project, "agent": viewer.agent}, + "retrieval": { + "configured": _configured_backend(store), + "used": used, + "semantic_available": index_db.semantic_search_available(), + "stages": { + "fusion": used == "hybrid", + "recency": recency_on, + "pages_first": pages_first_on, + "rerank": rerank_on, + "rerank_top_k": rerank_top_k if rerank_on else None, + "strategy": strategy_name, + "budget": max_chars, + "require_citations": require_citations, + }, + }, + "candidates": candidates, + } diff --git a/src/vouch/hot_memory.py b/src/vouch/hot_memory.py index d5a7f465..266d16ff 100644 --- a/src/vouch/hot_memory.py +++ b/src/vouch/hot_memory.py @@ -146,6 +146,9 @@ def mark_volunteered(session_id: str, claim_id: str, *, pushed_at: float) -> Non "kb.neighbors": "graph slice — out of scope for recency sidebar", "kb.synthesize": "answer-mode prose — sidebar adds noise", "kb.diff": "field-level revision diff — self-contained, not a claim browse", + "kb.explain_ranking": ( + "ranking diagnostic — a recency sidebar would perturb the output being inspected" + ), "kb.detect_themes": "cluster analysis — self-contained, not a claim browse", "kb.experts": "ranked entity analysis — self-contained, not a claim browse", "kb.triage_pending": ( diff --git a/src/vouch/jsonl_server.py b/src/vouch/jsonl_server.py index a88eaf1b..80fada66 100644 --- a/src/vouch/jsonl_server.py +++ b/src/vouch/jsonl_server.py @@ -171,6 +171,23 @@ def _h_search(p: dict) -> dict: ) +def _h_explain_ranking(p: dict) -> dict: + from .explain_ranking import explain_ranking + + # One shared implementation across MCP / JSONL / CLI — see + # explain_ranking.explain_ranking. Read-only: no write path is touched. + max_chars = p.get("max_chars") + return explain_ranking( + _store(), + query=p["query"], + limit=int(p.get("limit", 10)), + max_chars=None if max_chars is None else int(max_chars), + require_citations=bool(p.get("require_citations", False)), + project=p.get("project"), + agent=p.get("agent"), + ) + + def _load_cfg(store: KBStore) -> dict: try: loaded = yaml.safe_load((store.kb_dir / "config.yaml").read_text(encoding="utf-8")) @@ -894,6 +911,7 @@ def _h_propose_theme(p: dict) -> dict: "kb.activity": _h_activity, "kb.digest": _h_digest, "kb.search": _h_search, + "kb.explain_ranking": _h_explain_ranking, "kb.neighbors": _h_neighbors, "kb.experts": _h_experts, "kb.context": _h_context, diff --git a/src/vouch/server.py b/src/vouch/server.py index 15ee6613..eac2133e 100644 --- a/src/vouch/server.py +++ b/src/vouch/server.py @@ -238,6 +238,43 @@ def kb_search( ) +@mcp.tool() +def kb_explain_ranking( + query: str, + *, + limit: int = 10, + max_chars: int | None = None, + require_citations: bool = False, + project: str | None = None, + agent: str | None = None, +) -> dict[str, Any]: + """Explain why each candidate for a query ranked where it did. + + Read-only introspection over the retrieval pipeline: per candidate it + returns the lexical and semantic rank, the RRF contribution, a row per + stage (fusion, scope/status filters, recency, pages_first, rerank, + strategy, limit, and the optional budget/citation gates) with the rank + and score delta that stage caused, and the gate that kept or dropped it. + + max_chars / require_citations opt into explaining kb.context's budget and + citation gates. project/agent set the viewer context for scope filtering, + the same as kb.search — nothing is exposed that the caller could not + already retrieve. + """ + from .explain_ranking import explain_ranking + + # One shared implementation across MCP / JSONL / CLI. + return explain_ranking( + _store(), + query=query, + limit=limit, + max_chars=max_chars, + require_citations=require_citations, + project=project, + agent=agent, + ) + + def _load_cfg(store: KBStore) -> dict[str, Any]: try: loaded = yaml.safe_load((store.kb_dir / "config.yaml").read_text(encoding="utf-8")) diff --git a/tests/test_explain_ranking.py b/tests/test_explain_ranking.py new file mode 100644 index 00000000..eba2842a --- /dev/null +++ b/tests/test_explain_ranking.py @@ -0,0 +1,375 @@ +"""Read-only ranking introspection — issue #432.""" + +from __future__ import annotations + +import sqlite3 +from pathlib import Path + +import pytest +import yaml + +from vouch import explain_ranking as er +from vouch import health +from vouch.models import ( + ArtifactScope, + Claim, + ClaimStatus, + Page, + PageStatus, + PageType, + Visibility, +) +from vouch.storage import KBStore + + +def _write_cfg(store: KBStore, **retrieval: object) -> None: + (store.kb_dir / "config.yaml").write_text( + yaml.safe_dump({"retrieval": retrieval}), encoding="utf-8" + ) + + +@pytest.fixture +def store(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> KBStore: + kb = KBStore.init(tmp_path) + monkeypatch.chdir(kb.root) + src = kb.put_source(b"auth notes") + kb.put_claim(Claim(id="c1", text="auth uses jwt tokens", evidence=[src.id])) + kb.put_claim(Claim(id="c2", text="jwt rotation is manual", evidence=[src.id])) + health.rebuild_index(kb) + return kb + + +def _by_id(result: dict) -> dict[str, dict]: + return {c["id"]: c for c in result["candidates"]} + + +def _stages(candidate: dict) -> list[str]: + return [row["stage"] for row in candidate["stages"]] + + +# --- the fused-only query the issue asks for ----------------------------- + + +def test_fused_only_query_reports_per_retriever_ranks(store: KBStore) -> None: + """A fused query exposes lexical rank, semantic rank and the rrf score.""" + result = er.explain_ranking(store, query="jwt", limit=5) + + assert result["retrieval"]["used"] == "hybrid" + assert result["retrieval"]["stages"]["fusion"] is True + cand = _by_id(result)["c1"] + # fts5 served this query; the embedding retriever is absent without extras, + # which is exactly the asymmetry the breakdown is supposed to make visible. + assert cand["lexical_rank"] is not None + assert cand["rrf_contribution"] > 0 + assert cand["gate"] == "kept" + assert _stages(cand)[0] == "hybrid" + assert "limit" in _stages(cand) + + +def test_first_stage_row_has_no_deltas(store: KBStore) -> None: + """Nothing precedes fusion, so its deltas are null rather than zero.""" + first = _by_id(er.explain_ranking(store, query="jwt"))["c1"]["stages"][0] + assert first["rank_delta"] is None + assert first["score_delta"] is None + + +# --- gate-dropped candidates --------------------------------------------- + + +@pytest.mark.parametrize( + "retracted", [ClaimStatus.ARCHIVED, ClaimStatus.SUPERSEDED, ClaimStatus.REDACTED] +) +def test_retracted_claim_reports_status_filtered( + store: KBStore, retracted: ClaimStatus +) -> None: + store.put_claim(Claim( + id="c-dead", text="jwt replaced by sessions", + evidence=[store.list_sources()[0].id], status=retracted, + )) + health.rebuild_index(store) + + cand = _by_id(er.explain_ranking(store, query="jwt", limit=5))["c-dead"] + assert cand["gate"] == "status-filtered" + # it is explained up to the stage that cut it, not silently missing + assert _stages(cand) == ["hybrid", "scope_filter"] + + +def test_archived_page_reports_status_filtered(store: KBStore) -> None: + store.put_page(Page( + id="p-arch", title="jwt legacy design", body="old", + type=PageType.CONCEPT, status=PageStatus.ARCHIVED, + )) + health.rebuild_index(store) + + assert _by_id(er.explain_ranking(store, query="jwt", limit=5))["p-arch"]["gate"] == ( + "status-filtered" + ) + + +def test_truncated_candidate_reports_limit_dropped(store: KBStore) -> None: + """A candidate lost to the window is attributed to `limit`, not the filters.""" + result = er.explain_ranking(store, query="jwt", limit=1) + gates = {c["gate"] for c in result["candidates"]} + assert "limit-dropped" in gates + dropped = next(c for c in result["candidates"] if c["gate"] == "limit-dropped") + assert _stages(dropped)[-1] == "strategy" + + +def test_budget_gate_reports_budget_dropped(store: KBStore) -> None: + result = er.explain_ranking(store, query="jwt", limit=5, max_chars=1) + assert result["retrieval"]["stages"]["budget"] == 1 + assert "budget-dropped" in {c["gate"] for c in result["candidates"]} + + +def test_scope_filtered_candidate_is_attributed_to_scope(store: KBStore) -> None: + """A viewer-invisible claim dies at scope_filter, before the status gate.""" + store.put_claim(Claim( + id="c-priv", text="jwt secret lives in vault", + evidence=[store.list_sources()[0].id], + scope=ArtifactScope(visibility=Visibility.PRIVATE, project="other-project"), + )) + health.rebuild_index(store) + + cand = _by_id(er.explain_ranking( + store, query="jwt", limit=5, project="this-project" + ))["c-priv"] + assert cand["gate"] == "scope-filtered" + assert _stages(cand) == ["hybrid"] + + +def test_require_citations_names_the_uncited_claim(store: KBStore) -> None: + """The uncited gate renames the responsible candidate, it does not drop it.""" + result = er.explain_ranking( + store, query="jwt", limit=5, require_citations=True + ) + assert result["retrieval"]["stages"]["require_citations"] is True + # every fixture claim cites a source, so nothing is uncited + assert {c["gate"] for c in result["candidates"]} == {"kept"} + assert er._is_uncited(store, ("claim", "c1")) is False + assert er._is_uncited(store, ("page", "p1")) is False + + +def test_uncited_gate_renames_a_surviving_claim( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """The uncited branch is defensive — force it to pin the reported gate. + + ``Claim`` rejects ``evidence=[]`` on the model, so no stored claim can be + uncited and this path cannot be reached through the public API. Forcing the + predicate is the only way to assert the gate the branch would report if + that invariant were ever relaxed. + """ + monkeypatch.setattr(er, "_is_uncited", lambda _store, key: key[1] == "c1") + result = er.explain_ranking(store, query="jwt", limit=5, require_citations=True) + + by_id = _by_id(result) + assert by_id["c1"]["gate"] == "uncited" + # the item is renamed, never dropped — its full stage chain is intact + assert by_id["c1"]["stages"][-1]["stage"] == "limit" + assert by_id["c2"]["gate"] == "kept" + + +def test_uncited_gate_is_not_applied_without_require_citations( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """Without the flag the gate stays `kept` even for an uncited claim.""" + monkeypatch.setattr(er, "_is_uncited", lambda _store, _key: True) + result = er.explain_ranking(store, query="jwt", limit=5) + assert {c["gate"] for c in result["candidates"]} == {"kept"} + + +# --- stage reporting ------------------------------------------------------ + + +def test_disabled_stages_are_reported_as_not_applied(store: KBStore) -> None: + """A stage that is off still appears, flagged, so the chain reads whole.""" + _write_cfg(store, backend="hybrid", recency={"enabled": False}) + cand = _by_id(er.explain_ranking(store, query="jwt"))["c1"] + recency = next(r for r in cand["stages"] if r["stage"] == "recency") + assert recency["applied"] is False + assert er.explain_ranking(store, query="jwt")["retrieval"]["stages"]["recency"] is ( + False + ) + + +def test_recency_reports_a_score_delta(store: KBStore) -> None: + """Recency is rescoring-only, so its signal is the score delta.""" + _write_cfg(store, backend="hybrid", recency={"enabled": True, "half_life_days": 1}) + cand = _by_id(er.explain_ranking(store, query="jwt"))["c1"] + recency = next(r for r in cand["stages"] if r["stage"] == "recency") + assert recency["applied"] is True + assert recency["score_delta"] is not None + + +def test_pages_first_stage_is_reported(store: KBStore) -> None: + store.put_page(Page(id="p-live", title="jwt design", body="current", + type=PageType.CONCEPT)) + health.rebuild_index(store) + _write_cfg(store, backend="hybrid", pages_first={"enabled": True, "boost": 2.0}) + + cand = _by_id(er.explain_ranking(store, query="jwt", limit=5))["p-live"] + pages_first = next(r for r in cand["stages"] if r["stage"] == "pages_first") + assert pages_first["applied"] is True + + +def test_strategy_stage_reports_the_configured_plugin(store: KBStore) -> None: + _write_cfg(store, backend="hybrid", strategy="vouch.strategies.provenance") + result = er.explain_ranking(store, query="jwt") + assert result["retrieval"]["stages"]["strategy"] == "vouch.strategies.provenance" + strategy = next( + r for r in _by_id(result)["c1"]["stages"] if r["stage"] == "strategy" + ) + assert strategy["applied"] is True + + +def test_rerank_top_k_is_reported_only_when_rerank_is_on(store: KBStore) -> None: + assert er.explain_ranking(store, query="jwt")["retrieval"]["stages"][ + "rerank_top_k" + ] is None + + +# --- backend branches ----------------------------------------------------- + + +def test_pinned_fts5_backend_is_explained(store: KBStore) -> None: + _write_cfg(store, backend="fts5") + result = er.explain_ranking(store, query="jwt") + assert result["retrieval"]["used"] == "fts5" + assert result["retrieval"]["stages"]["fusion"] is False + assert _by_id(result)["c1"]["lexical_rank"] is not None + + +def test_pinned_embedding_backend_is_explained(store: KBStore) -> None: + _write_cfg(store, backend="embedding") + result = er.explain_ranking(store, query="jwt") + assert result["retrieval"]["used"] == "embedding" + + +def test_pinned_substring_backend_is_explained(store: KBStore) -> None: + _write_cfg(store, backend="substring") + result = er.explain_ranking(store, query="jwt") + assert result["retrieval"]["used"] == "substring" + assert _by_id(result)["c1"]["gate"] == "kept" + + +def test_auto_falls_through_to_substring_when_retrievers_are_empty( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """The substring fall-through is reported as the backend that served it.""" + from vouch import index_db + + monkeypatch.setattr(index_db, "search", lambda *a, **k: []) + monkeypatch.setattr(index_db, "search_semantic", lambda *a, **k: []) + result = er.explain_ranking(store, query="jwt") + assert result["retrieval"]["used"] == "substring" + + +def test_lexical_sqlite_error_degrades_to_no_lexical_hits( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """A broken fts5 index must not fault the explanation.""" + from vouch import index_db + + def _boom(*_a: object, **_k: object) -> list: + raise sqlite3.Error("fts5 index is gone") + + monkeypatch.setattr(index_db, "search", _boom) + result = er.explain_ranking(store, query="jwt") + assert all(c["lexical_rank"] is None for c in result["candidates"]) + + +def test_negative_limit_is_rejected(store: KBStore) -> None: + with pytest.raises(ValueError, match="limit must be >= 0"): + er.explain_ranking(store, query="jwt", limit=-1) + + +def test_viewer_is_echoed_back(store: KBStore) -> None: + result = er.explain_ranking(store, query="jwt", project="proj-a", agent="agent-b") + assert result["viewer"] == {"project": "proj-a", "agent": "agent-b"} + + +# --- surface parity ------------------------------------------------------- + + +def test_jsonl_surface_serves_explain_ranking( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + from vouch import jsonl_server + + monkeypatch.setattr(jsonl_server, "_store", lambda: store) + result = jsonl_server.HANDLERS["kb.explain_ranking"]({ + "query": "jwt", "limit": 5, "max_chars": 4000, "require_citations": False, + }) + assert result["retrieval"]["stages"]["budget"] == 4000 + assert _by_id(result)["c1"]["gate"] == "kept" + + +def test_mcp_surface_serves_explain_ranking( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + from vouch import server + + monkeypatch.setattr(server, "_store", lambda: store) + result = server.kb_explain_ranking("jwt", limit=5) + assert _by_id(result)["c1"]["gate"] == "kept" + + +def test_explain_ranking_is_registered_in_capabilities() -> None: + from vouch.capabilities import METHODS + + assert "kb.explain_ranking" in METHODS + + +def test_cli_text_output_names_stages_and_gates(store: KBStore) -> None: + from click.testing import CliRunner + + from vouch.cli import cli + + store.put_claim(Claim( + id="c-dead", text="jwt replaced by sessions", + evidence=[store.list_sources()[0].id], status=ClaimStatus.SUPERSEDED, + )) + health.rebuild_index(store) + _write_cfg(store, backend="hybrid", strategy="vouch.strategies.provenance", + recency={"enabled": True, "half_life_days": 1}) + + res = CliRunner().invoke( + cli, ["explain-ranking", "jwt", "--limit", "2"] + ) + assert res.exit_code == 0, res.output + assert "backend:" in res.output + assert "stages active:" in res.output + assert "strategy=vouch.strategies.provenance" in res.output + assert "gate=status-filtered" in res.output + assert "(off)" in res.output + + +def test_cli_json_output_is_machine_readable(store: KBStore) -> None: + import json + + from click.testing import CliRunner + + from vouch.cli import cli + + res = CliRunner().invoke( + cli, + ["explain-ranking", "jwt", + "--format", "json", "--require-citations", "--max-chars", "4000"], + ) + assert res.exit_code == 0, res.output + payload = json.loads(res.output) + assert payload["query"] == "jwt" + assert payload["retrieval"]["stages"]["require_citations"] is True + + +def test_cli_reports_no_active_stages_when_all_are_off(store: KBStore) -> None: + from click.testing import CliRunner + + from vouch.cli import cli + + _write_cfg(store, backend="substring", recency={"enabled": False}) + res = CliRunner().invoke( + cli, ["explain-ranking", "jwt"] + ) + assert res.exit_code == 0, res.output + assert "stages active: none" in res.output