FabledScribe/tests/test_services_retrieval_telemetry.py

"""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

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


# ─── suppression: "not measured" is not "none" (#3497) ───────────────────────


def test_a_caller_that_cannot_measure_suppression_stores_null():
    """The distinction the whole column exists for.

    A surface that passes its exclusions into the search never sees what was
    dropped. Storing 0 would assert a clean run nobody observed — reading an
    artifact as a measurement, which is exactly #3311's mistake.
    """
    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["suppressed_count"] is None, "unmeasured must not render as zero"


def test_a_caller_that_measured_no_suppression_stores_zero():
    """The other side of it. Zero is a real observation and must survive."""
    p = _build_payload(
        user_id=1, source="pre_tool_rule", query="git status", threshold=0.6,
        limit=1, project_id=None, is_task=None, results=[], duration_ms=None,
        suppressed=0,
    )
    assert p["suppressed_count"] == 0


def test_the_count_of_hits_the_reader_already_held_is_carried():
    p = _build_payload(
        user_id=1, source="write_path_rule", query="code", threshold=0.6,
        limit=2, project_id=None, is_task=None, results=[], duration_ms=None,
        suppressed=2,
    )
    assert p["result_count"] == 0
    assert p["suppressed_count"] == 2, (
        "a zero row that was really two repeats must be distinguishable from "
        "a zero row where the ranker found nothing"
    )


def test_the_readout_reports_unmeasured_suppression_as_none():
    """`_bucket` renders the aggregate. No row reporting it → null, never a
    zeroed dict: a zeroed dict states a measurement nobody made."""
    from scribe.services.retrieval_telemetry import _bucket

    # calls, zero, p10, p50, p90, min, max, avg_n, dur,
    # measured, supp_calls, supp_zero, miss_calls, miss_p50, miss_p90, miss_max
    unmeasured = _bucket([326, 114, 0.6, 0.68, 0.77, 0.55, 0.85, 1.7, 130.9,
                          0, 0, 0, 0, None, None, None])
    assert unmeasured["suppression"] is None

    measured = _bucket([35, 34, 0.75, 0.75, 0.75, 0.75, 0.75, 0.03, 51.9,
                        35, 9, 9, 0, None, None, None])
    assert measured["suppression"] == {
        "measured_calls": 35,
        "calls_with_suppression": 9,
        "zero_because_already_shown": 9,
    }
    # The number the threshold is actually tuned from.
    assert measured["zero_result_calls"] - 9 == 25


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.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()


# ─── the read half: retrieval_summary (integration) ──────────────────────────
# Integration, not mocked, and deliberately so. #2663 is the bug where a
# GROUP BY the database rejected was swallowed by a broad except, so every
# counter read zero in production while the writes landed fine and the mocked
# tests passed. `retrieval_summary` runs a grouped aggregate with
# percentile_cont ... WITHIN GROUP and a two-label CASE — precisely the shape
# that failed then. Only a real Postgres can say it parses.


@pytest.mark.integration
@pytest.mark.asyncio
async def test_retrieval_summary_reads_what_the_writer_wrote(_dispose_engine):
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.note_usage import NoteUsageEvent
    from scribe.models.retrieval_log import RetrievalLog
    from scribe.services.retrieval_telemetry import (
        _insert_retrieval_log, retrieval_summary,
    )

    UID = 990002
    # Three auto_inject calls at a 0.55 bar: two clear it, one does not.
    # Plus one call that returned nothing at all — a different failure from a
    # low-scoring one, and the readout must not blend them.
    for score in (0.91, 0.72, 0.40):
        await _insert_retrieval_log(_build_payload(
            user_id=UID, source="auto_inject", query="q", threshold=0.55,
            limit=3, project_id=None, is_task=None,
            results=[(score, _note(1))], duration_ms=5.0,
        ))
    await _insert_retrieval_log(_build_payload(
        user_id=UID, source="auto_inject", query="q", threshold=0.55,
        limit=3, project_id=None, is_task=None, results=[], duration_ms=5.0,
    ))
    # A second surface, so the GROUP BY has something to separate.
    await _insert_retrieval_log(_build_payload(
        user_id=UID, source="mcp_search", query="q", threshold=0.45,
        limit=10, project_id=None, is_task=None,
        results=[(0.80, _note(2))], duration_ms=11.0,
    ))
    # Corpus side: two ranked surfacings, one ambient, one pull.
    async with async_session() as s:
        s.add_all([
            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
            NoteUsageEvent(user_id=UID, note_id=2, event="surfaced", source="auto_inject"),
            NoteUsageEvent(user_id=UID, note_id=3, event="surfaced", source="enter_project"),
            NoteUsageEvent(user_id=UID, note_id=1, event="pulled", source="mcp_get_note"),
        ])
        await s.commit()

    try:
        out = await retrieval_summary(UID, days=30)

        assert out["read_failed"] is False, "the aggregate did not execute"
        ai = out["sources"]["auto_inject"]
        assert ai["calls"] == 4
        assert ai["zero_result_calls"] == 1
        assert "cleared_threshold" not in ai, (
            "the tautology is back: it was true exactly when result_count > 0, "
            "so it reported nothing zero_result_calls did not (#3670)"
        )
        # p50 over the three scored calls; the empty one contributes no score.
        assert ai["top_score"]["p50"] == pytest.approx(0.72, abs=1e-4)
        assert ai["top_score"]["min"] == pytest.approx(0.40, abs=1e-4)
        assert ai["top_score"]["max"] == pytest.approx(0.91, abs=1e-4)
        assert out["sources"]["mcp_search"]["calls"] == 1

        u = out["usage"]
        assert u["surfaced"] == 2 and u["ambient"] == 1 and u["pulled"] == 1
        assert u["distinct_notes_surfaced"] == 2
        # The pull came from `mcp_get_note`, so it counts as an AGENT pull
        # and drives pull_through; a human `rest_*` pull would not.
        assert u["pulled_by_agent"] == 1 and u["pulled_by_human"] == 0
        assert u["pull_through"] == pytest.approx(0.5)
    finally:
        async with async_session() as s:
            await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == UID))
            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
            await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_retrieval_summary_is_empty_not_broken_for_a_fresh_install(_dispose_engine):
    """Rule #115: an install with no telemetry gets a coherent zero readout,
    and `read_failed` stays False — the distinction #2663 says must exist."""
    from scribe.services.retrieval_telemetry import retrieval_summary

    out = await retrieval_summary(990003, days=30)
    assert out["read_failed"] is False
    assert out["sources"] == {}
    assert out["usage"]["pull_through"] is None   # no division by zero
    assert out["usage"]["surfaced"] == 0
    # An empty dict, not a missing key and not a failure flag — the same
    # "no rows" / "read broke" distinction the rest of this readout keeps.
    assert out["usage"]["by_source"] == {}
    assert "by_source_failed" not in out["usage"]


@pytest.mark.integration
@pytest.mark.asyncio
async def test_retrieval_summary_sees_only_its_own_users_telemetry(_dispose_engine):
    """A retrieval log records what one user's agent asked for, query text
    included. The owner filter is the access rule, so it gets a test."""
    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,
    )

    await _insert_retrieval_log(_build_payload(
        user_id=990004, source="auto_inject", query="theirs", threshold=0.55,
        limit=3, project_id=None, is_task=None, results=[(0.9, _note(1))],
        duration_ms=1.0,
    ))
    try:
        assert (await retrieval_summary(990005, days=30))["sources"] == {}
        assert (await retrieval_summary(990004, days=30))["sources"]["auto_inject"]["calls"] == 1
    finally:
        async with async_session() as s:
            await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == 990004))
            await s.commit()


# ─── per-source pull-through (#3311) ─────────────────────────────────────────
# Integration for the same reason the block above is: this is a self-join with
# two DISTINCT subqueries and a LIKE escape, which is a new SQL shape in a
# module whose one production outage (#2663) was a new SQL shape the database
# rejected inside a broad except. A mock would pass on a query Postgres refuses.


@pytest.mark.integration
@pytest.mark.asyncio
async def test_by_source_separates_a_surface_that_earns_its_noise_from_one_that_does_not(
    _dispose_engine,
):
    """The whole point: the corpus average cannot say WHICH surface is working.

    Two ranked surfaces, identical volume, opposite outcomes — and a top-level
    ratio that describes neither of them.
    """
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.note_usage import NoteUsageEvent
    from scribe.services.retrieval_telemetry import retrieval_summary

    UID = 990010
    async with async_session() as s:
        s.add_all([
            # auto_inject chose note 1 three times and note 2 once. Three
            # surfacings of one note is ONE note surfaced — the DISTINCT that
            # keeps the join from multiplying rows is what this pins.
            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
            NoteUsageEvent(user_id=UID, note_id=2, event="surfaced", source="auto_inject"),
            # write_path_semantic chose two notes and got nothing opened.
            NoteUsageEvent(user_id=UID, note_id=3, event="surfaced", source="write_path_semantic"),
            NoteUsageEvent(user_id=UID, note_id=4, event="surfaced", source="write_path_semantic"),
            # One agent pull, of a note only auto_inject surfaced.
            NoteUsageEvent(user_id=UID, note_id=1, event="pulled", source="mcp_get_note"),
        ])
        await s.commit()

    try:
        out = await retrieval_summary(UID, days=30)
        assert out["read_failed"] is False
        by_source = out["usage"]["by_source"]
        assert "by_source_failed" not in out["usage"], "the join did not execute"

        ai = by_source["auto_inject"]
        assert ai["notes_surfaced"] == 2, "three surfacings of note 1 are one note"
        assert ai["notes_pulled"] == 1
        assert ai["pull_through"] == pytest.approx(0.5)

        wp = by_source["write_path_semantic"]
        assert wp["notes_surfaced"] == 2
        assert wp["notes_pulled"] == 0
        # 0.0, NOT None. "This surface produced nothing" is a finding; None is
        # what a surface with no data reads as, and they must not look alike.
        assert wp["pull_through"] == 0.0

        # And the number that exists today, which is true of neither surface:
        # one agent pull over six ranked surfacings.
        assert out["usage"]["pull_through"] == pytest.approx(1 / 6, abs=1e-4)
    finally:
        async with async_session() as s:
            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
            await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_an_ambient_surface_reports_its_counts_but_no_ratio(_dispose_engine):
    """`enter_project` bulk-loads records; nothing CHOSE them. "Surfaced often,
    opened never" is not a judgment about a record that was never picked, so the
    counts stay visible and the ratio that would be misread is null."""
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.note_usage import NoteUsageEvent
    from scribe.services.retrieval_telemetry import retrieval_summary

    UID = 990011
    async with async_session() as s:
        s.add_all([
            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="enter_project"),
            NoteUsageEvent(user_id=UID, note_id=2, event="surfaced", source="enter_project"),
        ])
        await s.commit()

    try:
        row = (await retrieval_summary(UID, days=30))["usage"]["by_source"]["enter_project"]
        assert row["ambient"] is True
        assert row["notes_surfaced"] == 2
        assert row["pull_through"] is None
    finally:
        async with async_session() as s:
            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
            await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_the_agent_pull_filter_does_not_treat_its_underscore_as_a_wildcard(
    _dispose_engine,
):
    """`_` is a LIKE wildcard, so an unescaped `LIKE 'mcp_%'` also matches
    `mcpXsomething`. The Python half of this readout uses str.startswith and
    cannot have the bug; the SQL half needs autoescape to match it, and nothing
    else in the payload would reveal the difference."""
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.note_usage import NoteUsageEvent
    from scribe.services.retrieval_telemetry import retrieval_summary

    UID = 990012
    async with async_session() as s:
        s.add_all([
            NoteUsageEvent(user_id=UID, note_id=1, event="surfaced", source="auto_inject"),
            # Not an agent pull: the door is `mcpXget_note`, not `mcp_get_note`.
            NoteUsageEvent(user_id=UID, note_id=1, event="pulled", source="mcpXget_note"),
        ])
        await s.commit()

    try:
        row = (await retrieval_summary(UID, days=30))["usage"]["by_source"]["auto_inject"]
        assert row["notes_pulled"] == 0, "a wildcard match counted a non-agent pull"
        assert row["pull_through"] == 0.0
    finally:
        async with async_session() as s:
            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
            await s.commit()


# ─── rule usage (milestone 333 step 3) ───────────────────────────────────────
# Integration, for the same reason the block above is: these are real GROUP BYs
# and count(distinct) against a table that did not exist a commit ago, in a
# module whose one production outage (#2663) was a SQL shape the database
# rejected inside a broad except. A mock would agree with whatever the code
# does, including nothing.


async def _rule_events(uid, rows):
    """Write (event, source) pairs for one rule and hand back a cleanup."""
    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.rule_usage import RuleUsageEvent

    async with async_session() as s:
        s.add_all([
            RuleUsageEvent(user_id=uid, rule_id=rid, event=ev, source=src)
            for rid, ev, src in rows
        ])
        await s.commit()

    async def cleanup():
        async with async_session() as s:
            await s.execute(
                delete(RuleUsageEvent).where(RuleUsageEvent.user_id == uid)
            )
            await s.commit()

    return cleanup


@pytest.mark.integration
@pytest.mark.asyncio
async def test_rule_usage_is_a_coherent_zero_on_a_fresh_install(_dispose_engine):
    """Every rule in an existing install predates this table, so "no events" is
    the normal state for a while. It must read as zero, not as a missing key
    and not as a failure — the same "no rows" / "read broke" distinction the
    rest of this readout keeps (#2663).

    `pull_through` is None rather than 0.0, matching the note block: a ratio of
    zero asserts "rules were shown and none opened", which with an empty
    numerator AND denominator is a claim the data does not support.
    """
    from scribe.services.retrieval_telemetry import retrieval_summary

    out = await retrieval_summary(990010, days=30)
    assert out["read_failed"] is False
    assert "rule_usage_failed" not in out["rule_usage"]
    assert out["rule_usage"]["surfaced"] == 0
    assert out["rule_usage"]["pulled"] == 0
    assert out["rule_usage"]["distinct_rules_surfaced"] == 0
    assert out["rule_usage"]["pull_through"] is None


@pytest.mark.integration
@pytest.mark.asyncio
async def test_an_agent_reading_a_surfaced_rule_is_what_moves_the_ratio(_dispose_engine):
    """The whole point of the milestone: the arm can now be told apart from a
    bar it cannot fail to clear."""
    from scribe.services.retrieval_telemetry import retrieval_summary

    cleanup = await _rule_events(990011, [
        (5001, "surfaced", "write_path_rule"),
        (5002, "surfaced", "write_path_rule"),
        (5001, "pulled", "mcp_get_rule"),
    ])
    try:
        ru = (await retrieval_summary(990011, days=30))["rule_usage"]
        assert ru["surfaced"] == 2
        assert ru["pulled"] == 1
        assert ru["pulled_by_agent"] == 1
        assert ru["pulled_by_human"] == 0
        assert ru["distinct_rules_surfaced"] == 2
        assert ru["distinct_rules_pulled"] == 1
        assert ru["pull_through"] == 0.5
    finally:
        await cleanup()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_a_person_browsing_the_rule_list_does_not_move_the_ratio(_dispose_engine):
    """The mcp_/rest_ split, and it carries more weight here than for notes.

    The arm's claim is "this rule may apply to what you are writing". Only an
    agent opening it says that claim landed; a person clicking through the rule
    list in the web UI says nothing about the hint. Both are still counted in
    `pulled`, so "is this rule dead weight?" stays answerable.
    """
    from scribe.services.retrieval_telemetry import retrieval_summary

    cleanup = await _rule_events(990012, [
        (5003, "surfaced", "write_path_rule"),
        (5003, "pulled", "rest_rule"),
    ])
    try:
        ru = (await retrieval_summary(990012, days=30))["rule_usage"]
        assert ru["pulled"] == 1
        assert ru["pulled_by_human"] == 1
        assert ru["pulled_by_agent"] == 0
        # Surfaced once, opened by nobody who matters to this question.
        assert ru["pull_through"] == 0.0
    finally:
        await cleanup()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_rule_events_stay_out_of_the_note_block(_dispose_engine):
    """The separation, asserted rather than assumed.

    `usage` is what existing callers already read and compare across windows.
    If rule events leaked into it, that number would move for a reason nobody
    was told about — and the rule arm would still be invisible, because a few
    dozen rules against thousands of notes is noise on the note ratio.
    """
    from scribe.services.retrieval_telemetry import retrieval_summary

    cleanup = await _rule_events(990013, [
        (5004, "surfaced", "write_path_rule"),
        (5004, "pulled", "mcp_get_rule"),
    ])
    try:
        out = await retrieval_summary(990013, days=30)
        assert out["rule_usage"]["surfaced"] == 1
        # The note block saw none of it.
        assert out["usage"]["surfaced"] == 0
        assert out["usage"]["pulled"] == 0
        assert out["usage"]["pull_through"] is None
    finally:
        await cleanup()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_rule_usage_sees_only_its_own_users_events(_dispose_engine):
    """Same access rule as the rest of the readout — the owner filter IS the
    rule for telemetry, which is not a shared record kind."""
    from scribe.services.retrieval_telemetry import retrieval_summary

    cleanup = await _rule_events(990014, [
        (5005, "surfaced", "write_path_rule"),
        (5005, "pulled", "mcp_get_rule"),
    ])
    try:
        assert (await retrieval_summary(990015, days=30))["rule_usage"]["surfaced"] == 0
        assert (await retrieval_summary(990014, days=30))["rule_usage"]["surfaced"] == 1
    finally:
        await cleanup()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_the_preload_lands_in_ambient_and_never_in_the_ratio(_dispose_engine):
    """The split that makes the always-on set judgeable (#3473).

    Pull-through asks "was that hint any use", and only a surface that CHOSE
    what it showed can be judged by it. If the preload counted toward the
    denominator, growing 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. So the resident deliveries are counted, reported, and kept
    out of the ratio.
    """
    from scribe.services.retrieval_telemetry import retrieval_summary

    cleanup = await _rule_events(990012, [
        # One rule the arm actually chose, and opened.
        (5101, "surfaced", "write_path_rule"),
        (5101, "pulled", "mcp_get_rule"),
        # Four bulk deliveries across every shape of preload. Nobody chose any
        # of them, and none may touch the denominator.
        (5102, "surfaced", "session_start"),
        (5103, "surfaced", "list_always_on_rules"),
        (5104, "surfaced", "enter_project"),
        (5105, "surfaced", "get_milestone"),
    ])
    try:
        ru = (await retrieval_summary(990012, days=30))["rule_usage"]

        assert ru["surfaced"] == 1, "only the arm chose a rule"
        assert ru["ambient"] == 4, "the four bulk deliveries are reported, not dropped"

        # 1 agent pull over 1 RANKED surfacing. Were the ambient four folded in
        # the ratio would read 0.2 — the arm looking four times worse for
        # having a large resident set beside it.
        assert ru["pull_through"] == 1.0

        # Dead-weight detection needs both classes: a rule delivered by the
        # preload and never opened is the case that reading matters most for.
        assert ru["distinct_rules_surfaced"] == 5
        assert ru["distinct_rules_pulled"] == 1
    finally:
        await cleanup()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_ambient_alone_reports_no_ratio(_dispose_engine):
    """A brand-new install loads rules every session and may never trigger the
    arm. That must read as "no ranked surfacings yet", not as a precision of
    zero — the reading that would make a working install look broken."""
    from scribe.services.retrieval_telemetry import retrieval_summary

    cleanup = await _rule_events(990013, [
        (5201, "surfaced", "session_start"),
        (5202, "surfaced", "session_start"),
    ])
    try:
        ru = (await retrieval_summary(990013, days=30))["rule_usage"]
        assert ru["ambient"] == 2
        assert ru["surfaced"] == 0
        assert ru["pull_through"] is None
    finally:
        await cleanup()


# ── Window coverage (#3712) ────────────────────────────────────────────
#
# A counter added last week, read over a 30-day window, reports a real count
# against an imagined denominator. The result is a plausible FRACTION rather
# than an obvious zero, which is what makes it dangerous — #379 spent five
# planned steps on a defect that turned out to be a window opening before the
# recording it was measuring existed.


def test_coverage_says_nothing_rather_than_false_when_nothing_was_recorded():
    """Null, never False. "No measurement" is not "partial measurement".

    The same distinction `suppression`'s null carries (#3497): absent must not
    read as a verdict. A False here would assert the window is under-covered,
    which is a claim nobody is in a position to make.
    """
    from datetime import datetime, timezone

    from scribe.services.retrieval_telemetry import _coverage

    since = datetime(2026, 9, 1, tzinfo=timezone.utc)
    assert _coverage(None, since) == {
        "complete_from": None, "covers_window": None,
    }


def test_coverage_reads_a_start_before_the_window_as_covered():
    from datetime import datetime, timezone

    from scribe.services.retrieval_telemetry import _coverage

    since = datetime(2026, 9, 1, tzinfo=timezone.utc)
    older = datetime(2026, 8, 1, tzinfo=timezone.utc)
    newer = datetime(2026, 9, 5, tzinfo=timezone.utc)

    assert _coverage(older, since)["covers_window"] is True
    assert _coverage(newer, since)["covers_window"] is False, (
        "a counter that started inside the window covers only part of it"
    )
    assert _coverage(newer, since)["complete_from"] == newer.isoformat()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_coverage_is_per_source_because_the_table_is_older_than_its_arms(
    _dispose_engine,
):
    """THE grain question, and the reason a per-table answer is useless.

    `retrieval_logs` accumulates for months. A table-level "earliest row"
    therefore says months for every source it holds — including one added
    days ago whose counter means something quite different. The old source
    would vouch for the young one, which is exactly the reading this exists
    to prevent.
    """
    from datetime import datetime, timedelta, timezone

    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.retrieval_log import RetrievalLog
    from scribe.services.retrieval_telemetry import retrieval_summary

    UID = 990077
    now = datetime.now(timezone.utc)
    async with async_session() as s:
        # An old surface, recording since well before any window we ask for,
        # AND still recording inside it. Both rows are needed: `complete_from`
        # comes from the all-time query, but a source only gets a bucket at all
        # if it has rows in the window, so the 90-day row alone would leave
        # nothing to assert on.
        s.add(RetrievalLog(
            user_id=UID, source="auto_inject", result_count=1,
            created_at=now - timedelta(days=90),
        ))
        s.add(RetrievalLog(
            user_id=UID, source="auto_inject", result_count=1,
            created_at=now - timedelta(days=1),
        ))
        # A young arm, first written INSIDE the window below.
        s.add(RetrievalLog(
            user_id=UID, source="pre_tool_rule", result_count=1,
            created_at=now - timedelta(days=2),
        ))
        await s.commit()

    try:
        out = await retrieval_summary(UID, days=30)

        assert out["sources"]["auto_inject"]["covers_window"] is True
        assert out["sources"]["pre_tool_rule"]["covers_window"] is False, (
            "the young arm was reported as covering a 30-day window — the "
            "table's age has been allowed to vouch for one of its sources"
        )
    finally:
        async with async_session() as s:
            await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == UID))
            await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_a_surface_that_went_silent_is_not_the_same_as_one_that_never_ran(
    _dispose_engine,
):
    """#3720 — absent is how "never existed" renders, so it cannot also be how
    "stopped recording" renders.

    A surface losing its recorder is one of the failures this milestone exists
    to make visible, and dropping it from the readout is the most complete way
    to hide it. Zero here is a real measurement: the table proves the source
    was recording, and it made no calls across a window it fully covers.
    """
    from datetime import datetime, timedelta, timezone

    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.retrieval_log import RetrievalLog
    from scribe.services.retrieval_telemetry import retrieval_summary

    UID = 990078
    now = datetime.now(timezone.utc)
    async with async_session() as s:
        # Recorded once, well before the window, and never since.
        s.add(RetrievalLog(
            user_id=UID, source="auto_inject", result_count=3, top_score=0.81,
            created_at=now - timedelta(days=60),
        ))
        await s.commit()

    try:
        out = await retrieval_summary(UID, days=7)

        assert "auto_inject" in out["sources"], (
            "a source with rows in the table but none in the window was "
            "dropped from the readout — a surface that stopped recording now "
            "reads exactly like one that never existed"
        )
        quiet = out["sources"]["auto_inject"]
        assert quiet["calls"] == 0
        # The window IS covered; what was observed across it is nothing.
        assert quiet["covers_window"] is True
        # ...but nothing was sampled, so no distribution may be claimed. A
        # zeroed score would assert a measurement, which is #3311's mistake.
        assert quiet["top_score"] == {
            "p10": None, "p50": None, "p90": None, "min": None, "max": None,
        }
        assert quiet["suppression"] is None
        assert quiet["avg_result_count"] is None
    finally:
        async with async_session() as s:
            await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == UID))
            await s.commit()


@pytest.mark.integration
@pytest.mark.asyncio
async def test_a_section_is_complete_only_from_its_latest_contributor(
    _dispose_engine,
):
    """A sum is complete once EVERY contributor was being written — so the
    section takes the LATEST first-row, not the earliest.

    Taking the earliest would be worse than reporting nothing: it would pick
    the oldest source in the table and use it to certify a total that a
    newer source is still only partly contributing to. That is the original
    error in miniature.
    """
    from datetime import datetime, timedelta, timezone

    from sqlalchemy import delete

    from scribe.models import async_session
    from scribe.models.rule_usage import RuleUsageEvent
    from scribe.services.retrieval_telemetry import retrieval_summary

    UID = 990078
    now = datetime.now(timezone.utc)
    old = now - timedelta(days=90)
    young = now - timedelta(days=2)
    async with async_session() as s:
        s.add_all([
            RuleUsageEvent(
                user_id=UID, rule_id=1, event="surfaced",
                source="list_always_on_rules", created_at=old,
            ),
            RuleUsageEvent(
                user_id=UID, rule_id=2, event="surfaced",
                source="pre_tool_rule", created_at=young,
            ),
        ])
        await s.commit()

    try:
        out = await retrieval_summary(UID, days=30)
        ru = out["rule_usage"]

        assert ru["complete_from"] == young.isoformat(), (
            "the section reported completeness from its OLDEST source; a "
            "total is only as complete as its newest contributor"
        )
        assert ru["covers_window"] is False
    finally:
        async with async_session() as s:
            await s.execute(delete(RuleUsageEvent).where(RuleUsageEvent.user_id == UID))
            await s.commit()


# ─── the bar can only be judged from what it rejected (#3670) ────────────────
#
# `cleared_threshold` was the number the docstring told a reader to look at
# first. It was `calls - zero_result_calls` under another name: the search
# applies the bar before returning, so every returned result cleared it by
# construction and a call with nothing has no score to compare.
# `zero_result_calls + cleared_threshold == calls` held on all nineteen
# source/window readings ever taken — no near-misses, no exceptions.
#
# What replaced it cannot go the same way, and the reason is structural rather
# than careful naming: `near_misses` is measured on the calls the bar TURNED
# AWAY, using a score the bar never saw. No arrangement of `calls`,
# `zero_result_calls` and `result_count` derives it.


def test_a_call_that_returned_nothing_still_records_what_it_nearly_showed():
    """The whole point, at the payload grain.

    This is the row a threshold is tuned from and the one that used to carry no
    score at all: `top_score` and `min_score` are both null here, correctly, and
    a reader was left unable to tell a bar rejecting 0.71s from one rejecting
    0.30s. Both render as a zero-result call.
    """
    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,
    )
    assert p["result_count"] == 0
    assert p["top_score"] is None, "nothing was shown, so nothing has a top score"
    assert p["best_available_score"] == 0.7104, (
        "the losing score was discarded — the only figure that survives a call "
        "returning nothing, and the only one a bar can be judged from"
    )


def test_a_caller_that_did_not_measure_the_near_miss_stores_null():
    """Null, never 0.0. A zero here reads as "the corpus held nothing remotely
    relevant" — a claim about the corpus invented out of a caller's silence,
    which is #3311's substitution in a new field."""
    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


def test_the_readout_reports_unmeasured_near_misses_as_none():
    """`_bucket`'s half of the same discipline, and the reason it is a block
    rather than three loose keys: old rows predate the column, so a window can
    legitimately contain declines nobody measured."""
    from scribe.services.retrieval_telemetry import _bucket

    # calls, zero, p10, p50, p90, min, max, avg_n, dur,
    # measured, supp_calls, supp_zero, miss_calls, miss_p50, miss_p90, miss_max
    none_measured = _bucket([326, 114, 0.6, 0.68, 0.77, 0.55, 0.85, 1.7, 130.9,
                             0, 0, 0, 0, None, None, None])
    assert none_measured["near_misses"] is None

    measured = _bucket([326, 114, 0.6, 0.68, 0.77, 0.55, 0.85, 1.7, 130.9,
                        0, 0, 0, 114, 0.61, 0.7104, 0.7189])
    assert measured["near_misses"] == {
        "measured_calls": 114,
        "p50": 0.61,
        "p90": 0.7104,
        "max": 0.7189,
    }


def test_the_readout_carries_no_field_derivable_from_its_neighbours():
    """The guard that would have caught #3670 on the day it shipped.

    `cleared_threshold` survived because it had its own name and its own
    docstring paragraph, and nobody added the two numbers beside it. This
    asserts the identity that held on every reading ever taken — and if a
    future field reintroduces it under a new name, the sum below is where it
    shows up.
    """
    from scribe.services.retrieval_telemetry import _bucket

    b = _bucket([326, 114, 0.6, 0.68, 0.77, 0.55, 0.85, 1.7, 130.9,
                 0, 0, 0, 114, 0.61, 0.71, 0.72])
    derivable = {
        k for k, v in b.items()
        if isinstance(v, int) and not isinstance(v, bool)
        and k not in ("calls", "zero_result_calls")
        and v == b["calls"] - b["zero_result_calls"]
    }
    assert not derivable, (
        f"{sorted(derivable)} equals calls - zero_result_calls on this row. "
        f"That is how `cleared_threshold` read for its whole life (#3670): a "
        f"figure presented as an independent measurement that a reader can "
        f"compute from the two numbers next to it. Either it is a tautology, "
        f"or this fixture happens to make it look like one — check which "
        f"before adding an exemption."
    )


@pytest.mark.integration
@pytest.mark.asyncio
async def test_the_near_miss_distribution_is_a_query_postgres_accepts(_dispose_engine):
    """Integration, and NOT belt-and-braces on the unit tests above.

    `near_misses` is a `percentile_cont(...) WITHIN GROUP` over a CASE
    expression, inside the same grouped aggregate that already carries four
    other CASEs. That is within one step of the shape that produced #2663 — a
    query the database rejected, swallowed by this module's broad `except`, so
    every counter read zero in production while the writes landed fine and the
    mocked tests passed. Only a real Postgres can say this parses, and if it
    does not, the symptom is silence rather than an error.

    The numbers are chosen so a bar at 0.72 is visibly the wrong bar: three
    declines at 0.70, 0.71 and 0.7189, none of which a reader could see before.
    """
    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 = 990079
    for best in (0.70, 0.71, 0.7189):
        await _insert_retrieval_log(_build_payload(
            user_id=UID, source="pre_tool_rule", query="git push", threshold=0.72,
            limit=1, project_id=None, is_task=None, results=[],
            duration_ms=4.0, best_available=best,
        ))
    # A call that DID show something. Its best-available equals its top score,
    # so including it would drag the distribution toward the scores the bar
    # already accepts — the population has to be the declines alone.
    await _insert_retrieval_log(_build_payload(
        user_id=UID, source="pre_tool_rule", query="curl", threshold=0.72,
        limit=1, project_id=None, is_task=None,
        results=[(0.88, _note(7))], duration_ms=4.0, best_available=0.88,
    ))
    # An unmeasured decline, standing in for every row written before #3670.
    await _insert_retrieval_log(_build_payload(
        user_id=UID, source="pre_tool_rule", query="ls", threshold=0.72,
        limit=1, project_id=None, is_task=None, results=[], duration_ms=4.0,
    ))

    try:
        out = await retrieval_summary(UID, days=30)
        assert out["read_failed"] is False, (
            "the aggregate did not execute — a rejected query here reads as "
            "zeros everywhere, which is #2663 exactly"
        )
        src = out["sources"]["pre_tool_rule"]
        assert src["calls"] == 5
        assert src["zero_result_calls"] == 4

        nm = src["near_misses"]
        assert nm is not None, "the near-miss block did not survive the query"
        assert nm["measured_calls"] == 3, (
            "the population is declines that RECORDED a score: three measured, "
            "one unmeasured (excluded, not counted as a scoreless decline), and "
            "one call that showed something (excluded — its best-available is "
            "just its top score and says nothing about the bar)"
        )
        assert nm["max"] == pytest.approx(0.7189, abs=1e-4), (
            "the closest thing the bar turned away — 0.7189 against a 0.72 "
            "threshold, which is the reading the whole field exists to give"
        )
        assert nm["max"] < 0.72, "a near miss that cleared the bar is not a miss"
        assert 0.70 <= nm["p50"] <= 0.7189
    finally:
        async with async_session() as s:
            await s.execute(delete(RetrievalLog).where(RetrievalLog.user_id == UID))
            await s.commit()