feat(server): kb.explain_ranking — expose why a result ranked where it did (#432)

CHANGELOG.md

@@ -6,6 +6,17 @@ All notable changes to vouch are documented here. Format follows
 
 ## [Unreleased]
 
+### Added
+- `kb.explain_ranking` / `vouch explain-ranking` — read-only introspection over
+  the retrieval pipeline. per candidate it reports the lexical (fts5) rank, the
+  semantic (embedding) rank, the rrf contribution, the rerank delta (when the
+  reranker extras are installed), the salience factor, and the gate outcome
+  (`kept` / `budget-dropped` / `uncited` / `status-filtered`). re-runs the same
+  `_retrieve` primitives `kb.context` uses in an instrumented mode rather than
+  duplicating the scoring math, and is viewer-scoped identically so it can't
+  expose an artifact the caller couldn't already retrieve. recency/frequency
+  factors are reported null until #317 lands.
+
 ## [1.2.2] — 2026-07-07
 
 ### Packaging

src/vouch/capabilities.py

@@ -35,6 +35,7 @@
     "kb.search",
     "kb.neighbors",
     "kb.context",
+    "kb.explain_ranking",
     "kb.synthesize",
     "kb.read_page",
     "kb.read_claim",

src/vouch/cli.py

@@ -46,7 +46,7 @@
 from . import vault_sync as vault_sync_mod
 from . import verify as verify_mod
 from .capabilities import capabilities as build_caps
-from .context import build_context_pack
+from .context import build_context_pack, explain_ranking
 from .lifecycle import LifecycleError
 from .logging_config import configure_logging
 from .models import Proposal, ProposalKind, ProposalStatus
@@ -2353,6 +2353,71 @@ def search(
             click.echo(f"{k}/{i}\t{snip}  ({used})")
 
 
+@cli.command(name="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="Simulate the kb.context budget gate at this cap.")
+@click.option("--require-citations/--no-require-citations", default=False,
+              help="Flag uncited claims as gated.")
+@click.option("--rerank/--no-rerank", default=False,
+              help="Report the rerank delta (needs the reranker extras).")
+@click.option("--session-id", default=None,
+              help="Overlay the entity-salience reflex for this session.")
+@click.option("--project", default=None, help="Viewer project for scope filtering.")
+@click.option("--agent", default=None, help="Viewer agent for scope filtering.")
+@click.option("--format", "fmt", type=click.Choice(["text", "json"]), default="text",
+              show_default=True)
+def explain_ranking_cmd(
+    query: str,
+    limit: int,
+    max_chars: int | None,
+    require_citations: bool,
+    rerank: bool,
+    session_id: str | None,
+    project: str | None,
+    agent: str | None,
+    fmt: str,
+) -> None:
+    """Explain why each retrieval candidate ranked where it did.
+
+    Read-only introspection over the kb.context pipeline: per candidate it
+    reports the lexical/semantic rank, the RRF contribution, the rerank delta,
+    the salience factor, and the gate outcome. Viewer-scoped like kb.context.
+    """
+    store = _load_store()
+    result = explain_ranking(
+        store, query=query, limit=limit, max_chars=max_chars,
+        require_citations=require_citations, rerank=rerank,
+        project=project, agent=agent, session_id=session_id,
+    )
+    if fmt == "json":
+        _emit_json(result)
+        return
+
+    stages = result["stages"]
+    enabled = [name for name, on in stages.items() if on] or ["none"]
+    click.echo(f"query: {result['query']}  backend={result['backend']}  "
+               f"stages={','.join(enabled)}")
+    for note in result.get("notes", []):
+        click.echo(f"  note: {note}")
+    if not result["candidates"]:
+        click.echo("  (no candidates)")
+        return
+
+    def _fmt(v: object) -> str:
+        """Render a component score for the text table (None renders as '-')."""
+        return "-" if v is None else (f"{v:.4f}" if isinstance(v, float) else str(v))
+
+    for c in result["candidates"]:
+        click.echo(
+            f"[{c['gate']}] {c['kind']}/{c['id']}  "
+            f"lex={_fmt(c['lexical_rank'])} sem={_fmt(c['semantic_rank'])} "
+            f"rrf={_fmt(c['rrf_score'])} rerankΔ={_fmt(c['rerank_delta'])} "
+            f"salience={_fmt(c['salience_factor'])} score={_fmt(c['final_score'])}"
+        )
+
+
 @cli.command()
 @click.argument("node_id")
 @click.option("--depth", default=1, show_default=True, type=int)

src/vouch/context.py

@@ -126,6 +126,231 @@ def _retrieve(
     return [(k, i, s, sc, "substring") for k, i, s, sc in filtered]
 
 
+GateOutcome = Literal["kept", "budget-dropped", "uncited", "status-filtered"]
+
+
+def explain_ranking(
+    store: KBStore,
+    *,
+    query: str,
+    limit: int = 10,
+    max_chars: int | None = None,
+    require_citations: bool = False,
+    rerank: bool = False,
+    project: str | None = None,
+    agent: str | None = None,
+    session_id: str | None = None,
+) -> dict[str, Any]:
+    """Explain why each retrieval candidate ranked where it did (read-only).
+
+    Re-runs `_retrieve`'s primitives — semantic search, lexical FTS5 search,
+    RRF fusion, viewer scoping — in an instrumented mode rather than
+    duplicating the scoring math, then reports per candidate: its lexical
+    rank, semantic rank, the RRF contribution, the rerank delta (when #5 is
+    enabled), the recency/frequency factors (when #317 is enabled), the
+    salience factor, and the gate outcome that keeps or drops it
+    (`kept` / `budget-dropped` / `uncited` / `status-filtered`).
+
+    Viewer scoping matches `kb.context`: hits invisible to the viewer are
+    dropped by `filter_hits` before instrumentation, so this can never expose
+    an artifact the caller couldn't already retrieve. Stages reported reflect
+    what's wired in — fusion always; rerank only when `rerank=True` and the
+    reranker extras (#5) are installed; recency/frequency (#317) is not yet
+    implemented and its factors are reported as null. Near-duplicate
+    deduplication (a presentation pass in `build_context_pack`) is not
+    modelled: every scoped candidate is surfaced with its own gate.
+    """
+    viewer = viewer_from(config_path=store.config_path, project=project, agent=agent)
+    backend = _configured_backend(store)
+    fetch_limit = scoped_fetch_limit(limit, viewer)
+
+    def _ranks(hits: list[tuple[str, str, str, float]]) -> dict[tuple[str, str], int]:
+        """Map each hit's (kind, id) to its 1-based position in the list."""
+        return {(k, i): r for r, (k, i, _s, _sc) in enumerate(hits, start=1)}
+
+    sem_rank: dict[tuple[str, str], int] = {}
+    lex_rank: dict[tuple[str, str], int] = {}
+    fusion = False
+    used_backend = backend
+    filtered: list[tuple[str, str, str, float]] = []
+
+    if backend in ("auto", "hybrid"):
+        sem = index_db.search_semantic(store.kb_dir, query, limit=fetch_limit)
+        try:
+            lex = index_db.search(store.kb_dir, query, limit=fetch_limit)
+        except sqlite3.Error:
+            lex = []
+        fused = rrf_fuse(sem, lex, limit=fetch_limit)
+        if fused:
+            sem_rank, lex_rank = _ranks(sem), _ranks(lex)
+            fusion = True
+            used_backend = "hybrid"
+            filtered = filter_hits(store, fused, viewer, limit=limit)
+        else:
+            # both retrievers empty -> substring scan (mirrors _retrieve)
+            used_backend = "substring"
+            filtered = filter_hits(
+                store, store.search_substring(query, limit=fetch_limit), viewer, limit=limit
+            )
+    elif backend == "embedding":
+        sem = index_db.search_semantic(store.kb_dir, query, limit=fetch_limit)
+        sem_rank = _ranks(sem)
+        used_backend = "embedding"
+        filtered = filter_hits(store, sem, viewer, limit=limit) if sem else []
+    elif backend == "fts5":
+        try:
+            lex = index_db.search(store.kb_dir, query, limit=fetch_limit)
+        except sqlite3.Error:
+            lex = []
+        lex_rank = _ranks(lex)
+        used_backend = "fts5"
+        filtered = filter_hits(store, lex, viewer, limit=limit) if lex else []
+    else:
+        used_backend = "substring"
+        filtered = filter_hits(
+            store, store.search_substring(query, limit=fetch_limit), viewer, limit=limit
+        )
+
+    candidates: list[dict[str, Any]] = []
+    for fused_rank, (kind, cid, summary, score) in enumerate(filtered, start=1):
+        key = (kind, cid)
+        candidates.append({
+            "kind": kind,
+            "id": cid,
+            "summary": _enrich_summary(store, kind, cid, summary),
+            "fused_rank": fused_rank,
+            "lexical_rank": lex_rank.get(key),
+            "semantic_rank": sem_rank.get(key),
+            "rrf_score": score if fusion else None,
+            "final_score": score,
+            "rerank_delta": None,
+            "recency_factor": None,
+            "frequency_factor": None,
+            "salience_factor": None,
+            "gate": cast(GateOutcome, "kept"),
+        })
+
+    notes: list[str] = []
+    rerank_applied = _apply_rerank(query, candidates, notes) if rerank and candidates else False
+    if session_id:
+        _apply_salience(store, session_id, candidates)
+    _classify_gates(store, candidates, max_chars=max_chars, require_citations=require_citations)
+
+    result: dict[str, Any] = {
+        "query": query,
+        "backend": used_backend,
+        "limit": limit,
+        "viewer": {"project": viewer.project, "agent": viewer.agent},
+        "stages": {
+            "fusion": fusion,
+            "rerank": rerank_applied,
+            "recency_frequency": False,
+        },
+        "candidates": candidates,
+    }
+    if notes:
+        result["notes"] = notes
+    return result
+
+
+def _apply_rerank(query: str, candidates: list[dict[str, Any]], notes: list[str]) -> bool:
+    """Annotate `candidates` with `rerank_delta`; return whether #5 ran.
+
+    Delta is `fused_rank - reranked_rank`: positive means the reranker moved
+    the candidate up. No-op returning False if the reranker extras aren't
+    installed.
+    """
+    try:
+        from .embeddings.rerank import default_reranker
+        from .embeddings.rerank import rerank as do_rerank
+    except ImportError:
+        notes.append("rerank requested but reranker extras (#5) not installed; stage skipped")
+        return False
+    hits = [(c["kind"], c["id"], c["summary"], c["final_score"]) for c in candidates]
+    reranked = do_rerank(query=query, hits=hits, reranker=default_reranker(), top_k=len(hits))
+    new_rank = {(k, i): r for r, (k, i, _s, _sc) in enumerate(reranked, start=1)}
+    for c in candidates:
+        nr = new_rank.get((c["kind"], c["id"]))
+        if nr is not None:
+            c["rerank_delta"] = c["fused_rank"] - nr
+    return True
+
+
+def _apply_salience(
+    store: KBStore, session_id: str, candidates: list[dict[str, Any]]
+) -> None:
+    """Overlay the entity-salience reflex weight onto matching candidates.
+
+    Salience is a sidebar signal (see `salience` module), not a term in the
+    fused score; this surfaces the overlap so a tuner can see which surfaced
+    artifacts the reflex would also have prefetched. Best-effort.
+    """
+    from . import salience as salience_mod
+    try:
+        cfg = yaml.safe_load(store.config_path.read_text(encoding="utf-8"))
+    except (OSError, yaml.YAMLError):
+        cfg = None
+    _enabled, _window, top_k = salience_mod.reflex_cfg(cfg if isinstance(cfg, dict) else {})
+    records = salience_mod.compute_salience(store, session_id, top_k=top_k)
+    ent_factor: dict[str, float] = {}
+    claim_factor: dict[str, float] = {}
+    for pos, rec in enumerate(records):
+        weight = round(1.0 / (1 + pos), 4)
+        ent_factor[rec["entity_id"]] = weight
+        if rec.get("top_claim_id"):
+            claim_factor[rec["top_claim_id"]] = weight
+    for c in candidates:
+        if c["kind"] == "entity" and c["id"] in ent_factor:
+            c["salience_factor"] = ent_factor[c["id"]]
+        elif c["kind"] == "claim" and c["id"] in claim_factor:
+            c["salience_factor"] = claim_factor[c["id"]]
+
+
+def _classify_gates(
+    store: KBStore,
+    candidates: list[dict[str, Any]],
+    *,
+    max_chars: int | None,
+    require_citations: bool,
+) -> None:
+    """Set each candidate's `gate`, mirroring `build_context_pack`'s pipeline.
+
+    Priority: `status-filtered` (retracted/missing claim, dropped during item
+    building) > `budget-dropped` (evicted by the `max_chars` tail pop) >
+    `uncited` (a claim with no citations, flagged when `require_citations`) >
+    `kept`.
+    """
+    survivors: list[dict[str, Any]] = []
+    citations: dict[str, list[str]] = {}
+    for c in candidates:
+        if c["kind"] == "claim":
+            try:
+                claim = store.get_claim(c["id"])
+            except ArtifactNotFoundError:
+                c["gate"] = "status-filtered"
+                continue
+            if claim.status in _RETRACTED_CLAIM_STATUSES:
+                c["gate"] = "status-filtered"
+                continue
+            citations[c["id"]] = list(claim.evidence)
+        survivors.append(c)
+
+    if max_chars is not None:
+        def _clipped_len(s: str) -> int:
+            """Summary length after build_context_pack's 200-char clip."""
+            return len(s[:200] + "…") if len(s) > 200 else len(s)
+
+        if sum(_clipped_len(c["summary"]) for c in survivors) > max_chars:
+            while survivors and sum(_clipped_len(c["summary"]) for c in survivors) > max_chars:
+                survivors.pop()["gate"] = "budget-dropped"
+
+    for c in survivors:
+        if require_citations and c["kind"] == "claim" and not citations.get(c["id"]):
+            c["gate"] = "uncited"
+        else:
+            c["gate"] = "kept"
+
+
 def _enrich_summary(store: KBStore, kind: str, artifact_id: str, summary: str) -> str:
     """Return a non-empty summary, falling back to the stored artifact text."""
     if summary:

src/vouch/jsonl_server.py

@@ -38,7 +38,7 @@
 from . import trust as trust_mod
 from . import verify as verify_mod
 from .capabilities import capabilities as build_caps
-from .context import build_context_pack
+from .context import build_context_pack, explain_ranking
 from .logging_config import configure_logging
 from .models import ProposalStatus
 from .page_filters import filter_pages
@@ -225,6 +225,21 @@ def _h_context(p: dict) -> dict:
     return salience_mod.attach_salience(result, store, session_id, cfg)
 
 
+def _h_explain_ranking(p: dict) -> dict:
+    """JSONL handler for kb.explain_ranking (read-only ranking introspection)."""
+    return explain_ranking(
+        _store(),
+        query=p["query"],
+        limit=int(p.get("limit", 10)),
+        max_chars=int(p["max_chars"]) if p.get("max_chars") is not None else None,
+        require_citations=bool(p.get("require_citations", False)),
+        rerank=bool(p.get("rerank", False)),
+        project=p.get("project"),
+        agent=p.get("agent"),
+        session_id=p.get("session_id"),
+    )
+
+
 def _h_synthesize(p: dict) -> dict:
     return synthesize(
         _store(),
@@ -734,6 +749,7 @@ def _h_propose_theme(p: dict) -> dict:
     "kb.search": _h_search,
     "kb.neighbors": _h_neighbors,
     "kb.context": _h_context,
+    "kb.explain_ranking": _h_explain_ranking,
     "kb.synthesize": _h_synthesize,
     "kb.read_page": _h_read_page,
     "kb.read_claim": _h_read_claim,