"""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 typing import Any from datetime import datetime, timedelta, timezone from sqlalchemy import case, func, select from scribe.models import async_session from scribe.models.base import iso from scribe.models.note import Note from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent from scribe.models.rule_usage import PULLED as RULE_PULLED from scribe.models.rule_usage import SURFACED as RULE_SURFACED from scribe.models.rule_usage import RuleUsageEvent from scribe.services.rule_usage import is_ambient from scribe.models.retrieval_log import RetrievalLog logger = logging.getLogger(__name__) # Strong references to in-flight inserts — the loop holds tasks only weakly, # and an unreferenced fire-and-forget task can be collected before it runs # (same guard as note_usage, found via #2663). _pending: set[asyncio.Task] = set() # Whether this process already dropped its one warning about failing writes. _reported = False 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, suppressed: int | None = 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. `suppressed` is how many scored hits the caller dropped because the session had already been shown them, and it stays None for callers that cannot know. See the column's comment: None means "not measured here", which is a different fact from 0 and must never render as one. """ 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), "suppressed_count": (None if suppressed is None else int(suppressed)), "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: failures degrade, visibly. WARNING rather than debug — this table is the empirical basis for threshold tuning, and a silent write outage yields a dataset that looks complete while covering only part of the traffic (#2663's shape). Once per process is enough to be found; per-call would flood the log with what it already said. """ global _reported try: async with async_session() as session: session.add(RetrievalLog(**payload)) await session.commit() except Exception: if not _reported: _reported = True logger.warning("retrieval telemetry write failed", exc_info=True) else: 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, Any]], duration_ms: float | None = None, suppressed: int | None = None, ) -> None: """Fire-and-forget: record one retrieval call. `results` needs only `.id` on each record, which is why it is not typed to Note: rules are retrieved too (milestone 307) and land here rather than in note_usage_events. That table's ids are REMAPPED on a backup restore, so a rule id written into it would come back attached to whatever note happened to take that number — silent corruption of the very evidence this exists to provide. retrieval_logs is not restored at all, so it has no such hazard, and `source` already distinguishes the surfaces. 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, suppressed=suppressed, ) except Exception: logger.debug("retrieval telemetry payload build failed", exc_info=True) return try: task = 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") return _pending.add(task) task.add_done_callback(_pending.discard) # --- The read half (#2975) --------------------------------------------------- # Until this existed, `retrieval_logs` was WRITE-ONLY: rows accrued and the only # `select()` over them in the whole tree lived in a test. That made #1038's gate # — "build the reranker once telemetry shows precision is the bottleneck" — # unsatisfiable by construction, and it is why the one real tuning decision on # record (the 0.68 write-path threshold, #2223) was reached by hand-probing the # live instance with eight payloads instead of by reading what was collected. def _bucket(rows: list) -> dict: """A score readout a human can act on, from one aggregate row.""" (calls, zero, cleared, p10, p50, p90, lo, hi, avg_n, dur, measured, supp_calls, supp_zero) = rows return { "calls": int(calls or 0), # A call that returned nothing is not a low-scoring call — it is a # different failure (nothing indexed, filter too narrow), and averaging # it into the score distribution would hide both. "zero_result_calls": int(zero or 0), # How often the best hit actually cleared the threshold in force for # that call. THE precision-adjacent number: a surface that clears its # bar on almost every call is either well-tuned or too loose, and the # score spread below says which. "cleared_threshold": int(cleared or 0), # Of the zeros above, which were the RANKER declining and which were # the reader having seen it already? `zero_result_calls` cannot say, # and only the first kind is evidence about the threshold. # # None — not a zeroed dict — when no row in the window reported it. A # surface that filters inside the search genuinely does not know, and # rendering that as `{"calls": 0}` would state a measurement nobody # made. That substitution is the whole of #3311. "suppression": ( None if not int(measured or 0) else { "measured_calls": int(measured or 0), "calls_with_suppression": int(supp_calls or 0), # Subtract from zero_result_calls for the true ranker declines. "zero_because_already_shown": int(supp_zero or 0), } ), "top_score": { "p10": _round(p10), "p50": _round(p50), "p90": _round(p90), "min": _round(lo), "max": _round(hi), }, "avg_result_count": _round(avg_n), "p90_duration_ms": _round(dur, 1), } def _round(v, places: int = 4): return None if v is None else round(float(v), places) async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict: """What the retrieval telemetry says, per surface, over a window. Three aggregates side by side, each read from the table built for it — NOT a join. `usage` is notes, `rule_usage` is rules, and they stay apart because a few dozen eligible rules blended into thousands of notes is the note ratio with noise on it (milestone 333). `NoteUsageEvent`'s own docstring is explicit that the two are complements ("RetrievalLog tunes the threshold, this tunes the corpus") and that RetrievalLog's JSONB `result_ids` "can't be indexed at" the per-note grain. So the score distribution comes from `retrieval_logs` on its indexed columns, and surfaced-vs-pulled comes from `note_usage_events` at the grain it was built for. Reading each from its own table is both cheaper and more honest than correlating them through JSONB. `usage["by_source"]` is the one join, and it stays INSIDE `note_usage_events` — surfaced rows against pulled rows on note_id. That answers "of the notes this surface chose, how many were opened", which the top-level ratio averages away. It does not cross into `retrieval_logs`, so the sentence above still holds. Scoped to one user's own telemetry. There is no sharing model for a retrieval log — it records what THIS user's agent asked for, including the query text — so an owner filter is the whole access rule here rather than a shortcut around `services/access.py` (P#78 governs shared record kinds). Never raises: a telemetry readout that can break its caller is worse than no readout. It does distinguish "no rows" from "the read failed", because #2663 is exactly the bug where those two looked identical for weeks. """ since = datetime.now(timezone.utc) - timedelta(days=max(1, int(days))) out: dict = { "window_days": int(days), "since": iso(since), "sources": {}, "usage": {}, "rule_usage": {}, "read_failed": False, } cleared = case( ( (RetrievalLog.threshold.isnot(None)) & (RetrievalLog.top_score.isnot(None)) & (RetrievalLog.top_score >= RetrievalLog.threshold), 1, ), else_=0, ) zero = case((RetrievalLog.result_count == 0, 1), else_=0) # Three sums rather than one, because "not measured" and "measured as zero" # are different answers and a single counter cannot hold both. measured = case((RetrievalLog.suppressed_count.isnot(None), 1), else_=0) supp_calls = case((RetrievalLog.suppressed_count > 0, 1), else_=0) supp_zero = case( ((RetrievalLog.result_count == 0) & (RetrievalLog.suppressed_count > 0), 1), else_=0, ) def pct(p: float): return func.percentile_cont(p).within_group(RetrievalLog.top_score.asc()) # Assigned inside the try below; named here so the readout can tell # "this query failed" from "this window has no rows" (#2663). by_source_rows = None rule_rows = None distinct_rules_surfaced = distinct_rules_pulled = 0 try: async with async_session() as session: rows = ( await session.execute( select( RetrievalLog.source, func.count().label("calls"), func.sum(zero).label("zero"), func.sum(cleared).label("cleared"), pct(0.1), pct(0.5), pct(0.9), func.min(RetrievalLog.top_score), func.max(RetrievalLog.top_score), func.avg(RetrievalLog.result_count), func.percentile_cont(0.9).within_group( RetrievalLog.duration_ms.asc() ), func.sum(measured).label("measured"), func.sum(supp_calls).label("supp_calls"), func.sum(supp_zero).label("supp_zero"), ) .where( RetrievalLog.created_at >= since, RetrievalLog.user_id == user_id, ) .group_by(RetrievalLog.source) ) ).all() for row in rows: out["sources"][row[0]] = _bucket(list(row[1:])) # The corpus side, at its own grain. `ambient` mirrors # note_usage.usage_for_notes: an ambient surfacing was not a scored # CHOICE, so folding it into pull-through would understate it. # Grouped by RAW source, then classified in Python. The # alternative — CASE expressions in the GROUP BY — is the shape # that produced #2663: a second case() renders its own expanding # bind names, the database sees two different expressions and # rejects the query, and the broad except swallows it. One CASE is # provably fine (usage_for_notes does it); two is where it broke. # `source` has a handful of distinct values, so grouping on it # directly is cheap and cannot fail that way at all. urows = ( await session.execute( select( NoteUsageEvent.event, NoteUsageEvent.source, func.count().label("n"), ) .where( NoteUsageEvent.created_at >= since, NoteUsageEvent.user_id == user_id, ) .group_by(NoteUsageEvent.event, NoteUsageEvent.source) ) ).all() # Distinct-note counts need their OWN queries, and this is not # fussiness: count(distinct note_id) per (event, source) group # cannot be summed across groups — a note surfaced by two sources # is one distinct note and would be counted twice. A wrong number # labelled "distinct" is worse than no number. from scribe.services.note_usage import AMBIENT_SOURCES as _AMB distinct_surfaced = ( await session.execute( select(func.count(func.distinct(NoteUsageEvent.note_id))).where( NoteUsageEvent.created_at >= since, NoteUsageEvent.user_id == user_id, NoteUsageEvent.event == SURFACED, NoteUsageEvent.source.notin_(_AMB), ) ) ).scalar_one() distinct_pulled = ( await session.execute( select(func.count(func.distinct(NoteUsageEvent.note_id))).where( NoteUsageEvent.created_at >= since, NoteUsageEvent.user_id == user_id, NoteUsageEvent.event == PULLED, ) ) ).scalar_one() # Per-source pull-through, at the NOTE grain (#3311). # # The `urows` query above already groups by source and the loop # below then throws the source away, so until now this readout # could say what the corpus's overall pull-through was and nothing # about WHICH surface earned it. The data was always here; only # the aggregation discarded it. # # It cannot be had by grouping the PULLED rows by source: a pull # records the door it came through (`mcp_get_note`), not the # surface that put the record in front of the agent. Correlating # those within a session is what #2085 ruled out — there is no # session identity server-side and inventing one would mean # threading a client-supplied token through every read path. The # note grain answers the question without one: of the distinct # notes surface X chose, how many did an agent open in this window? # # Guarded separately from the reads above, on #2663's actual # lesson. That outage was a NOVEL SQL SHAPE the database rejected # inside a broad except. This join is the novel shape here, and a # failure in it must not take down two readouts that already work. try: pulled_ids = ( select(NoteUsageEvent.note_id) .where( NoteUsageEvent.created_at >= since, NoteUsageEvent.user_id == user_id, NoteUsageEvent.event == PULLED, # autoescape because `_` is a LIKE wildcard: a bare # like("mcp_%") also matches "mcpX…". The Python half # of this readout uses str.startswith and has no such # hazard; this is the SQL half's version of it. NoteUsageEvent.source.startswith("mcp_", autoescape=True), ) .distinct() .subquery() ) surfaced_pairs = ( select(NoteUsageEvent.source, NoteUsageEvent.note_id) .where( NoteUsageEvent.created_at >= since, NoteUsageEvent.user_id == user_id, NoteUsageEvent.event == SURFACED, ) .distinct() .subquery() ) # DISTINCT on (source, note_id) FIRST, which is what lets the # outer aggregate be a plain count(): the pairs are already # unique, so the left join cannot multiply them and no # count(DISTINCT) is needed to undo damage that never happens. by_source_rows = ( await session.execute( select( surfaced_pairs.c.source, func.count().label("notes_surfaced"), func.count(pulled_ids.c.note_id).label("notes_pulled"), ) .select_from( surfaced_pairs.outerjoin( pulled_ids, pulled_ids.c.note_id == surfaced_pairs.c.note_id, ) ) .group_by(surfaced_pairs.c.source) ) ).all() except Exception: logger.warning("per-source pull-through read failed", exc_info=True) by_source_rows = None # Rules, at their own grain and in their own block (milestone 333). # # Guarded separately from the reads above for the reason `by_source` # is: this table is NEW, and an instance running upgraded code # against un-migrated schema would otherwise take down two readouts # that work perfectly in order to report a third that cannot. # # The queries themselves are the note block's shapes, not novel # ones — a group-by on two indexed columns and two count(distinct). # The distinct counts need their own queries for the same reason # the note ones do: count(distinct rule_id) per group cannot be # summed across groups without double-counting a rule two sources # both touched. try: rule_rows = ( await session.execute( select( RuleUsageEvent.event, RuleUsageEvent.source, func.count().label("n"), ) .where( RuleUsageEvent.created_at >= since, RuleUsageEvent.user_id == user_id, ) .group_by(RuleUsageEvent.event, RuleUsageEvent.source) ) ).all() # The rows carry `source`, so the ranked/ambient split is done # below rather than in SQL — the bulk surfaces started emitting # on 2026-09-03 (#3473), so there IS an ambient class now. # # `distinct_rules_surfaced` deliberately counts BOTH classes. It # answers "how many distinct rules did this install put in front # of an agent at all", which is the denominator for dead weight # — and a rule delivered by the preload a hundred times and # never opened is the most important case that question has. distinct_rules_surfaced = ( await session.execute( select(func.count(func.distinct(RuleUsageEvent.rule_id))) .where( RuleUsageEvent.created_at >= since, RuleUsageEvent.user_id == user_id, RuleUsageEvent.event == RULE_SURFACED, ) ) ).scalar_one() distinct_rules_pulled = ( await session.execute( select(func.count(func.distinct(RuleUsageEvent.rule_id))) .where( RuleUsageEvent.created_at >= since, RuleUsageEvent.user_id == user_id, RuleUsageEvent.event == RULE_PULLED, ) ) ).scalar_one() except Exception: logger.warning("rule usage read failed", exc_info=True) rule_rows = None distinct_rules_surfaced = distinct_rules_pulled = 0 except Exception: logger.warning("retrieval summary read failed", exc_info=True) out["read_failed"] = True return out from scribe.services.note_usage import AMBIENT_SOURCES usage = { "surfaced": 0, "ambient": 0, "pulled": 0, "pulled_by_agent": 0, "pulled_by_human": 0, "distinct_notes_surfaced": int(distinct_surfaced or 0), "distinct_notes_pulled": int(distinct_pulled or 0), } for event, source, n in urows: n = int(n) if event == SURFACED: if source in AMBIENT_SOURCES: usage["ambient"] += n else: usage["surfaced"] += n elif event == PULLED: usage["pulled"] += n # The mcp_/rest_ split is load-bearing (see NoteUsageEvent's own # comment, which names #1038 — this readout's whole purpose). "Is # this record dead weight?" is answered by ANY pull; "was that # injected line useful to the agent?" only by an AGENT pull. So # pull-through, which exists to answer the second, counts mcp_* # only. Both halves are reported so the first question is still # answerable from the same payload. if source.startswith("mcp_"): usage["pulled_by_agent"] += n else: usage["pulled_by_human"] += n # Ranked surfacings in the denominator, agent pulls in the numerator: the # "surfaced often, opened never" reading is only valid where a scored # surface CHOSE the record and an agent was the one who declined it. usage["pull_through"] = ( round(usage["pulled_by_agent"] / usage["surfaced"], 4) if usage["surfaced"] else None ) # The same question, per surface — which is the one the top-level ratio # cannot answer. A corpus average of 0.05 is compatible with one surface # earning its noise and another producing none, and tuning a threshold # needs to know which. # # UPPER BOUND, and say so where it will be read: a pull records the door, # not the surface that led to it, so a note surfaced by two surfaces and # opened once counts as pulled for both. Attribution would need the session # identity #2085 declined to invent. The bound is still decisive in the # direction that matters — a surface reading near zero here is not being # flattered by the double-count. if by_source_rows is None: usage["by_source"] = {} # Distinct from an empty window, for the same reason `read_failed` is. usage["by_source_failed"] = True else: by_source: dict[str, dict] = {} for source, n_surfaced, n_pulled in by_source_rows: n_surfaced, n_pulled = int(n_surfaced or 0), int(n_pulled or 0) ambient = source in AMBIENT_SOURCES by_source[source] = { "notes_surfaced": n_surfaced, "notes_pulled": n_pulled, # None rather than a number on an ambient surface: nothing # CHOSE those records, so "surfaced often, opened never" is not # a judgment about them. The counts stay visible; the ratio # that would be misread does not. "pull_through": ( None if ambient or not n_surfaced else round(n_pulled / n_surfaced, 4) ), "ambient": ambient, } usage["by_source"] = by_source out["usage"] = usage # ── Rules, deliberately a SEPARATE block ──────────────────────────── # # Not folded into `usage`, for two reasons and the second is the one that # bites. The corpora differ by orders of magnitude — a few dozen eligible # rules against thousands of notes — so one blended ratio would be the note # ratio with a little noise on it, and the rule arm's own behaviour would # be undetectable inside it. And `usage` is what existing callers already # read: silently changing what it counts would move a number people have # been comparing across windows, without telling them it now measures # something else. # # `ambient` now carries the bulk deliveries — the SessionStart preload, # `list_always_on_rules`, and every `rules_payload` surface (#3473). Before # they emitted, this block had no ambient key and said the absence was a # fact about the data. It was, and it was also the thing that made the # always-on set impossible to judge: the largest rule surface in the # product was the one surface its own scoreboard could not see. # # READ THE TWO SEPARATELY, ALWAYS. `surfaced` is a claim a ranker made and # a pull can settle. `ambient` is a delivery nobody chose, so a high count # says the set is large and resident, never that it is useful. rule_usage = { "surfaced": 0, "ambient": 0, "pulled": 0, "pulled_by_agent": 0, "pulled_by_human": 0, "distinct_rules_surfaced": int(distinct_rules_surfaced or 0), "distinct_rules_pulled": int(distinct_rules_pulled or 0), } if rule_rows is None: # The FLAG is added, the shape is kept — matching `by_source_failed` # one block up. A caller that renders this must not have to choose # between crashing on a missing key and quietly showing zeros it has no # right to: the keys let it render, and the flag tells it the zeros are # "we could not find out" rather than "nothing happened" (#2663). rule_usage["rule_usage_failed"] = True else: for event, source, n in rule_rows: n = int(n) if event == RULE_SURFACED: # One definition of ranked-vs-ambient, imported rather than # restated — the per-rule badge readout reads the same # predicate, and two spellings of "what counts as surfaced" is # precisely the uneven wiring #3246 found across this system. if is_ambient(source): rule_usage["ambient"] += n else: rule_usage["surfaced"] += n elif event == RULE_PULLED: rule_usage["pulled"] += n # Same split, and it carries MORE weight here than for notes. # The arm's whole claim is "this rule may apply to what you are # writing", and only an agent opening it says the claim landed. # A person browsing the rule list says nothing about the hint. if source.startswith("mcp_"): rule_usage["pulled_by_agent"] += n else: rule_usage["pulled_by_human"] += n # None, not 0.0, when nothing was surfaced — matching the note block. A # ratio of zero asserts "we showed rules and none were opened"; with an # empty numerator AND denominator that is a claim the data does not # support, and it is the reading that would make a brand-new install look # like a broken one. # # RANKED SURFACINGS ONLY in the denominator, and this is the load-bearing # line of the whole change. Pull-through asks "was that hint any use", and # only a surface that CHOSE what it showed can be judged by it. Folding the # preload in would divide the same pulls by a number that grows with every # session and every rule added to the resident set — so enlarging the # always-on set would DEPRESS the arm's measured precision, and trimming it # would flatter it, neither for any reason to do with the arm. The ambient # count sits beside it, unaveraged, and is read as size rather than skill. rule_usage["pull_through"] = ( round(rule_usage["pulled_by_agent"] / rule_usage["surfaced"], 4) if rule_usage["surfaced"] else None ) out["rule_usage"] = rule_usage return out