feat(telemetry): retrieval_telemetry says what is wrong (#3431)

The tool returned distributions and left the reading to the caller, so every
readout was the same four checks done by hand — #3430's baseline, #3835's rule
near-misses, the #1038 rerank gate. Mechanical, and therefore forgettable.

Tonight's acceptance pass on #3898 was the case for doing this. Reading it by
hand meant catching that two surfaces had `covers_window: false`, that
`prompt_rule`'s floor had moved three times inside the window (which made the
readout self-contradictory: deliveries at 0.622 beside refusals at 0.7199),
and that 15 of 20 near-misses were one record against text no operator wrote.
Miss any of those and the obvious conclusion was "the bar is too tight" — a
floor change that would have injected one preference into every notification.

`warnings` is always present and empty when clean, so its emptiness is an
answer rather than a gap. Each entry carries the numbers that produced it:
"345 calls, 0 declined" is the analysis, "check write_path_rule" is an
instruction to redo it. Five codes — cannot_decline, band_hugs_floor,
no_duration, surfaced_never_pulled, unregistered_source.

cannot_decline has three guards, each a bug it would otherwise cause. Asked
surfaces are exempt (a search returning a list every time is working). An arm
not known to log unconditionally is exempt — that is #3497 exactly, where both
rule arms recorded only their hits, so a decline count of zero was a LOGGING
defect and this warning would have sent the reader to a threshold that was
never involved. Unregistered sources get numbers but no verdict.

`silent_surfaces` is the half the rows cannot show: an arm that emitted
nothing is invisible to every row-based check and looks exactly like an arm
that does not exist. It is driven by a new declared registry,
`retrieval_registry.POINTS` — deliberately NOT `retrieval_surfaces.SURFACES`,
which answers "what can be tuned" and excludes the reserved slots because a
budget of 1 is their feature. This answers "what can be measured", and the
reserved slots belong in it precisely because they are judgeable without being
tunable. A test asserts the two cannot drift apart.

The registry test derives sources from the call sites with `ast`, not grep,
and the difference is not theoretical: `wide_net` and `report_preference`
reach their recorder as `source=SOURCE` through a module constant, so a grep
for `source="` is blind to both — the narrowing #3191 warns about. Three sites
pass `source` as a variable and are declared in FAN_OUT_SITES; the test pins
those sites but not the values they can pass, which is why the
`unregistered_source` warning exists to catch the rest at first fire.

Thresholds are settings (rule 25) defaulted so a fresh install with almost no
data produces no warnings at all (rule 115) — a new user's first readout
naming five broken things would be describing the emptiness.

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

tests/test_retrieval_warnings.py

@@ -0,0 +1,268 @@
+"""`retrieval_telemetry` says what is wrong, rather than implying it (#3431).
+
+WHY THIS IS TESTED AS ARITHMETIC AND NOT THROUGH THE DATABASE. The warnings
+are a pure function of the blocks the readout already built — that is the
+design, so a verdict can never disagree with the numbers printed beside it —
+and the risk in them is not whether rows load. It is whether a rule fires on
+the wrong shape. Every case below is a shape that once produced, or would
+produce, a wrong reading:
+
+  * an arm that never declines, which is either a missing floor or a missing
+    LOG (#3497 — the rule arms recorded only their hits, so their decline
+    count was structurally zero and the obvious warning would have sent a
+    reader to move a threshold that was never involved);
+  * a search that never declines, which is a search working correctly;
+  * a quiet window, where firing every check would describe an empty database
+    rather than a broken one (rule 115);
+  * a floor the band is sitting on, which is the case where tuning changes
+    volume while looking like it changes quality.
+
+The ε and N boundaries are tested from BOTH sides. A threshold asserted only
+where it fires is half-tested: the expensive failure here is a false positive,
+because a readout that cries wolf is one nobody reads.
+"""
+from __future__ import annotations
+
+import pytest
+
+from scribe.services.retrieval_telemetry import (
+    WARN_FLOOR_EPSILON_DEFAULT, WARN_MIN_CALLS_DEFAULT,
+    _compute_warnings, _num, _silent_surfaces,
+)
+
+N = WARN_MIN_CALLS_DEFAULT
+EPS = WARN_FLOOR_EPSILON_DEFAULT
+
+
+def src(**kw) -> dict:
+    """One source block, shaped like the readout builds it."""
+    b = {
+        "calls": kw.pop("calls", 0),
+        "zero_result_calls": kw.pop("zero_result_calls", 0),
+        "top_score": {"p10": kw.pop("p10", None), "p50": None, "p90": None,
+                      "min": None, "max": None},
+        "avg_result_count": None,
+        "p90_duration_ms": kw.pop("p90_duration_ms", 12.0),
+    }
+    b.update(kw)
+    return b
+
+
+def warn(sources, usage=None, rule_usage=None, floors=None,
+         min_calls=N, epsilon=EPS) -> list[dict]:
+    return _compute_warnings(
+        sources, usage or {}, rule_usage or {}, floors or {}, min_calls, epsilon,
+    )
+
+
+def codes(ws, source=None) -> set[str]:
+    return {w["code"] for w in ws if source is None or w["source"] == source}
+
+
+# ── cannot_decline ────────────────────────────────────────────────────────
+
+def test_an_unbidden_arm_that_never_declines_is_flagged() -> None:
+    ws = warn({"auto_inject": src(calls=300, zero_result_calls=0)})
+    assert "cannot_decline" in codes(ws, "auto_inject")
+
+
+def test_the_warning_carries_the_numbers_that_produced_it() -> None:
+    """Not "check auto_inject" — that is an instruction to redo the analysis."""
+    w = next(w for w in warn({"auto_inject": src(calls=300)})
+             if w["code"] == "cannot_decline")
+    assert w["numbers"] == {"calls": 300, "zero_result_calls": 0}
+    assert "300" in w["detail"]
+
+
+def test_a_quiet_window_flags_nothing() -> None:
+    """Three calls returning nothing is an afternoon, not a defect (rule 115)."""
+    assert codes(warn({"auto_inject": src(calls=3, zero_result_calls=0)})) == set()
+
+
+def test_the_call_count_boundary_holds_on_both_sides() -> None:
+    assert "cannot_decline" not in codes(warn({"auto_inject": src(calls=N - 1)}))
+    assert "cannot_decline" in codes(warn({"auto_inject": src(calls=N)}))
+
+
+def test_an_asked_surface_is_never_flagged_for_not_declining() -> None:
+    """A search returning a list every time is a search doing its job."""
+    for asked in ("mcp_search", "wide_net", "rest_search", "browse_search"):
+        assert "cannot_decline" not in codes(
+            warn({asked: src(calls=500, zero_result_calls=0)}), asked)
+
+
+def test_an_arm_that_does_decline_is_not_flagged() -> None:
+    ws = warn({"auto_inject": src(calls=300, zero_result_calls=1)})
+    assert "cannot_decline" not in codes(ws)
+
+
+def test_an_arm_not_known_to_log_unconditionally_is_exempt(monkeypatch) -> None:
+    """#3497: a structurally-zero decline count is a logging bug, not a floor.
+
+    Flagging it would report a ranking problem and send the reader to a
+    threshold that was never involved.
+    """
+    from dataclasses import replace
+
+    from scribe.services import retrieval_registry as reg
+
+    patched = dict(reg.POINTS)
+    patched["auto_inject"] = replace(patched["auto_inject"],
+                                     logs_unconditionally=False)
+    monkeypatch.setattr(reg, "POINTS", patched)
+    assert "cannot_decline" not in codes(
+        warn({"auto_inject": src(calls=300, zero_result_calls=0)}))
+
+
+# ── band_hugs_floor ───────────────────────────────────────────────────────
+
+def test_a_band_sitting_on_its_floor_is_flagged() -> None:
+    ws = warn({"auto_inject": src(calls=100, zero_result_calls=5, p10=0.705)},
+              floors={"auto_inject": 0.70})
+    assert "band_hugs_floor" in codes(ws, "auto_inject")
+
+
+def test_a_band_clear_of_its_floor_is_not() -> None:
+    ws = warn({"auto_inject": src(calls=100, zero_result_calls=5, p10=0.80)},
+              floors={"auto_inject": 0.70})
+    assert "band_hugs_floor" not in codes(ws)
+
+
+@pytest.mark.parametrize("gap,flagged", [
+    (EPS / 2, True),     # inside
+    (EPS, False),        # exactly at the boundary is NOT "hugging"
+    (EPS * 2, False),    # clear
+])
+def test_the_epsilon_boundary_holds_on_both_sides(gap, flagged) -> None:
+    ws = warn({"auto_inject": src(calls=100, zero_result_calls=5,
+                                  p10=round(0.70 + gap, 6))},
+              floors={"auto_inject": 0.70})
+    assert ("band_hugs_floor" in codes(ws)) is flagged
+
+
+def test_an_arm_with_no_readable_floor_is_not_guessed_at() -> None:
+    """Reserved slots borrow a parent's floor; naming it here would attribute
+    the parent's setting to a child that has no dial of its own."""
+    ws = warn({"preference_slot": src(calls=100, zero_result_calls=5, p10=0.7001)},
+              floors={})
+    assert "band_hugs_floor" not in codes(ws)
+
+
+# ── no_duration ───────────────────────────────────────────────────────────
+
+def test_rows_without_timings_are_a_logging_gap() -> None:
+    ws = warn({"auto_inject": src(calls=1, zero_result_calls=1,
+                                  p90_duration_ms=None)})
+    assert "no_duration" in codes(ws)
+
+
+def test_one_untimed_call_is_already_the_defect() -> None:
+    """No minimum on this one — unlike the others, it is not about volume."""
+    ws = warn({"auto_inject": src(calls=1, p90_duration_ms=None)})
+    assert "no_duration" in codes(ws)
+    assert "cannot_decline" not in codes(ws), "volume rules still need volume"
+
+
+def test_a_source_with_no_calls_reports_no_timing_gap() -> None:
+    assert "no_duration" not in codes(warn({"auto_inject": src(calls=0)}))
+
+
+# ── surfaced_never_pulled ─────────────────────────────────────────────────
+
+def test_records_shown_and_never_opened_are_reported_per_corpus() -> None:
+    ws = warn(
+        {},
+        usage={"distinct_notes_surfaced": 171, "distinct_notes_pulled": 40},
+        rule_usage={"distinct_rules_surfaced": 69, "distinct_rules_pulled": 11},
+    )
+    found = {w["numbers"]["corpus"]: w["numbers"]
+             for w in ws if w["code"] == "surfaced_never_pulled"}
+    assert found["notes"]["never_pulled"] == 131
+    assert found["rules"]["never_pulled"] == 58
+
+
+def test_everything_opened_reports_nothing() -> None:
+    ws = warn({}, usage={"distinct_notes_surfaced": 5, "distinct_notes_pulled": 5})
+    assert "surfaced_never_pulled" not in codes(ws)
+
+
+def test_an_empty_corpus_reports_nothing_rather_than_zero() -> None:
+    ws = warn({}, usage={"distinct_notes_surfaced": 0, "distinct_notes_pulled": 0})
+    assert "surfaced_never_pulled" not in codes(ws)
+
+
+# ── unregistered_source ───────────────────────────────────────────────────
+
+def test_a_source_missing_from_the_registry_is_reported() -> None:
+    """The check that covers what the static test cannot reach — a source
+    arriving through a parameter or a dict key."""
+    ws = warn({"a_fourth_write_path_arm": src(calls=50)})
+    assert codes(ws, "a_fourth_write_path_arm") == {"unregistered_source"}
+
+
+def test_an_unregistered_source_gets_no_other_verdict() -> None:
+    """Its numbers are real, but nothing says whether it was asked."""
+    ws = warn({"mystery": src(calls=500, zero_result_calls=0)})
+    assert "cannot_decline" not in codes(ws)
+
+
+def test_every_registered_source_stays_quiet_when_healthy() -> None:
+    """The negative control for the whole suite (rule 167).
+
+    A guard that cannot pass cleanly is not a guard — if a healthy window
+    produced warnings, every assertion above would be meaningless.
+    """
+    from scribe.services.retrieval_registry import POINTS
+
+    healthy = {s: src(calls=100, zero_result_calls=40, p10=0.90)
+               for s in POINTS}
+    assert warn(healthy, floors={s: 0.70 for s in POINTS}) == []
+
+
+# ── silent surfaces ───────────────────────────────────────────────────────
+
+def test_a_registered_point_with_no_rows_is_reported() -> None:
+    quiet = _silent_surfaces({"auto_inject": src(calls=100)}, {}, {}, active=True)
+    assert "prompt_rule" in {p["source"] for p in quiet}
+
+
+def test_a_point_that_emitted_is_not_reported() -> None:
+    quiet = _silent_surfaces({"auto_inject": src(calls=100)}, {}, {}, active=True)
+    assert "auto_inject" not in {p["source"] for p in quiet}
+
+
+def test_a_deliberately_quiet_point_is_never_reported() -> None:
+    """A justified silence must not read as a gap (#2475)."""
+    quiet = {p["source"] for p in _silent_surfaces({}, {}, {}, active=True)}
+    for web_only in ("rest_search", "browse_search", "rest_note", "rest_rule"):
+        assert web_only not in quiet
+
+
+def test_an_inactive_window_reports_no_silence_at_all() -> None:
+    """On a fresh install every point is silent; the list would be the
+    registry printed back (rule 115)."""
+    assert _silent_surfaces({}, {}, {}, active=False) == []
+
+
+def test_a_point_seen_only_in_usage_counts_as_having_emitted() -> None:
+    """The write-path arms and the pull sources never reach `sources` — they
+    live in note_usage_events — so reading only retrieval_logs would report
+    every one of them as silent."""
+    quiet = {p["source"] for p in _silent_surfaces(
+        {}, {"by_source": {"write_path_place": {}}},
+        {"by_source": {"enter_project": {}}}, active=True)}
+    assert "write_path_place" not in quiet
+    assert "enter_project" not in quiet
+
+
+# ── settings parsing ──────────────────────────────────────────────────────
+
+@pytest.mark.parametrize("raw,fallback,want", [
+    ("50", 30, 50),
+    ("0.05", 0.02, 0.05),
+    ("", 30, 30),
+    ("banana", 30, 30),        # a malformed setting must not take the readout down
+    (None, 0.02, 0.02),
+])
+def test_a_setting_falls_back_rather_than_raising(raw, fallback, want) -> None:
+    assert _num(raw, fallback) == want