From 807f478cac2a4291a13058abffe6500dd4b6cabb Mon Sep 17 00:00:00 2001 From: Bryan Van Deusen Date: Mon, 22 Jun 2026 20:10:15 -0400 Subject: [PATCH] =?UTF-8?q?feat(search):=20retrieval=20telemetry=20?= =?UTF-8?q?=E2=80=94=20log=20every=20semantic=20retrieval?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add retrieval_logs (migration 0068) + services/retrieval_telemetry with a fire-and-forget record_retrieval(), wired into the MCP search tool (source=mcp_search) and the REST search route (source=rest_search). Captures query, effective params, and the per-result score distribution so KB-injection thresholds can be tuned from data rather than guessed. Scribe: project 2, milestone 93, task 1032. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Xz4j1H7pjYSjKsEpgcNH5E --- alembic/versions/0068_retrieval_logs.py | 60 +++++++++++ src/scribe/mcp/tools/search.py | 12 ++- src/scribe/models/__init__.py | 1 + src/scribe/models/retrieval_log.py | 70 +++++++++++++ src/scribe/routes/search.py | 16 ++- src/scribe/services/retrieval_telemetry.py | 115 +++++++++++++++++++++ tests/test_services_retrieval_telemetry.py | 112 ++++++++++++++++++++ 7 files changed, 384 insertions(+), 2 deletions(-) create mode 100644 alembic/versions/0068_retrieval_logs.py create mode 100644 src/scribe/models/retrieval_log.py create mode 100644 src/scribe/services/retrieval_telemetry.py create mode 100644 tests/test_services_retrieval_telemetry.py diff --git a/alembic/versions/0068_retrieval_logs.py b/alembic/versions/0068_retrieval_logs.py new file mode 100644 index 0000000..b225816 --- /dev/null +++ b/alembic/versions/0068_retrieval_logs.py @@ -0,0 +1,60 @@ +"""retrieval_logs: per-call semantic-retrieval telemetry for KB-injection tuning + +Revision ID: 0068 +Revises: 0067 +Create Date: 2026-06-22 + +One row per semantic-retrieval call (MCP search tool, REST search route, and — +once it lands — the title-first auto-inject path). Captures the effective query +params and the score distribution of the results so the similarity threshold +and top-k can be tuned from real usage. FK-free on user_id (mirrors app_logs): +telemetry should outlive the row it describes. +""" +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects.postgresql import JSONB + + +revision = "0068" +down_revision = "0067" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.create_table( + "retrieval_logs", + sa.Column("id", sa.Integer(), primary_key=True), + sa.Column( + "created_at", sa.DateTime(timezone=True), nullable=False, + server_default=sa.text("now()"), + ), + sa.Column("user_id", sa.Integer(), nullable=True), + sa.Column("source", sa.Text(), nullable=False), + sa.Column("query", sa.Text(), nullable=True), + sa.Column("threshold", sa.Float(), nullable=True), + sa.Column("limit_n", sa.Integer(), nullable=True), + sa.Column("project_id", sa.Integer(), nullable=True), + sa.Column("is_task", sa.Boolean(), nullable=True), + sa.Column("result_count", sa.Integer(), nullable=False, server_default="0"), + sa.Column("top_score", sa.Float(), nullable=True), + sa.Column("min_score", sa.Float(), nullable=True), + sa.Column("result_ids", JSONB(), nullable=False, server_default=sa.text("'[]'::jsonb")), + sa.Column("duration_ms", sa.Float(), nullable=True), + ) + op.create_index("ix_retrieval_logs_created_at", "retrieval_logs", ["created_at"]) + op.create_index("ix_retrieval_logs_user_id", "retrieval_logs", ["user_id"]) + op.create_index("ix_retrieval_logs_source", "retrieval_logs", ["source"]) + op.create_index( + "ix_retrieval_logs_source_created_at", + "retrieval_logs", + ["source", sa.text("created_at DESC")], + ) + + +def downgrade() -> None: + op.drop_index("ix_retrieval_logs_source_created_at", table_name="retrieval_logs") + op.drop_index("ix_retrieval_logs_source", table_name="retrieval_logs") + op.drop_index("ix_retrieval_logs_user_id", table_name="retrieval_logs") + op.drop_index("ix_retrieval_logs_created_at", table_name="retrieval_logs") + op.drop_table("retrieval_logs") diff --git a/src/scribe/mcp/tools/search.py b/src/scribe/mcp/tools/search.py index 215df01..12bc32b 100644 --- a/src/scribe/mcp/tools/search.py +++ b/src/scribe/mcp/tools/search.py @@ -7,8 +7,11 @@ working. Differences from fable-mcp: """ from __future__ import annotations +import time + from scribe.mcp._context import current_user_id -from scribe.services.embeddings import semantic_search_notes +from scribe.services.embeddings import DEFAULT_SIMILARITY_THRESHOLD, semantic_search_notes +from scribe.services.retrieval_telemetry import record_retrieval async def search( @@ -43,10 +46,17 @@ async def search( uid = current_user_id() limit = max(1, min(limit, 50)) is_task = {"note": False, "task": True}.get(content_type) # None => any + t0 = time.perf_counter() raw = await semantic_search_notes( uid, q, limit=limit, is_task=is_task, project_id=project_id or None, ) + record_retrieval( + user_id=uid, source="mcp_search", query=q, + threshold=DEFAULT_SIMILARITY_THRESHOLD, limit=limit, + project_id=project_id or None, is_task=is_task, results=raw, + duration_ms=(time.perf_counter() - t0) * 1000.0, + ) return { "results": [ { diff --git a/src/scribe/models/__init__.py b/src/scribe/models/__init__.py index 080af91..982ccee 100644 --- a/src/scribe/models/__init__.py +++ b/src/scribe/models/__init__.py @@ -26,6 +26,7 @@ from scribe.models.app_log import AppLog # noqa: E402, F401 from scribe.models.password_reset import PasswordResetToken # noqa: E402, F401 from scribe.models.invitation import InvitationToken # noqa: E402, F401 from scribe.models.embedding import NoteEmbedding # noqa: E402, F401 +from scribe.models.retrieval_log import RetrievalLog # noqa: E402, F401 from scribe.models.project import Project # noqa: E402, F401 from scribe.models.event import Event # noqa: E402, F401 from scribe.models.milestone import Milestone # noqa: E402, F401 diff --git a/src/scribe/models/retrieval_log.py b/src/scribe/models/retrieval_log.py new file mode 100644 index 0000000..6630364 --- /dev/null +++ b/src/scribe/models/retrieval_log.py @@ -0,0 +1,70 @@ +from datetime import datetime, timezone + +from sqlalchemy import Boolean, DateTime, Float, Index, Integer, Text +from sqlalchemy.dialects.postgresql import JSONB +from sqlalchemy.orm import Mapped, mapped_column + +from scribe.models import Base + + +class RetrievalLog(Base): + """One row per semantic-retrieval call, for KB-injection tuning. + + Captures what a query asked for, what came back, and the score + distribution of the results — the empirical basis for tuning the + similarity threshold and top-k per surface. `result_ids` holds the ranked + hits (id + score + rank) so a later pass can correlate "what we surfaced" + against "what the agent then fetched/referenced". + + Deliberately FK-free on user_id (mirrors AppLog): telemetry should outlive + the row it describes, and a deleted user shouldn't cascade away history. + """ + + __tablename__ = "retrieval_logs" + + id: Mapped[int] = mapped_column(primary_key=True) + created_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), default=lambda: datetime.now(timezone.utc) + ) + user_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + # Retrieval surface: 'mcp_search' | 'rest_search' | 'auto_inject' | ... + source: Mapped[str] = mapped_column(Text, nullable=False) + query: Mapped[str | None] = mapped_column(Text, nullable=True) + # Effective parameters actually used for this call. + threshold: Mapped[float | None] = mapped_column(Float, nullable=True) + limit_n: Mapped[int | None] = mapped_column(Integer, nullable=True) + project_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + # The content-type filter as passed to semantic_search_notes: True=tasks, + # False=notes, NULL=any. + is_task: Mapped[bool | None] = mapped_column(Boolean, nullable=True) + result_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0) + top_score: Mapped[float | None] = mapped_column(Float, nullable=True) + min_score: Mapped[float | None] = mapped_column(Float, nullable=True) + # [{"id": int, "score": float, "rank": int}, ...], highest-first. + result_ids: Mapped[list] = mapped_column(JSONB, nullable=False, default=list) + duration_ms: Mapped[float | None] = mapped_column(Float, nullable=True) + + __table_args__ = ( + Index("ix_retrieval_logs_created_at", "created_at"), + Index("ix_retrieval_logs_user_id", "user_id"), + Index("ix_retrieval_logs_source", "source"), + Index("ix_retrieval_logs_source_created_at", "source", created_at.desc()), + ) + + def to_dict(self) -> dict: + return { + "id": self.id, + "created_at": self.created_at.isoformat() if self.created_at else None, + "user_id": self.user_id, + "source": self.source, + "query": self.query, + "threshold": self.threshold, + "limit_n": self.limit_n, + "project_id": self.project_id, + "is_task": self.is_task, + "result_count": self.result_count, + "top_score": self.top_score, + "min_score": self.min_score, + "result_ids": self.result_ids, + "duration_ms": self.duration_ms, + } diff --git a/src/scribe/routes/search.py b/src/scribe/routes/search.py index ba34878..6e5254e 100644 --- a/src/scribe/routes/search.py +++ b/src/scribe/routes/search.py @@ -1,7 +1,14 @@ +import time + from quart import Blueprint, jsonify, request from scribe.auth import login_required, get_current_user_id from scribe.services.embeddings import semantic_search_notes +from scribe.services.retrieval_telemetry import record_retrieval + +# This route searches with a looser floor than the MCP tool default — it powers +# an interactive feed where loosely-related hits still have value. +_REST_SEARCH_THRESHOLD = 0.3 search_bp = Blueprint("search", __name__, url_prefix="/api/search") @@ -27,8 +34,15 @@ async def search_route(): limit = min(request.args.get("limit", 10, type=int), 50) is_task = _content_type_to_is_task(content_type) + t0 = time.perf_counter() results = await semantic_search_notes( - uid, q, limit=limit, is_task=is_task, threshold=0.3 + uid, q, limit=limit, is_task=is_task, threshold=_REST_SEARCH_THRESHOLD + ) + record_retrieval( + user_id=uid, source="rest_search", query=q, + threshold=_REST_SEARCH_THRESHOLD, limit=limit, + project_id=None, is_task=is_task, results=results, + duration_ms=(time.perf_counter() - t0) * 1000.0, ) return jsonify({ "results": [ diff --git a/src/scribe/services/retrieval_telemetry.py b/src/scribe/services/retrieval_telemetry.py new file mode 100644 index 0000000..7ab8d8a --- /dev/null +++ b/src/scribe/services/retrieval_telemetry.py @@ -0,0 +1,115 @@ +"""Retrieval telemetry — one RetrievalLog row per semantic-retrieval call. + +This is the empirical basis for KB-injection tuning: it records what each query +asked for, the score distribution of what came back, and the effective params, +so the similarity threshold and top-k can be tuned from data rather than guessed. + +Design notes: + - Fire-and-forget, mirroring upsert_note_embedding: `record_retrieval` extracts + the primitives it needs SYNCHRONOUSLY (while the caller's Note objects are + still valid) and schedules the DB insert as a background task, so logging + never adds latency to — or can break — the search response. + - Result objects are reduced to {id, score, rank} before scheduling; the + background writer touches only plain data, never a possibly-detached ORM row. + - Every failure path is swallowed: telemetry must never take down retrieval. +""" +from __future__ import annotations + +import asyncio +import logging + +from scribe.models import async_session +from scribe.models.note import Note +from scribe.models.retrieval_log import RetrievalLog + +logger = logging.getLogger(__name__) + + +def _build_payload( + *, + user_id: int | None, + source: str, + query: str | None, + threshold: float | None, + limit: int | None, + project_id: int | None, + is_task: bool | None, + results: list[tuple[float, Note]], + duration_ms: float | None, +) -> dict: + """Reduce a retrieval call to a flat, JSON-safe RetrievalLog payload. + + Pure and synchronous (no DB, no event loop) so it is unit-testable and safe + to run inline before scheduling the write. `results` is the + `(score, Note)` list from semantic_search_notes, already highest-first. + """ + items = [ + {"id": int(note.id), "score": round(float(score), 5), "rank": rank} + for rank, (score, note) in enumerate(results) + ] + scores = [it["score"] for it in items] + return { + "user_id": user_id, + "source": source, + "query": query, + "threshold": threshold, + "limit_n": limit, + "project_id": project_id, + "is_task": is_task, + "result_count": len(items), + "top_score": (scores[0] if scores else None), + "min_score": (scores[-1] if scores else None), + "result_ids": items, + "duration_ms": (round(duration_ms, 2) if duration_ms is not None else None), + } + + +async def _insert_retrieval_log(payload: dict) -> None: + """Persist one RetrievalLog row. Best-effort: all errors are swallowed.""" + try: + async with async_session() as session: + session.add(RetrievalLog(**payload)) + await session.commit() + except Exception: + logger.debug("retrieval telemetry write skipped", exc_info=True) + + +def record_retrieval( + *, + user_id: int | None, + source: str, + query: str | None, + threshold: float | None, + limit: int | None, + project_id: int | None, + is_task: bool | None, + results: list[tuple[float, Note]], + duration_ms: float | None = None, +) -> None: + """Fire-and-forget: record one retrieval call. + + Builds the payload inline (synchronously) then schedules the insert so the + caller returns immediately. Never raises — telemetry must not affect search. + """ + try: + payload = _build_payload( + user_id=user_id, + source=source, + query=query, + threshold=threshold, + limit=limit, + project_id=project_id, + is_task=is_task, + results=results, + duration_ms=duration_ms, + ) + except Exception: + logger.debug("retrieval telemetry payload build failed", exc_info=True) + return + + try: + asyncio.get_running_loop().create_task(_insert_retrieval_log(payload)) + except RuntimeError: + # No running loop (e.g. called from sync context outside the app) — + # skip rather than block. The app paths always run on the loop. + logger.debug("retrieval telemetry skipped — no running event loop") diff --git a/tests/test_services_retrieval_telemetry.py b/tests/test_services_retrieval_telemetry.py new file mode 100644 index 0000000..5194eb8 --- /dev/null +++ b/tests/test_services_retrieval_telemetry.py @@ -0,0 +1,112 @@ +"""Tests for services.retrieval_telemetry. + +_build_payload is pure (no DB, no loop) and gets unit coverage. The persistence +path (_insert_retrieval_log + the RetrievalLog model / JSONB roundtrip) is an +integration test against real Postgres. +""" +from types import SimpleNamespace + +import pytest +import pytest_asyncio + +from scribe.services.retrieval_telemetry import ( + _build_payload, + record_retrieval, +) + + +def _note(nid): + """Minimal stand-in — _build_payload only reads .id.""" + return SimpleNamespace(id=nid) + + +# ─── _build_payload (pure) ─────────────────────────────────────────────────── + + +def test_build_payload_ranks_and_score_bounds(): + results = [(0.91, _note(11)), (0.72, _note(22)), (0.55, _note(33))] + p = _build_payload( + user_id=7, source="mcp_search", query="hello", threshold=0.45, + limit=10, project_id=3, is_task=None, results=results, duration_ms=12.345, + ) + assert p["result_count"] == 3 + assert p["top_score"] == 0.91 + assert p["min_score"] == 0.55 + assert [it["rank"] for it in p["result_ids"]] == [0, 1, 2] + assert [it["id"] for it in p["result_ids"]] == [11, 22, 33] + assert p["duration_ms"] == 12.35 # rounded to 2dp + assert p["user_id"] == 7 and p["project_id"] == 3 and p["threshold"] == 0.45 + + +def test_build_payload_empty_results(): + p = _build_payload( + user_id=1, source="rest_search", query="x", threshold=0.3, + limit=5, project_id=None, is_task=False, results=[], duration_ms=None, + ) + assert p["result_count"] == 0 + assert p["top_score"] is None and p["min_score"] is None + assert p["result_ids"] == [] + assert p["duration_ms"] is None + + +def test_build_payload_rounds_scores_to_5dp(): + p = _build_payload( + user_id=1, source="mcp_search", query="q", threshold=0.45, + limit=1, project_id=None, is_task=None, + results=[(0.123456789, _note(1))], duration_ms=0.0, + ) + assert p["result_ids"][0]["score"] == 0.12346 + + +def test_record_retrieval_without_event_loop_is_safe(): + """Called from a sync context (no running loop) it must swallow and return, + never raise — telemetry can't be allowed to break a caller.""" + # No event loop running in this plain sync test. + assert record_retrieval( + user_id=1, source="mcp_search", query="q", threshold=0.45, + limit=10, project_id=None, is_task=None, + results=[(0.9, _note(1))], + ) is None + + +# ─── persistence (integration) ─────────────────────────────────────────────── + + +@pytest_asyncio.fixture +async def _dispose_engine(): + from scribe.models import engine + yield + await engine.dispose() + + +@pytest.mark.integration +@pytest.mark.asyncio +async def test_insert_retrieval_log_roundtrip(_dispose_engine): + from sqlalchemy import delete, select + + from scribe.models import async_session + from scribe.models.retrieval_log import RetrievalLog + from scribe.services.retrieval_telemetry import _insert_retrieval_log + + payload = _build_payload( + user_id=990001, source="mcp_search", query="pgvector tuning", + threshold=0.45, limit=10, project_id=None, is_task=None, + results=[(0.88, _note(501)), (0.61, _note(502))], duration_ms=9.9, + ) + await _insert_retrieval_log(payload) + + async with async_session() as s: + row = ( + await s.execute( + select(RetrievalLog).where(RetrievalLog.user_id == 990001) + ) + ).scalars().first() + assert row is not None + assert row.source == "mcp_search" + assert row.result_count == 2 + assert row.top_score == 0.88 + # JSONB roundtrips as a list of dicts with the expected shape. + assert row.result_ids[0] == {"id": 501, "score": 0.88, "rank": 0} + assert row.created_at is not None # server_default now() + await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == 990001)) + await s.commit()