feat(telemetry): every counter says when it started being recorded (#3712)

A window that opens before a counter existed reports that counter as though
it had been measured throughout. The reader cannot tell "zero because
nothing happened" from "zero because nobody was counting yet", and — worse
— cannot tell a partial count from a complete one. That middle case yields
a plausible FRACTION rather than an obvious zero, which is what makes it
dangerous.

It is not hypothetical. A 7-day window opened while the ranked rule
surfacing recorders were four days old produced an apparent 64% write loss,
which survived a code review, four ruled-out alternative causes and a
five-step milestone before an identity check falsified it in one read.

Every counter block now carries `complete_from` and `covers_window`.

THE GRAIN IS THE SOURCE. retrieval_logs accumulates for months, so a
per-table earliest row says months for every source it holds — including an
arm added days ago whose counter means something else entirely. The old
source would vouch for the young one, which is the exact reading this
prevents.

A SECTION TAKES ITS LATEST CONTRIBUTOR, NOT ITS EARLIEST. A figure summing
several sources is complete only once every one of them was being written,
so "*" is a max. Using min would reproduce the original error in miniature.

`covers_window` is null, never false, when nothing was ever recorded: "no
measurement" is not "partial measurement" — the null convention #3497
established for `suppression`, one level up.

Also corrects a stale claim in the tool docstring: it still taught readers
that write_path_rule "has never once declined to fire" (#3311). That was
the arm writing its retrieval_logs row only on calls that found something;
#3497 fixed it, and the arm declines the large majority of its calls.

_complete_from takes the caller's session rather than opening its own,
departing from the services canon (#2860) because it runs inside an
existing block; to be recorded against the ledger once it ingests.

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

src/scribe/services/retrieval_telemetry.py

@@ -246,6 +246,57 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
     no readout. It does distinguish "no rows" from "the read failed", because
     #2663 is exactly the bug where those two looked identical for weeks.
     """
+async def _complete_from(session, model, user_id) -> dict[str, Any]:
+    """When each source in `model` started being recorded, and the instant the
+    WHOLE table is complete from. Returns {source: earliest_row, "*": latest}.
+
+    THE GRAIN IS THE SOURCE, and that is the whole point. `retrieval_logs` has
+    rows going back months, so a table-level "earliest row" says months and
+    tells a reader their window is fully covered — while a source added last
+    week has a week of rows and a counter that silently means something else.
+    Per-source is the only grain at which partial coverage is visible.
+
+    THE AGGREGATE USES THE LATEST, NOT THE EARLIEST. A number that sums several
+    sources is complete only once EVERY contributor was recording, so "*" is a
+    max over the sources, not a min. Taking the min here would reproduce the
+    exact reading this exists to prevent: the oldest source vouching for the
+    youngest.
+
+    All-time, deliberately unfiltered by the window — a query bounded by
+    `since` can only ever report something at or after `since`, which answers
+    nothing.
+    """
+    rows = (
+        await session.execute(
+            select(model.source, func.min(model.created_at))
+            .where(model.user_id == user_id)
+            .group_by(model.source)
+        )
+    ).all()
+    out: dict[str, Any] = {src: ts for src, ts in rows if ts is not None}
+    stamps = list(out.values())
+    out["*"] = max(stamps) if stamps else None
+    return out
+
+
+def _coverage(complete_from, since) -> dict:
+    """The two keys every counter block carries, from one timestamp.
+
+    `covers_window` is None — never False — when nothing was ever recorded.
+    "No rows at all" is not "partial coverage", it is no measurement, and the
+    null convention #3497 established for `suppression` holds here for the
+    same reason: absent must not read as a verdict.
+    """
+    return {
+        # iso() already returns None for an unset value (#2845) — the guard
+        # belongs on covers_window, which is a verdict, not a serialisation.
+        "complete_from": iso(complete_from),
+        "covers_window": (
+            None if complete_from is None else complete_from <= since
+        ),
+    }
+
+
     since = datetime.now(timezone.utc) - timedelta(days=max(1, int(days)))
     out: dict = {
         "window_days": int(days),
@@ -283,6 +334,9 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
     by_source_rows = None
     rule_rows = None
     distinct_rules_surfaced = distinct_rules_pulled = 0
+    # None means the coverage read did not happen — distinct from a table with
+    # no rows, which is {"*": None}. Same reason `read_failed` exists.
+    note_complete = rule_complete = None
 
     try:
         async with async_session() as session:
@@ -311,8 +365,15 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
                     .group_by(RetrievalLog.source)
                 )
             ).all()
+            log_complete = await _complete_from(session, RetrievalLog, user_id)
             for row in rows:
-                out["sources"][row[0]] = _bucket(list(row[1:]))
+                source = row[0]
+                bucket = _bucket(list(row[1:]))
+                # Per SOURCE, not per table: retrieval_logs goes back months
+                # while any individual arm may be days old, and the table's
+                # age would vouch for an arm that has barely started.
+                bucket.update(_coverage(log_complete.get(source), since))
+                out["sources"][source] = bucket
 
             # The corpus side, at its own grain. `ambient` mirrors
             # note_usage.usage_for_notes: an ambient surfacing was not a scored
@@ -339,6 +400,7 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
                     .group_by(NoteUsageEvent.event, NoteUsageEvent.source)
                 )
             ).all()
+            note_complete = await _complete_from(session, NoteUsageEvent, user_id)
 
             # Distinct-note counts need their OWN queries, and this is not
             # fussiness: count(distinct note_id) per (event, source) group
@@ -466,6 +528,9 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
                         .group_by(RuleUsageEvent.event, RuleUsageEvent.source)
                     )
                 ).all()
+                rule_complete = await _complete_from(
+                    session, RuleUsageEvent, user_id,
+                )
                 # 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.
@@ -575,6 +640,10 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
             }
         usage["by_source"] = by_source
 
+    # The SECTION's coverage, from the latest source to start recording — a
+    # figure that sums several sources is complete only once every one of them
+    # was being written. `_complete_from` computes that as "*".
+    usage.update(_coverage((note_complete or {}).get("*"), since))
     out["usage"] = usage
 
     # ── Rules, deliberately a SEPARATE block ────────────────────────────
@@ -652,6 +721,7 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
         round(rule_usage["pulled_by_agent"] / rule_usage["surfaced"], 4)
         if rule_usage["surfaced"] else None
     )
+    rule_usage.update(_coverage((rule_complete or {}).get("*"), since))
     out["rule_usage"] = rule_usage
 
     return out