diff --git a/src/vouch/context.py b/src/vouch/context.py index 71827be0..0ad97578 100644 --- a/src/vouch/context.py +++ b/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"} diff --git a/src/vouch/index_db.py b/src/vouch/index_db.py index 29d6a859..38d904c0 100644 --- a/src/vouch/index_db.py +++ b/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, diff --git a/src/vouch/jsonl_server.py b/src/vouch/jsonl_server.py index e3785576..d6a12525 100644 --- a/src/vouch/jsonl_server.py +++ b/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: diff --git a/src/vouch/server.py b/src/vouch/server.py index 7f4bfde5..fbfe3bac 100644 --- a/src/vouch/server.py +++ b/src/vouch/server.py @@ -47,7 +47,7 @@ reject, reject_auto_extracted, ) -from .scoping import filter_hits, scoped_fetch_limit, viewer_from +from .scoping import viewer_from from .stats import collect_activity, collect_stats from .storage import ( ArtifactNotFoundError, @@ -190,79 +190,31 @@ def kb_search( query: str, *, limit: int = 10, - backend: str = "auto", + backend: str | None = None, min_score: float = 0.0, project: str | None = None, agent: str | None = None, ) -> dict[str, Any]: """Search the KB. - backend: "auto" (default, embedding then fts5 then substring), - "embedding", "fts5", "substring", or "hybrid". + backend: None (default) defers to retrieval.backend in config.yaml; + "auto"/"hybrid" fuse embedding + fts5 via RRF (substring fallback under + "auto"), or pin "embedding", "fts5", or "substring". The response + retrieval block reports the backend that actually served the query. project/agent: optional viewer context for scope filtering. """ - from . import index_db - store = _store() - viewer = viewer_from( - config_path=store.config_path, + from .context import search_kb + + # One shared implementation across MCP / JSONL — see context.search_kb. + return search_kb( + _store(), + query=query, + limit=limit, + backend=backend, + min_score=min_score, project=project, agent=agent, ) - fetch_limit = scoped_fetch_limit(limit, viewer) - hits: list[tuple[str, str, str, float]] = [] - - def _to_dicts(h: list[tuple[str, str, str, float]], used: str) -> dict[str, Any]: - scoped = filter_hits(store, h, 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 - ], - } - - if backend in ("auto", "embedding"): - hits = index_db.search_semantic( - store.kb_dir, query, limit=fetch_limit, min_score=min_score, - ) - if hits: - return _to_dicts(hits, "embedding") - if backend == "embedding": - return _to_dicts([], "embedding") - - if backend in ("auto", "fts5"): - try: - hits = index_db.search(store.kb_dir, query, limit=fetch_limit) - except Exception: - hits = [] - if hits: - return _to_dicts(hits, "fts5") - if backend == "fts5": - return _to_dicts([], "fts5") - - if backend in ("auto", "substring"): - hits = store.search_substring(query, limit=fetch_limit) - return _to_dicts(hits, "substring") - - if backend == "hybrid": - from .embeddings.fusion import ( # type: ignore[import-not-found,import-untyped,unused-ignore] - rrf_fuse, - ) - # Hybrid must honour min_score (the embedding side can return - # low-relevance noise otherwise) and survive FTS failures the same - # way the dedicated fts5 branch does. - 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 Exception: - fts = [] - hits = rrf_fuse(emb, fts, limit=fetch_limit) - return _to_dicts(hits, "hybrid") - - raise ValueError(f"unknown backend: {backend}") def _load_cfg(store: KBStore) -> dict[str, Any]: diff --git a/src/vouch/storage.py b/src/vouch/storage.py index b2214d7b..887fe979 100644 --- a/src/vouch/storage.py +++ b/src/vouch/storage.py @@ -105,6 +105,14 @@ def _starter_config() -> dict[str, Any]: # embedding | fts5 | substring. See context._retrieve. "backend": "hybrid", "default_limit": 10, + # blend a half-life decay into context-pack scores so fresher + # knowledge outranks equally-relevant stale knowledge. new KBs + # get it on; existing KBs keep byte-identical ordering until + # they add this key. + "recency": { + "enabled": True, + "half_life_days": 90, + }, }, "agents": { "recommended_loop": [ diff --git a/tests/test_retrieval_backend.py b/tests/test_retrieval_backend.py index a3ccd181..1ee6bb6b 100644 --- a/tests/test_retrieval_backend.py +++ b/tests/test_retrieval_backend.py @@ -358,3 +358,167 @@ def test_dedupe_preserves_input_order_not_score_order() -> None: score=0.32, backend="graph", citations=[], freshness="unknown") out = _dedupe_near_duplicates([a, b]) # distinct summaries, a first but lower-scored assert [i.id for i in out] == ["a", "b"] + + +# --- recency decay (retrieval.recency) -------------------------------------- + + +def _set_recency( + store: KBStore, *, enabled: bool, half_life_days: float | None = None +) -> None: + cfg = yaml.safe_load(store.config_path.read_text()) + recency_cfg: dict = {"enabled": enabled} + if half_life_days is not None: + recency_cfg["half_life_days"] = half_life_days + cfg.setdefault("retrieval", {})["recency"] = recency_cfg + store.config_path.write_text(yaml.safe_dump(cfg)) + + +def _backdate_claim(store: KBStore, claim_id: str, *, days: int) -> None: + from datetime import UTC, datetime, timedelta + + path = store.kb_dir / "claims" / f"{claim_id}.yaml" + raw = yaml.safe_load(path.read_text()) + stamp = (datetime.now(UTC) - timedelta(days=days)).isoformat() + raw["created_at"] = stamp + raw["updated_at"] = stamp + path.write_text(yaml.safe_dump(raw)) + + +def _two_claim_fts(monkeypatch: pytest.MonkeyPatch) -> None: + """Deterministic lexical ranking: c1 above c2, no embeddings.""" + monkeypatch.setattr(context.index_db, "search_semantic", lambda *a, **k: []) + monkeypatch.setattr( + context.index_db, "search", + lambda *a, **k: [ + ("claim", "c1", "JWT token rotation", 1.0), + ("claim", "c2", "JWT token rotation policy", 0.9), + ], + ) + + +@pytest.fixture +def two_claim_store(tmp_path: Path) -> KBStore: + s = KBStore.init(tmp_path) + src = s.put_source(b"e") + s.put_claim(Claim(id="c1", text="JWT token rotation", evidence=[src.id])) + s.put_claim(Claim(id="c2", text="JWT token rotation policy", evidence=[src.id])) + health.rebuild_index(s) + return s + + +def test_recency_prefers_fresh_over_stale( + two_claim_store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """A year-old artifact loses to an equally-relevant fresh one when + retrieval.recency is enabled (the starter-config default).""" + _two_claim_fts(monkeypatch) + _set_backend(two_claim_store, "hybrid") + _set_recency(two_claim_store, enabled=True, half_life_days=90) + _backdate_claim(two_claim_store, "c1", days=365) + + pack = context.build_context_pack(two_claim_store, query="JWT", limit=2) + + assert [item["id"] for item in pack["items"]] == ["c2", "c1"] + assert pack["retrieval"]["recency"] is True + + +def test_recency_disabled_keeps_fused_order( + two_claim_store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """Existing KBs (no opt-in) keep byte-identical ordering.""" + _two_claim_fts(monkeypatch) + _set_backend(two_claim_store, "hybrid") + _set_recency(two_claim_store, enabled=False) + _backdate_claim(two_claim_store, "c1", days=365) + + pack = context.build_context_pack(two_claim_store, query="JWT", limit=2) + + assert [item["id"] for item in pack["items"]] == ["c1", "c2"] + assert pack["retrieval"]["recency"] is False + + +# --- search_kb: the one shared kb.search implementation ---------------------- + + +def test_search_kb_auto_reports_actual_backend_and_degradation( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """A semantic-capable backend with no embeddings serves lexical hits and + says so: used=fts5, degraded=true — never a hollow \"hybrid\".""" + monkeypatch.setattr(context.index_db, "search_semantic", lambda *a, **k: []) + monkeypatch.setattr( + context.index_db, "semantic_search_available", lambda: False + ) + + result = context.search_kb(store, query="JWT token rotation") + + assert result["hits"], result + assert result["backend"] == "fts5" + assert result["retrieval"]["configured"] == "hybrid" # starter default + assert result["retrieval"]["used"] == "fts5" + assert result["retrieval"]["semantic_available"] is False + assert result["retrieval"]["degraded"] is True + + +def test_search_kb_hybrid_label_requires_both_retrievers( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr( + context.index_db, "search_semantic", + lambda *a, **k: [("claim", "c1", "JWT token rotation", 0.99)], + ) + monkeypatch.setattr( + context.index_db, "semantic_search_available", lambda: True + ) + + result = context.search_kb(store, query="JWT token rotation", backend="hybrid") + + assert result["backend"] == "hybrid" + assert result["retrieval"]["degraded"] is False + + +def test_search_kb_none_backend_defers_to_config( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + _set_backend(store, "substring") + + result = context.search_kb(store, query="JWT token rotation") + + assert result["retrieval"]["configured"] == "substring" + assert result["backend"] == "substring" + assert result["hits"], result + + +def test_search_kb_auto_substring_fallback_when_all_empty( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + monkeypatch.setattr(context.index_db, "search_semantic", lambda *a, **k: []) + monkeypatch.setattr(context.index_db, "search", lambda *a, **k: []) + _set_backend(store, "auto") + + result = context.search_kb(store, query="JWT token rotation") + + assert result["backend"] == "substring" + assert result["hits"], result + + +def test_search_surfaces_share_search_kb( + store: KBStore, monkeypatch: pytest.MonkeyPatch +) -> None: + """MCP and JSONL kb.search both return the retrieval honesty block — + they delegate to context.search_kb rather than carrying copies.""" + monkeypatch.chdir(store.root) + monkeypatch.setattr(context.index_db, "search_semantic", lambda *a, **k: []) + + from vouch.jsonl_server import handle_request + + jsonl = handle_request( + {"id": 1, "method": "kb.search", "params": {"query": "JWT token rotation"}} + ) + assert jsonl["result"]["retrieval"]["used"] == "fts5" + + from vouch import server as mcp_server + + mcp_result = mcp_server.kb_search(query="JWT token rotation") + assert mcp_result["retrieval"]["used"] == "fts5"