feat(telemetry): record WHAT the bar turned away, not only how close it came (#3807)

#3670 added `best_available_score` so a threshold could be judged from its
rejections. It records how CLOSE the bar came to firing and not WHAT it
refused, and that is the half a decision actually needs.

Live, pre_tool_rule sits at a ~0.72 bar with a near-miss p90 of 0.7071 —
about 117 declines a day within 0.013 of firing. Dropping to 0.707 would
take that arm from 22 hits a day to roughly 139: six-fold, on a surface
that runs before every Bash call. The percentile says the mass is there.
Nothing said whether it was worth showing.

NEITHER OBVIOUS INSTRUMENT ANSWERS IT. Pull-through cannot: the injected
rule line already carries title and trigger, so a session can comply
without ever calling get_rule, and rule pull-through understates
usefulness by construction. Reading the rejected records can — and
`result_ids` holds only what was RETURNED, so on a zero-result call the
near-missed record had no name at all.

So the id, from the SAME ranked candidate as the score. Both searches
unpack `best` once and read both fields off it, because splitting that
into two expressions is exactly how a later edit pairs a score with its
neighbour's id — and a score attached to the wrong record is worse than no
id, since it invites judging the wrong one and concluding the bar is fine.

write_path withholds the id on the same condition it withholds the score
(#3739): a surviving id beside a null score names a record without saying
what it scored, the pair disagreeing in the other direction.

THE READ PATH IS A LISTING, NOT A STATISTIC — an id cannot be percentiled,
and a reader tuning a bar needs to go and read the records. Opt-in via
`near_miss_samples` (0-20, default 0) so the ordinary readout keeps its
size, and deliberately NOT a window function: this module's one production
outage was a grouped query Postgres rejected, swallowed by the broad
except, every counter reading zero while the mocked tests passed (#2663).
One flat ordered query, overfetched, bucketed in Python — the shape that
lesson prescribes.

Migration 0097, nullable and unbackfilled. Not a foreign key: the table
spans record types and `source` says which, exactly as result_ids works.

The integration guard pins the listing as PER SOURCE. A global LIMIT would
let a noisy source eat the whole quota and leave the surface being tuned
showing nothing — which reads as "nothing was close", the misreading this
milestone has spent itself correcting.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011cPyzNnegXHr5iRMzzy5KJ

tests/test_services_retrieval_telemetry.py

@@ -1120,3 +1120,112 @@ def test_a_caller_that_never_asked_is_assumed_to_have_searched():
             limit=10, project_id=None, is_task=None, results=[],
         )
         assert build.called
+
+
+# ─── the bar's refusals have names now (#3807) ───────────────────────────────
+#
+# #3670 recorded how CLOSE the bar came to firing. That is the half a decision
+# does not need: live, pre_tool_rule sits at a ~0.72 bar with a near-miss p90 of
+# 0.7071, so dropping to 0.707 would take it from 22 hits a day to roughly 139.
+# The percentile says the mass is there and says nothing about whether it is
+# worth showing, and pull-through cannot referee it because the injected rule
+# line already carries title and trigger — a session can comply without ever
+# calling get_rule.
+#
+# Reading the rejected records is the method that answers it, and until now
+# `result_ids` held only what was RETURNED, so on a zero-result call the
+# near-missed record had no name.
+
+
+def test_the_rejected_record_is_named_beside_the_score_it_scored():
+    """Both halves, from one payload, because either alone is unusable.
+
+    A score with no id says a bar nearly fired and not what it nearly fired
+    ABOUT. An id with no score names a record without saying how close it came.
+    """
+    p = _build_payload(
+        user_id=1, source="pre_tool_rule", query="git push --force",
+        threshold=0.72, limit=1, project_id=None, is_task=None,
+        results=[], duration_ms=None, best_available=0.7104,
+        best_available_id=168,
+    )
+    assert p["result_count"] == 0
+    assert p["best_available_score"] == 0.7104
+    assert p["best_available_id"] == 168
+
+
+def test_an_unmeasured_near_miss_names_nothing():
+    """Null, on both halves, and for the reason its sibling is null: a row that
+    did not measure must not invent a record any more than it invents a score."""
+    p = _build_payload(
+        user_id=1, source="auto_inject", query="q", threshold=0.6,
+        limit=3, project_id=None, is_task=None, results=[], duration_ms=None,
+    )
+    assert p["best_available_score"] is None
+    assert p["best_available_id"] is None
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_the_listing_names_the_highest_declines_per_source(_dispose_engine):
+    """Integration, because the listing is a second query against real Postgres
+    and this module's one outage was a query the database rejected in silence.
+
+    Also pins that the listing is PER SOURCE. One flat `ORDER BY score DESC`
+    with a global limit would let a noisy source consume the whole quota and
+    leave the surface you are actually tuning unrepresented — which reads as
+    "nothing was close" for that source, the exact misreading this milestone
+    has spent itself correcting.
+    """
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.retrieval_log import RetrievalLog
+    from scribe.services.retrieval_telemetry import (
+        _insert_retrieval_log, retrieval_summary,
+    )
+
+    UID = 990080
+    # A source whose declines score HIGH, and one whose declines score low.
+    for score, rid in ((0.71, 501), (0.70, 502), (0.69, 503)):
+        await _insert_retrieval_log(_build_payload(
+            user_id=UID, source="pre_tool_rule", query=f"cmd {rid}",
+            threshold=0.72, limit=1, project_id=None, is_task=None,
+            results=[], duration_ms=1.0,
+            best_available=score, best_available_id=rid,
+        ))
+    await _insert_retrieval_log(_build_payload(
+        user_id=UID, source="auto_inject", query="a quieter ask",
+        threshold=0.6, limit=3, project_id=None, is_task=None,
+        results=[], duration_ms=1.0,
+        best_available=0.31, best_available_id=901,
+    ))
+
+    try:
+        out = await retrieval_summary(UID, days=30, near_miss_samples=2)
+        assert out["read_failed"] is False, (
+            "the listing query did not execute — a rejected query here reads "
+            "as an absent listing, which is #2663 again"
+        )
+
+        top = out["sources"]["pre_tool_rule"]["near_miss_records"]
+        assert [r["record_id"] for r in top] == [501, 502], (
+            "the listing must be the HIGHEST declines, in order, capped at the "
+            "requested count"
+        )
+        assert top[0]["score"] == pytest.approx(0.71, abs=1e-4)
+        assert top[0]["query"] == "cmd 501", "the ask is what makes a hit judgeable"
+
+        # The low-scoring source keeps its own slot rather than being crowded
+        # out by the high scorer — this is what a global LIMIT would break.
+        quiet = out["sources"]["auto_inject"]["near_miss_records"]
+        assert [r["record_id"] for r in quiet] == [901]
+
+        off = await retrieval_summary(UID, days=30)
+        assert "near_miss_records" not in off["sources"]["pre_tool_rule"], (
+            "the listing is opt-in; the ordinary readout must not grow"
+        )
+    finally:
+        async with async_session() as s:
+            await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == UID))
+            await s.commit()