feat(retrieval): one search path, recency decay, honest degradation

src/vouch/context.py

@@ -14,6 +14,7 @@
 from __future__ import annotations
 
 import sqlite3
+from datetime import UTC, datetime
 from typing import Any, Literal, cast
 
 import yaml
@@ -107,6 +108,87 @@ def _configured_rerank(store: KBStore, *, limit: int) -> tuple[bool, int]:
     return enabled, top_k
 
 
+def _configured_recency(store: KBStore) -> tuple[bool, float]:
+    """Resolve the optional recency-decay stage from config.yaml.
+
+    Defaults to disabled so existing KBs keep byte-identical ordering unless
+    they opt in with ``retrieval.recency.enabled: true`` (new KBs get it from
+    the starter config). ``half_life_days`` is the age at which an artifact's
+    score contribution halves; <= 0 falls back to the 90-day default.
+    """
+    try:
+        loaded = yaml.safe_load(store.config_path.read_text(encoding="utf-8"))
+    except (OSError, yaml.YAMLError):
+        return False, 90.0
+    if not isinstance(loaded, dict):
+        return False, 90.0
+    retrieval = loaded.get("retrieval")
+    if not isinstance(retrieval, dict):
+        return False, 90.0
+    recency = retrieval.get("recency")
+    if not isinstance(recency, dict):
+        return False, 90.0
+
+    enabled = recency.get("enabled", False)
+    enabled = enabled if isinstance(enabled, bool) else False
+
+    half_life = recency.get("half_life_days", 90.0)
+    half_life = (
+        float(half_life)
+        if isinstance(half_life, (int, float)) and not isinstance(half_life, bool)
+        and half_life > 0
+        else 90.0
+    )
+    return enabled, half_life
+
+
+def _artifact_timestamp(store: KBStore, kind: str, artifact_id: str) -> datetime | None:
+    try:
+        if kind == "claim":
+            claim = store.get_claim(artifact_id)
+            return claim.updated_at or claim.created_at
+        if kind == "page":
+            page = store.get_page(artifact_id)
+            return page.updated_at or page.created_at
+        if kind == "entity":
+            entity = store.get_entity(artifact_id)
+            return entity.updated_at or entity.created_at
+    except (ArtifactNotFoundError, OSError):
+        return None
+    return None
+
+
+def _maybe_recency(
+    store: KBStore,
+    *,
+    hits: list[tuple[str, str, str, float]],
+) -> list[tuple[str, str, str, float]]:
+    """Blend a recency half-life decay into hit scores, newest-favouring.
+
+    Rescoring-only: ``score * (0.5 + 0.5 * decay)`` keeps every hit in the
+    set (an old artifact loses at most half its score, it never vanishes),
+    and artifacts with no readable timestamp are left at full weight.
+    """
+    enabled, half_life_days = _configured_recency(store)
+    if not enabled or not hits:
+        return hits
+    now = datetime.now(UTC)
+    rescored: list[tuple[str, str, str, float]] = []
+    for kind, artifact_id, summary, score in hits:
+        ts = _artifact_timestamp(store, kind, artifact_id)
+        if ts is None:
+            rescored.append((kind, artifact_id, summary, score))
+            continue
+        # Whole days only: sub-day age is noise at a 90-day half-life, and
+        # quantizing keeps repeat queries byte-identical within a day
+        # (fresh artifacts decay 1.0, so same-day scores never drift).
+        age_days = float(int(max((now - ts).total_seconds() / 86400.0, 0.0)))
+        decay = 0.5 ** (age_days / half_life_days)
+        rescored.append((kind, artifact_id, summary, score * (0.5 + 0.5 * decay)))
+    rescored.sort(key=lambda h: h[3], reverse=True)
+    return rescored
+
+
 def _default_reranker_cached() -> Any:
     global _RERANKER_CACHE
     if _RERANKER_CACHE is None:
@@ -186,6 +268,7 @@ def _retrieve(
         fused = rrf_fuse(sem, lex, limit=fetch_limit)
         if fused:
             filtered = filter_hits(store, fused, viewer, limit=limit)
+            filtered = _maybe_recency(store, hits=filtered)
             filtered = _maybe_rerank(store, query=query, hits=filtered, limit=limit)
             return [(k, i, s, sc, "hybrid") for k, i, s, sc in filtered]
         # both retrievers empty -> fall through to the substring scan below.
@@ -212,6 +295,98 @@ def _retrieve(
     return [(k, i, s, sc, "substring") for k, i, s, sc in filtered]
 
 
+def search_kb(
+    store: KBStore,
+    *,
+    query: str,
+    limit: int = 10,
+    backend: str | None = None,
+    min_score: float = 0.0,
+    project: str | None = None,
+    agent: str | None = None,
+) -> dict[str, Any]:
+    """The one `kb.search` implementation every surface delegates to.
+
+    MCP, JSONL, and the CLI used to carry three copies of the backend
+    waterfall and drifted (fusion landed in one, not the others). Keep the
+    logic here only.
+
+    ``backend=None`` defers to ``retrieval.backend`` in config.yaml; "auto"
+    then fuses embedding + FTS5 via RRF and falls back to a substring scan
+    only when both are empty. The ``retrieval`` block reports what actually
+    served the query — a base install degrades to "fts5" and says so.
+    """
+    backend_arg = backend or _configured_backend(store)
+    viewer = viewer_from(
+        config_path=store.config_path,
+        project=project,
+        agent=agent,
+    )
+    fetch_limit = scoped_fetch_limit(limit, viewer)
+    hits: list[tuple[str, str, str, float]] = []
+    used = backend_arg
+
+    valid_backends = {"auto", "embedding", "fts5", "substring", "hybrid"}
+    if backend_arg not in valid_backends:
+        raise ValueError(
+            f"unknown backend: {backend_arg!r} "
+            f"(expected one of {sorted(valid_backends)})"
+        )
+
+    if backend_arg in ("auto", "hybrid"):
+        emb = index_db.search_semantic(
+            store.kb_dir, query, limit=fetch_limit * 2, min_score=min_score,
+        )
+        try:
+            fts = index_db.search(store.kb_dir, query, limit=fetch_limit * 2)
+        except sqlite3.Error:
+            fts = []
+        hits = rrf_fuse(emb, fts, limit=fetch_limit)
+        if emb and fts:
+            used = "hybrid"
+        elif emb:
+            used = "embedding"
+        elif fts:
+            used = "fts5"
+        if not hits and backend_arg == "auto":
+            hits = store.search_substring(query, limit=fetch_limit)
+            used = "substring"
+    elif backend_arg == "embedding":
+        hits = index_db.search_semantic(
+            store.kb_dir, query, limit=fetch_limit, min_score=min_score,
+        )
+        used = "embedding"
+    elif backend_arg == "fts5":
+        try:
+            hits = index_db.search(store.kb_dir, query, limit=fetch_limit)
+        except sqlite3.Error:
+            hits = []
+        used = "fts5"
+    else:  # substring
+        hits = store.search_substring(query, limit=fetch_limit)
+        used = "substring"
+
+    semantic_ok = index_db.semantic_search_available()
+    scoped = filter_hits(store, hits, viewer, limit=limit)
+    return {
+        "backend": used,
+        "retrieval": {
+            "configured": backend_arg,
+            "used": used,
+            "semantic_available": semantic_ok,
+            "degraded": (
+                backend_arg in ("auto", "hybrid", "embedding")
+                and not semantic_ok
+            ),
+        },
+        "viewer": {"project": viewer.project, "agent": viewer.agent},
+        "hits": [
+            {"kind": k, "id": i, "snippet": sn, "score": sc, "backend": used}
+            for k, i, sn, sc in scoped
+        ],
+    }
+
+
 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:
@@ -440,6 +615,21 @@ def build_context_pack(
     }
     # Determine the backend used (all hits share the same backend in _retrieve).
     result["backend"] = hits[0][4] if hits else "none"
+    # Honesty block: say when a semantic-capable backend actually served
+    # lexical-only results (embeddings extra absent / no embedder registered)
+    # instead of letting "hybrid" imply semantic coverage that never happened.
+    configured = _configured_backend(store)
+    semantic_ok = index_db.semantic_search_available()
+    recency_enabled, _ = _configured_recency(store)
+    result["retrieval"] = {
+        "configured": configured,
+        "used": result["backend"],
+        "semantic_available": semantic_ok,
+        "degraded": (
+            configured in ("auto", "hybrid", "embedding") and not semantic_ok
+        ),
+        "recency": recency_enabled,
+    }
     if explain:
         result["explain"] = [
             {"kind": k, "id": i, "score": sc, "backend": hits[0][4] if hits else "none"}

src/vouch/index_db.py

@@ -477,6 +477,25 @@ def search_embedding(
     return scored[:limit]
 
 
+def semantic_search_available() -> bool:
+    """True when the embeddings extra is installed and an embedder resolves.
+
+    Availability probe only — search_semantic already degrades to [] on its
+    own. Callers use this to *report* degradation (`retrieval.degraded`)
+    instead of silently serving lexical results under a semantic-capable
+    backend name.
+    """
+    try:
+        from .embeddings import get_embedder
+    except ImportError:
+        return False
+    try:
+        get_embedder()
+    except (KeyError, ImportError):
+        return False
+    return True
+
+
 def search_semantic(
     kb_dir: Path,
     query: str,

src/vouch/jsonl_server.py

@@ -132,70 +132,20 @@ def _h_digest(p: dict) -> dict:
 
 
 def _h_search(p: dict) -> dict:
-    from . import index_db
-    from .scoping import filter_hits, scoped_fetch_limit, viewer_from
+    from .context import search_kb
 
-    s = _store()
-    q = p["query"]
-    limit = int(p.get("limit", 10))
-    backend_arg = p.get("backend", "auto")
-    min_score = float(p.get("min_score", 0.0))
-    viewer = viewer_from(
-        config_path=s.config_path,
+    # One shared implementation across MCP / JSONL — see context.search_kb.
+    # No explicit backend on the wire -> the deployment's retrieval.backend
+    # decides, same as kb.context.
+    return search_kb(
+        _store(),
+        query=p["query"],
+        limit=int(p.get("limit", 10)),
+        backend=p.get("backend"),
+        min_score=float(p.get("min_score", 0.0)),
         project=p.get("project"),
         agent=p.get("agent"),
     )
-    fetch_limit = scoped_fetch_limit(limit, viewer)
-    hits: list[tuple[str, str, str, float]] = []
-    used = backend_arg
-
-    valid_backends = {"auto", "embedding", "fts5", "substring", "hybrid"}
-    if backend_arg not in valid_backends:
-        raise ValueError(
-            f"unknown backend: {backend_arg!r} "
-            f"(expected one of {sorted(valid_backends)})"
-        )
-
-    if backend_arg in ("auto", "embedding"):
-        hits = index_db.search_semantic(
-            s.kb_dir, q, limit=fetch_limit, min_score=min_score,
-        )
-        if hits:
-            used = "embedding"
-    if not hits and backend_arg in ("auto", "fts5"):
-        try:
-            hits = index_db.search(s.kb_dir, q, limit=fetch_limit)
-            used = "fts5" if hits else used
-        except Exception:
-            hits = []
-    if not hits and backend_arg in ("auto", "substring"):
-        hits = s.search_substring(q, limit=fetch_limit)
-        used = "substring"
-    if backend_arg == "hybrid":
-        from .embeddings.fusion import (  # type: ignore[import-not-found,import-untyped,unused-ignore]
-            rrf_fuse,
-        )
-        # Hybrid must honour min_score and survive FTS failures the same
-        # way the dedicated fts5 branch does.
-        emb = index_db.search_semantic(
-            s.kb_dir, q, limit=fetch_limit * 2, min_score=min_score,
-        )
-        try:
-            fts = index_db.search(s.kb_dir, q, limit=fetch_limit * 2)
-        except Exception:
-            fts = []
-        hits = rrf_fuse(emb, fts, limit=fetch_limit)
-        used = "hybrid"
-
-    scoped = filter_hits(s, hits, viewer, limit=limit)
-    return {
-        "backend": used,
-        "viewer": {"project": viewer.project, "agent": viewer.agent},
-        "hits": [
-            {"kind": k, "id": i, "snippet": sn, "score": sc, "backend": used}
-            for k, i, sn, sc in scoped
-        ],
-    }
 
 
 def _load_cfg(store: KBStore) -> dict: