docs/api-keys-and-mcp.md

@@ -43,8 +43,10 @@ client straight to the URL with a Bearer token.
 
 Authenticate with an API key generated from **Settings → API Keys** (see above),
 sent as `Authorization: Bearer fmcp_<key>`. A `read`-scoped key may call only the
-read tools (`get_*`, `list_*`, `search`, `enter_project`); any write/delete tool
-is rejected with `403`. A `write`-scoped key may call everything.
+read tools (`get_*`, `list_*`, `search`, `enter_project`, `retrieval_telemetry`);
+any write/delete tool is rejected with `403`. The allow-list is explicit rather
+than derived from the name — see `_READ_ONLY_TOOLS`, which is why the two reads
+without a read-shaped name are spelled out here. A `write`-scoped key may call everything.
 
 ### Claude Code (Project-scoped)
 
@@ -85,7 +87,7 @@ table here. The tools are grouped by family:
 | Notes | `create_note`, `get_note`, `update_note`, `delete_note`, `list_notes` | Free-form knowledge |
 | Tasks | `create_task`, `update_task`, `add_task_log`, `start_planning` | Actionable work + plans |
 | Projects / Milestones | `enter_project`, `get_project`, `create_milestone`, … | Containers and outcomes |
-| Search / Recall | `search`, `get_recent`, `list_tags` | Semantic + structured recall |
+| Search / Recall | `search`, `get_recent`, `list_tags`, `retrieval_telemetry` | Semantic + structured recall, and the readout its thresholds are tuned from |
 | Systems | `create_system`, `list_systems`, `list_system_records` | Reusable per-project subsystems/areas |
 | Rulebooks | `list_always_on_rules`, `list_rules`, `create_rule`, `create_project_rule`, `subscribe_project_to_rulebook`, … | Engineering/workflow rules |
 | Processes | `list_processes`, `get_process`, `create_process` | Saved prompts/workflows |

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe system-of-record for Claude Code: MCP tools over your notes/tasks/projects/rules, a session-start push channel that surfaces your always-on rules + active-project context, process-skills (writing-plans, systematic-debugging, verification, brainstorming, reusing-code), and your saved Scribe Processes auto-surfaced as skills (/scribe:sync). Replaces superpowers + file-memory with one app-backed plugin.",
-  "version": "0.1.45",
+  "version": "0.1.46",
   "author": { "name": "Bryan Van Deusen" },
   "mcpServers": {
     "scribe": {

plugin/hooks/scribe_after_write.sh

@@ -104,10 +104,7 @@ while IFS=$'\t' read -r path sha; do
 done <<< "$current"
 [ -n "$changed" ] || exit 0
 
-url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_API_ENDPOINT:-}}
-token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_API_TOKEN:-}}
-case "$url"   in *'${'*) url="" ;;  esac
-case "$token" in *'${'*) token="" ;; esac
+scribe_config || :   # sets url/token; the call below is guarded on them
 repo=$(git -C "$repo_root" remote get-url origin 2>/dev/null || true)
 repo_q=""
 if [ -n "$repo" ]; then

plugin/hooks/scribe_autoinject.sh

@@ -23,6 +23,9 @@
 # note is injected at most once per session. Passed back as exclude_ids.
 set -uo pipefail
 
+# shellcheck source=plugin/hooks/scribe_defs.sh
+. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
+
 command -v jq   >/dev/null 2>&1 || exit 0
 command -v curl >/dev/null 2>&1 || exit 0
 
@@ -35,13 +38,8 @@ event_cwd=$(printf '%s' "$event" | jq -r '.cwd // empty' 2>/dev/null) || event_c
 # Nothing to retrieve against.
 [ -n "$prompt" ] || exit 0
 
-url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_API_ENDPOINT:-}}
-token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_API_TOKEN:-}}
-# Guard against an unexpanded ${...} placeholder arriving as a literal.
-case "$url"   in *'${'*) url="" ;;  esac
-case "$token" in *'${'*) token="" ;; esac
 # Unconfigured install → silent (auto-inject is pure enrichment).
-[ -n "$url" ] && [ -n "$token" ] || exit 0
+scribe_config || exit 0
 
 # Cap the query length — a giant prompt makes a giant URL for no extra signal.
 # `head -c`, not `cut -c1-2000`: cut is line-oriented and caps EACH LINE, so a

plugin/hooks/scribe_defs.sh

@@ -1,6 +1,6 @@
 #!/usr/bin/env bash
 # shellcheck shell=bash
-# Scribe plugin — the pieces the two write-path hooks share (#2901).
+# Scribe plugin — the pieces the hooks share (#2901, #2278).
 #
 # scribe_prior_art.sh fires BEFORE a Write/Edit tool call; scribe_after_write.sh
 # fires AFTER a Bash tool call and diffs the working tree, so code written by
@@ -14,6 +14,8 @@
 #   scribe_unreached STATE SID SECS REL  the "Scribe didn't answer" line, once
 #                              per outage (#2932) — or nothing, if said lately
 #   scribe_reached STATE SID   the server answered: the next outage speaks again
+#   scribe_config              sets `url` + `token` from the env, returns 0
+#                              only if BOTH are usable (#2278)
 #
 # Sourced, not executed: `. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"`.
 
@@ -128,6 +130,34 @@ scribe_local_dups() {
 # the time it last spoke; within ten minutes of that it stays quiet, and a
 # successful call clears it so the next outage announces itself afresh.
 # Unconfigured installs never reach this: no URL/token means no call was owed.
+# Where every hook gets its endpoint and credential. Four lines, and each of
+# the five hooks carried its own copy until #2278 — which is exactly the
+# missing-sibling shape: the `${...}` guard below is a correctness detail a
+# sixth hook would have forgotten, and nothing would have failed loudly.
+#
+# Sets `url` and `token` as globals rather than echoing them: a token must not
+# pass through a subshell's output, where it could land in a log or an `xtrace`
+# line. Returns 0 only when both are usable, so a caller can either bail
+# (`scribe_config || exit 0`) or carry on degraded — the session-context hook
+# still owes its static floor when Scribe is unconfigured.
+# Declared here, not just assigned inside the function: `scribe_defs.sh` owns
+# these two names, and a sourcing hook should have them defined the moment it
+# sources — before any code path that might reference them. It also lets
+# the linter see the assignment, which it cannot follow into a function in
+# another file without -x (SC2154).
+url=""
+token=""
+
+scribe_config() {
+  url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_API_ENDPOINT:-}}
+  token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_API_TOKEN:-}}
+  # An unexpanded `${...}` placeholder arriving as a literal would be sent as a
+  # garbage Bearer token and 401. Treat it as unset.
+  case "$url"   in *'${'*) url="" ;;  esac
+  case "$token" in *'${'*) token="" ;; esac
+  [ -n "$url" ] && [ -n "$token" ]
+}
+
 _SCRIBE_UNREACHED_QUIET=600
 
 scribe_unreached() {

plugin/hooks/scribe_prior_art.sh

@@ -121,11 +121,7 @@ if [ -n "$shapes" ]; then
   [ -n "$enc" ] && shapes_q="&shapes=${enc}"
 fi
 
-url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_API_ENDPOINT:-}}
-token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_API_TOKEN:-}}
-# Guard against an unexpanded ${...} placeholder arriving as a literal.
-case "$url"   in *'${'*) url="" ;;  esac
-case "$token" in *'${'*) token="" ;; esac
+scribe_config || :   # sets url/token; unconfigured is handled just below
 # Unconfigured install → the recorded-prior-art arms are skipped, but the local
 # arm above already ran and may have something to say.
 if [ -z "$url" ] || [ -z "$token" ]; then

plugin/hooks/scribe_session_context.sh

@@ -39,6 +39,9 @@
 # allowed to fail quietly; see the #2198 comment at the status block below.
 set -uo pipefail
 
+# shellcheck source=plugin/hooks/scribe_defs.sh
+. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
+
 command -v jq >/dev/null 2>&1 || exit 0   # needed to emit the JSON envelope safely
 
 # `CDPATH= cd` is deliberate, not a typo'd assignment: it runs this one `cd`
@@ -87,13 +90,9 @@ if [ -f "$manifest" ]; then
 fi
 
 # --- Tier 2: dynamic rules + active-project context (best-effort) ---
-url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_API_ENDPOINT:-}}
-token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_API_TOKEN:-}}
-
-# Guard against an unexpanded `${...}` placeholder reaching us as a literal — it
-# would otherwise be sent as a garbage Bearer token and 401. Treat as unset.
-case "$url"   in *'${'*) url="" ;;  esac
-case "$token" in *'${'*) token="" ;; esac
+# Unconfigured is NOT a failure here: tier 1's static floor is still owed,
+# so this records the answer rather than acting on it.
+scribe_config || :
 
 dyn=""
 status=""

plugin/hooks/scribe_sync_processes.sh

@@ -23,15 +23,13 @@
 # #2198), with SCRIBE_URL / SCRIBE_TOKEN as the override.
 set -uo pipefail
 
+# shellcheck source=plugin/hooks/scribe_defs.sh
+. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
+
 command -v jq >/dev/null 2>&1 || exit 0
 command -v curl >/dev/null 2>&1 || exit 0
 
-url=${SCRIBE_URL:-${CLAUDE_PLUGIN_OPTION_API_ENDPOINT:-}}
-token=${SCRIBE_TOKEN:-${CLAUDE_PLUGIN_OPTION_API_TOKEN:-}}
-# Guard against an unexpanded `${...}` placeholder arriving as a literal.
-case "$url"   in *'${'*) url="" ;; esac
-case "$token" in *'${'*) token="" ;; esac
-[ -n "$url" ] && [ -n "$token" ] || exit 0
+scribe_config || exit 0
 
 body=$(curl -fsS --max-time 8 \
   -H "Authorization: Bearer ${token}" \

scripts/check_plugin.py

@@ -159,7 +159,12 @@ def check_shellcheck() -> None:
         return
     for script in hook_scripts():
         proc = subprocess.run(
-            [exe, "--severity=warning", "--shell=bash", str(script)],
+            # -x FOLLOWS `# shellcheck source=` directives into the sourced
+            # file. Without it the shared helpers in scribe_defs.sh are
+            # invisible, so every variable they set reads as unassigned
+            # (SC2154) and every bug inside them goes unlinted at the call
+            # site — which is the opposite of what sharing them was for.
+            [exe, "--severity=warning", "--shell=bash", "-x", str(script)],
             capture_output=True, text=True,
         )
         rel = script.relative_to(ROOT)

src/scribe/mcp/server.py

@@ -112,6 +112,11 @@ _READ_ONLY_TOOLS = frozenset({
     # The shape ledger's todo query (#2789). Reads only — classify_shapes is
     # the write, and it is deliberately NOT here.
     "list_shapes", "shape_history",
+    # The retrieval telemetry readout (#2975). Aggregates two log tables and
+    # writes nothing. Listed explicitly because its name carries no read
+    # prefix, so the completeness test below cannot derive it — the same
+    # reason `enter_project` is spelled out above.
+    "retrieval_telemetry",
 })
 
 # Read-SHAPED tools that must NOT be reachable with a read key — a getter that

src/scribe/mcp/tools/processes.py

@@ -15,13 +15,17 @@ from scribe.services import trash as trash_svc
 from scribe.services.note_usage import record_pulled
 
 
-async def list_processes(q: str = "", tag: str = "", limit: int = 50) -> dict:
+async def list_processes(
+    q: str = "", tag: str = "", limit: int = 50, offset: int = 0,
+) -> dict:
     """List stored processes (reusable saved prompts).
 
     Args:
         q: Free-text search across title + body (optional).
         tag: Filter to a single tag (optional).
         limit: Max results (1-100).
+        offset: Skip this many before returning — page past the cap.
+            `total` is the unpaged count, so it says whether more remains.
 
     Returns {"processes": [{id, title, tags, preview}], "total": int}. An entry
     marked `shared: true` with an `owner` is another person's procedure — treat
@@ -34,7 +38,8 @@ async def list_processes(q: str = "", tag: str = "", limit: int = 50) -> dict:
     uid = current_user_id()
     items, total = await knowledge_svc.query_knowledge(
         user_id=uid, note_type="process", tags=[tag] if tag else [],
-        sort="modified", q=q or None, limit=max(1, min(limit, 100)), offset=0,
+        sort="modified", q=q or None, limit=max(1, min(limit, 100)),
+        offset=max(0, offset),
     )
     labelled = await access_svc.label_shared_items(uid, items)
     procs = [{"id": it["id"], "title": it["title"], "tags": it.get("tags", []),

src/scribe/mcp/tools/search.py

@@ -12,7 +12,7 @@ import time
 from scribe.mcp._context import current_user_id
 from scribe.services.access import owner_names_for
 from scribe.services.embeddings import DEFAULT_SIMILARITY_THRESHOLD, semantic_search_notes
-from scribe.services.retrieval_telemetry import record_retrieval
+from scribe.services.retrieval_telemetry import record_retrieval, retrieval_summary
 
 
 async def search(
@@ -95,5 +95,52 @@ async def search(
     }
 
 
+async def retrieval_telemetry(days: int = 30) -> dict:
+    """What the retrieval telemetry says about YOUR surfaces, over a window.
+
+    The read half of the loop the ranker's thresholds are meant to be tuned
+    from (#2975). Reach for it before changing a similarity threshold, a top-k,
+    or deciding whether a reranker is worth building — the alternative is
+    hand-probing the live instance, which is how the last such decision had to
+    be made.
+
+    Two readouts, from the two tables built for them:
+
+    `sources` — per retrieval surface (`auto_inject`, `write_path`,
+    `mcp_search`, …), from `retrieval_logs`: `calls`, `zero_result_calls`,
+    `cleared_threshold` (how often the best hit beat the threshold in force for
+    that call), the `top_score` spread (p10/p50/p90/min/max), `avg_result_count`
+    and `p90_duration_ms`. THE number to read first is `cleared_threshold`
+    against `calls`, with the spread beside it: a surface that clears its bar
+    on nearly every call is either well-tuned or too loose, and p10 says which.
+
+    `usage` — from `note_usage_events`, at the per-note grain
+    `retrieval_logs` cannot be indexed at: `surfaced` (ranked surfacings — a
+    scored surface CHOSE the record), `ambient` (the rest), `pulled` split into
+    `pulled_by_agent` / `pulled_by_human`, the distinct-note counts, and
+    `pull_through`. That ratio is the corpus-side precision signal: records
+    surfaced often and opened never are dead weight competing for the injection
+    budget every turn.
+
+    `pull_through` is AGENT pulls over RANKED surfacings, and both halves of
+    that matter. "Is this record dead weight?" is answered by any pull; "was
+    that injected line useful?" — the question a threshold or a reranker is
+    tuned against — only by a pull the agent made. Aggregating across the
+    mcp_/rest_ prefix would silently answer the wrong one.
+
+    Scoped to your own telemetry — a retrieval log records what your agent
+    asked for, query text included, and is not a shared record kind.
+
+    `read_failed: true` means the query itself failed — deliberately distinct
+    from an empty window, because those two looked identical for weeks once
+    (#2663) and every counter silently read zero.
+
+    Args:
+        days: window size, default 30. Clamped to at least 1.
+    """
+    return await retrieval_summary(current_user_id(), days=days)
+
+
 def register(mcp) -> None:
     mcp.tool(name="search")(search)
+    mcp.tool(name="retrieval_telemetry")(retrieval_telemetry)

src/scribe/mcp/tools/snippets.py

@@ -20,7 +20,8 @@ from scribe.services import systems as systems_svc
 
 
 async def list_snippets(
-    q: str = "", tag: str = "", limit: int = 50, project_id: int = 0,
+    q: str = "", tag: str = "", limit: int = 50, offset: int = 0,
+    project_id: int = 0,
     repo: str = "", path: str = "", symbol: str = "", verification: str = "",
 ) -> dict:
     """List recorded snippets — the project's pattern library.
@@ -41,6 +42,9 @@ async def list_snippets(
             well as wording, so describe what you need the code to DO.
         tag: Filter to a single tag, e.g. a language like "python" (optional).
         limit: Max results (1-100).
+        offset: Skip this many before returning — page through a corpus
+            larger than one call. `total` is the unpaged count, so
+            offset+limit against it says whether more remains.
         project_id: Narrow to one project. 0 (default) searches every project —
             usually what you want, since a helper you need here may well have
             been written somewhere else.
@@ -81,6 +85,7 @@ async def list_snippets(
     uid = current_user_id()
     items, total = await snippets_svc.list_snippets(
         uid, q=q or None, tag=tag, limit=max(1, min(limit, 100)),
+        offset=max(0, offset),
         project_id=project_id or None,
         repo=repo, path=path, symbol=symbol, verification=verification,
     )
@@ -207,6 +212,13 @@ async def get_snippet(snippet_id: int) -> dict:
     the source moved on — trust the location over the cached body and
     consider verify_snippet after you look.
 
+    A record kept VERBATIM is confirmed by containment. A deliberately
+    ANNOTATED one — commentary the source does not carry — cannot be, so it
+    reads "current" on the authority of a standing `ok` verdict stamped at
+    the very commit just fetched (#2782); `verification` in the same payload
+    shows that basis. Edit the record, or let the file move past that commit,
+    and it reads "diverged" again until someone re-runs verify_snippet.
+
     When the shape ledger has judgments against this snippet, the response
     carries `instances` (shapes classified as conforming to it — the
     structured consumer map) and/or `variants` (named departures, each with

src/scribe/services/retrieval_telemetry.py

@@ -18,8 +18,14 @@ from __future__ import annotations
 import asyncio
 import logging
 
+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.retrieval_log import RetrievalLog
 
 logger = logging.getLogger(__name__)
@@ -135,3 +141,205 @@ def record_retrieval(
         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 = 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),
+        "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.
+
+    Two aggregates side by side, each read from the table built for it — NOT a
+    join. `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.
+
+    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": {},
+        "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)
+
+    def pct(p: float):
+        return func.percentile_cont(p).within_group(RetrievalLog.top_score.asc())
+
+    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()
+                        ),
+                    )
+                    .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()
+    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
+    )
+    out["usage"] = usage
+    return out

src/scribe/services/snippets.py

@@ -987,6 +987,46 @@ async def _refresh_provenance(note, commit_sha: str) -> None:
     await notes_svc.update_note(note.user_id, note.id, data=data)
 
 
+def _verdict_still_vouches(note, fields: dict, fetched_commit_sha: str) -> bool:
+    """Does a standing `ok` verdict still speak for this body, at this commit?
+
+    Containment (cached code ∈ fetched file) is the fast path, and it is right
+    for a record kept verbatim. It is WRONG for a deliberately annotated one
+    (#2782): a record whose job is to say why the shape is what it is carries
+    commentary the source does not, so containment fails forever and the record
+    reads `diverged` on every pull. That turns the one honest drift signal into
+    a permanent false positive — and annotation is a sanctioned record style,
+    so this is two deliberate designs colliding, not a malformed record.
+
+    The escape hatch is the verdict itself. `verify_snippet` is precisely where
+    a human or agent already judged this body a faithful rendering of that
+    source, and `verification.commit_sha` records the repo commit they judged
+    it at — a field whose own docstring (#2688) anticipated this use: "makes
+    'the REPO moved on since the check' computable, once the forge integration
+    can compare it against the current head." This is that comparison.
+
+    All four conditions, and none is optional:
+      - the verdict says `ok`;
+      - it has not EXPIRED — `verification_view` recomputes `code_sha` against
+        the record's current body, so editing the record retires the verdict;
+      - it was not INVALIDATED by a push touching the location (#2691);
+      - the file we just fetched is at the very commit the verdict was stamped
+        at. Any later commit means nobody has judged what is there now.
+
+    The last one is what keeps this honest: it vouches for a body against ONE
+    known commit, never against whatever the source has become since. The
+    moment the file moves, containment resumes as the authority and the record
+    reads `diverged` until someone re-verifies — which is the correct outcome,
+    because at that point nobody has looked.
+    """
+    if not fetched_commit_sha:
+        return False
+    view = verification_view(note, fields)
+    if view.get("status") != VERIFY_OK or view.get("needs_attention"):
+        return False
+    return view.get("commit_sha") == fetched_commit_sha
+
+
 async def attach_live_body(note, data: dict) -> None:
     """Decorate a PULL response with forge-checked freshness (#2690).
 
@@ -1115,6 +1155,14 @@ async def attach_live_body(note, data: dict) -> None:
                     _refresh_provenance(note, fetched.commit_sha),
                     site="pull provenance-refresh",
                 )
+    elif _verdict_still_vouches(note, fields, fetched.commit_sha or ""):
+        # Containment failed, but an unexpired `ok` verdict stamped at exactly
+        # this commit already judged this body a faithful rendering of it —
+        # the annotated-record case (#2782). Trust the judgment over the
+        # substring test; `data["verification"]` travels in the same payload,
+        # so a reader can see the basis rather than take "current" on faith.
+        data["body_source"] = "forge"
+        data["body_freshness"] = "current"
     else:
         data["body_source"] = "cache"
         data["body_freshness"] = "diverged"

tests/test_mcp_list_family.py

@@ -0,0 +1,114 @@
+"""Cross-family contracts for the `list_*` MCP tools (#2278, shape 4).
+
+Per-tool tests cover what each list tool does. Nothing covered what the FAMILY
+owes its callers, which is where the missing-sibling shape hides: a capability
+added to one member and not its neighbour changes no return value, so no
+behavioural test can see it. Source inspection can.
+
+WHAT THIS DELIBERATELY DOES NOT ASSERT. The 19 `list_*` tools are genuinely
+heterogeneous — 8 take `project_id`, 6 take `limit`, and six take no arguments
+at all (`list_projects`, `list_trash`, `list_rulebooks`, `list_design_systems`,
+`list_repo_bindings`, `list_starter_role_groups`). Requiring a common parameter
+across them would be inventing a convention the API does not have, which the
+DRY process's over-DRY guard (§5) warns against by name: a wrong abstraction is
+worse than the duplication. So this file asserts ONE contract, the one that is
+a real promise rather than a shape coincidence.
+
+THE CONTRACT: a `limit` without an `offset` is a truncation with no
+continuation. The caller is told there are 250 results and handed 50, with no
+way to ask for the rest. Both tools that had this were capped over a service
+that already accepted an offset — `snippets_svc.list_snippets(offset=0)` was
+simply not exposed, and `list_processes` passed a hardcoded `offset=0` into
+`query_knowledge`. The capability existed one layer down in both cases; only
+the door was missing.
+
+As in `test_mcp_auth`, the CANDIDATES are derived and the DECISION is explicit.
+Deriving the exemption too would make the contract follow a naming convention,
+so any future `list_*` could opt itself out by accident.
+"""
+import ast
+import pathlib
+
+TOOLS_DIR = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe" / "mcp" / "tools"
+
+# `limit` here caps a RANKED top-N, not a page into a corpus, so there is no
+# "rest" to ask for — the 51st most-used tag is not what the caller wanted and
+# an offset into that ordering answers no question. Anything added here needs a
+# reason of that kind, not "it isn't paged yet".
+_DELIBERATELY_UNPAGED = {
+    "list_tags",        # most-used tags by count, over a bounded vocabulary
+}
+
+
+def _list_tools() -> dict[str, set[str]]:
+    """{tool name: parameter names} for every `list_*` in the tools package."""
+    out: dict[str, set[str]] = {}
+    for path in sorted(TOOLS_DIR.glob("*.py")):
+        if path.name == "__init__.py":
+            continue
+        for node in ast.parse(path.read_text()).body:
+            if (
+                isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef))
+                and node.name.startswith("list_")
+            ):
+                out[node.name] = {
+                    a.arg for a in node.args.args
+                } | {a.arg for a in node.args.kwonlyargs}
+    return out
+
+
+def test_the_tools_package_is_where_we_think_it_is():
+    """If this fails the sweep below is silently checking nothing."""
+    tools = _list_tools()
+    assert len(tools) >= 15, f"found only {len(tools)} list tools — did the package move?"
+
+
+def test_every_capped_list_tool_can_be_paged():
+    """A `limit` promises a cap; without an `offset` it also imposes a ceiling."""
+    tools = _list_tools()
+    capped = {name for name, args in tools.items() if "limit" in args}
+    assert capped, "no list tool takes a limit — the sweep is not finding signatures"
+
+    unpageable = sorted(
+        name for name in capped
+        if "offset" not in tools[name] and name not in _DELIBERATELY_UNPAGED
+    )
+    assert not unpageable, (
+        f"these list tools cap their results with no way to page past the cap: "
+        f"{unpageable}. Each hands the caller a `total` it cannot reach. Add an "
+        f"`offset` (check the service first — it usually already takes one), or "
+        f"add the tool to _DELIBERATELY_UNPAGED with a reason saying why there "
+        f"is no 'rest' to ask for."
+    )
+
+
+def test_the_unpaged_exemptions_still_exist():
+    """A stale exemption is an exemption for nothing, and it hides the next
+    tool that inherits the name. Same reverse check `test_mcp_auth` runs on
+    its allow-lists."""
+    tools = _list_tools()
+    missing = sorted(_DELIBERATELY_UNPAGED - set(tools))
+    assert not missing, (
+        f"_DELIBERATELY_UNPAGED names tools that no longer exist: {missing}. "
+        f"Renamed or deleted — drop them from the set."
+    )
+    still_capped = sorted(
+        name for name in _DELIBERATELY_UNPAGED
+        if name in tools and "limit" not in tools[name]
+    )
+    assert not still_capped, (
+        f"these are exempted from paging but no longer take a `limit` at all, "
+        f"so the exemption is moot: {still_capped}."
+    )
+
+
+def test_offset_never_appears_without_limit():
+    """The inverse, and it is a real bug rather than a style point: an offset
+    with no cap pages through an unbounded result set, so page 2 of an
+    ever-growing list silently returns everything after the skip."""
+    tools = _list_tools()
+    bad = sorted(
+        name for name, args in tools.items()
+        if "offset" in args and "limit" not in args
+    )
+    assert not bad, f"these take an offset but no limit: {bad}"

tests/test_services_retrieval_telemetry.py

@@ -102,3 +102,123 @@ async def test_insert_retrieval_log_roundtrip(_dispose_engine):
         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 ai["cleared_threshold"] == 2      # 0.91 and 0.72, not 0.40
+        # 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
+
+
+@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()

tests/test_snippet_live_body.py

@@ -272,3 +272,89 @@ async def test_forge_failure_inside_lookup_never_breaks_the_pull():
         data = _data()
         await svc.attach_live_body(_note(), data)
     assert "body_source" not in data
+
+
+# --- #2782: an annotated record is not a diverged one ------------------------
+# Containment is right for a verbatim record and wrong for a deliberately
+# annotated one: the commentary that makes the record worth reading is exactly
+# what makes `cached in fetched` false, forever. These pin the escape hatch —
+# a standing `ok` verdict stamped at the commit we just fetched — and, just as
+# importantly, every condition that must switch it back off.
+
+ANNOTATED = "# Membership is the contract — this record says WHY, the source can't.\n" + CODE
+
+
+def _ok_verdict(code=ANNOTATED, commit=SHA, **extra):
+    verdict = svc.compose_verification(
+        status=svc.VERIFY_OK, checked_code_sha=svc.code_sha(code), commit_sha=commit
+    )
+    verdict.update(extra)
+    return verdict
+
+
+async def _freshness(data, *, file_commit=SHA, content=CODE):
+    forge = _forge_with(lambda r: _file_response(content, commit_sha=file_commit))
+    with _patched(forge), patch.object(svc.notes_svc, "update_note", AsyncMock()):
+        await svc.attach_live_body(_note(), data)
+        await background.drain()
+    return data["body_source"], data["body_freshness"]
+
+
+async def test_annotated_record_with_a_standing_verdict_reads_current():
+    """The bug: the record's commentary is absent from the source, so
+    containment fails and every pull said `diverged`. A verdict that already
+    judged this body faithful, at this very commit, outranks the substring."""
+    data = _data(code=ANNOTATED, verification=_ok_verdict())
+    assert await _freshness(data) == ("forge", "current")
+    assert data["snippet"]["code"] == ANNOTATED     # still never rewritten
+
+
+async def test_the_verdict_vouches_for_one_commit_only():
+    """The guard that keeps the hatch honest. The file has moved past the
+    commit the verdict was stamped at, so nobody has judged what is there
+    now — containment resumes as the authority and the record reads diverged
+    until someone re-verifies."""
+    data = _data(code=ANNOTATED, verification=_ok_verdict(commit="a" * 40))
+    assert await _freshness(data) == ("cache", "diverged")
+
+
+async def test_an_expired_verdict_does_not_vouch():
+    """The record was edited after the check, so `code_sha` no longer matches
+    and the verdict describes a body that is not this one."""
+    data = _data(code=ANNOTATED, verification=_ok_verdict(code="def other(): pass"))
+    assert await _freshness(data) == ("cache", "diverged")
+
+
+async def test_a_push_invalidated_verdict_does_not_vouch():
+    """A push touched the recorded location since the check (#2691) — the repo
+    moved under the verdict even though the record didn't."""
+    data = _data(code=ANNOTATED, verification=_ok_verdict(invalidated_by="c" * 40))
+    assert await _freshness(data) == ("cache", "diverged")
+
+
+async def test_only_an_ok_verdict_vouches():
+    """A drifted verdict is evidence AGAINST the body, not for it."""
+    verdict = svc.compose_verification(
+        status=svc.VERIFY_CHANGED, checked_code_sha=svc.code_sha(ANNOTATED), commit_sha=SHA
+    )
+    data = _data(code=ANNOTATED, verification=verdict)
+    assert await _freshness(data) == ("cache", "diverged")
+
+
+async def test_a_verbatim_record_still_takes_the_containment_path():
+    """No regression: the happy path does not route through the hatch, and an
+    unverified verbatim record is still confirmed by containment alone."""
+    data = _data(code=CODE)
+    assert await _freshness(data) == ("forge", "current")
+
+
+async def test_a_verdict_predating_commit_stamping_does_not_vouch():
+    """Verdicts recorded before `commit_sha` existed (#2688) carry no commit to
+    compare, so they cannot tie the body to a known state of the source. They
+    fall through to containment rather than vouching on age alone."""
+    verdict = svc.compose_verification(
+        status=svc.VERIFY_OK, checked_code_sha=svc.code_sha(ANNOTATED)
+    )
+    assert "commit_sha" not in verdict
+    data = _data(code=ANNOTATED, verification=verdict)
+    assert await _freshness(data) == ("cache", "diverged")