test(telemetry): pin the identity that falsified this milestone (#3668)

Reverts the deliberate break from 5e19a1b. `plugin_context.py` is now
byte-identical to before it; the only change against that baseline is the
guard itself.

FALSIFIED, not argued (rule 167). CI run 6046 with `results=hits` in both
arms failed exactly the predicted four cases —

    [one-already-held-write_path]   FAILED
    [one-already-held-pre_tool]     FAILED
    [all-already-held-write_path]   FAILED
    [all-already-held-pre_tool]     FAILED
    [nothing-held-*]                passed

— and the nothing-held cases passing is the point, not a gap: with no
exclusions both recorders see the same list however wrongly they are
wired, so that case can never discriminate and a guard built only from it
would read as coverage while catching nothing.

WHAT IS PINNED. Both arms build one `fresh` list and hand it to two
recorders in one function, so the call log and the surfacing log cannot
disagree about what a single call showed. Ids, not counts: equal counts
drawn from different lists is a real way for this to break, and a count
comparison would call it agreement.

Three hits, where production returns at most one. The identity holds at
any limit because both recorders read the same list, and stating it that
way survives RULEHINT_LIMIT moving again — it has moved once already
(2 → 1, 2385100), and that move is half of why the original
reconstruction misread its own numbers.

NOT DONE, deliberately: the readout-level self-check the task also
proposed. `cleared_threshold` counts CALLS that beat the bar while
`surfaced` counts RULES, so that identity holds only while the limit is
1 — it would fire on a healthy system the moment the limit rises. The
arm-level form has no such coupling, which is why the guard lives here.

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

alembic/versions/0095_retrieval_log_suppressed_count.py

@@ -0,0 +1,52 @@
+"""add retrieval_logs.suppressed_count — tell a ranker decline from a repeat (#3497)
+
+Revision ID: 0095
+Revises: 0094
+Create Date: 2026-09-03
+
+`result_count == 0` has always meant "this surface said nothing", which is the
+right number for "was the hint any use" and the wrong one for tuning a
+threshold. It folds together two unrelated events:
+
+  - the ranker found nothing above the bar — the ONLY evidence a threshold is
+    set too high; and
+  - the ranker found something the session had already been shown — a decline
+    that says nothing whatever about the bar.
+
+The rule arms filter in Python after the search, so they can count the second
+kind exactly. The note arms pass `exclude_ids` INTO semantic_search_notes, so
+the dropped rows never come back and there is nothing to count.
+
+NULLABLE, AND THE NULL IS THE POINT. A surface that does not measure
+suppression stores NULL, not 0, and the readout renders it as "not measured"
+rather than "none". Defaulting to 0 would make an unmeasured surface look like
+a perfectly clean one — the exact substitution of an artifact for a
+measurement that #3311 made and that #3497 exists to correct. Doing it again,
+in the migration that fixes it, would be its own small joke.
+
+No backfill for the same reason: existing rows genuinely do not know, and
+saying so is the honest state. `retrieval_logs` is not restored from backup,
+so no importer changes.
+
+Downgrade drops the column. Purely observational — nothing reads it for
+correctness.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0095"
+down_revision = "0094"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "retrieval_logs",
+        sa.Column("suppressed_count", sa.Integer(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("retrieval_logs", "suppressed_count")

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": "2026.09.03.0329",
+  "version": "2026.09.04.0140",
   "author": {
     "name": "Bryan Van Deusen"
   },

plugin/hooks/scribe_static_context.md

@@ -21,6 +21,16 @@ for the operator's work, and as your own working memory across sessions.
   compaction — call `list_always_on_rules()` (and `enter_project()` when a
   project is in scope) BEFORE acting. When a loaded rule and a default habit
   disagree, the rule wins; if no rule speaks to it, ask rather than assume.
+- **What you loaded is not all of the rules.** Only the always-on tier arrives
+  that way; conditional rules are RETRIEVED, and one you were never handed
+  binds exactly as hard. So before a consequential act, `search` for a rule
+  about it (`content_type="rule"`) rather than concluding from an empty
+  loaded set that nothing applies. "I was not told" is not the same as "there
+  is no rule," and only one of those is checkable.
+  This bites hardest on which TOOL to reach for — curling an API that has an
+  MCP client, standing up a local stack, running a suite CI owns. Those feel
+  like mechanics rather than decisions, so they raise no doubt and generate no
+  query; the moment you are most confident is the moment to look.
 - **Recall before acting** — before you answer anything about the operator's
   work or start a task, `search` Scribe first; assume a related note, task, or
   decision already exists. Concretely, reach for recall whenever a request

plugin/skills/using-scribe/SKILL.md

@@ -56,10 +56,31 @@ Two constraints on *how* that's achieved:
    re-deriving it or opening a duplicate. When a project is in scope, pass its
    `project_id` so results stay scoped.
 
-2. **Standing rules are binding.** Load them via `list_always_on_rules()` at
-   session start (see "Do this first"); treat every one as binding. Pull a
-   rule's full statement with `get_rule(id)` when it's about to bite. When a
-   project is in scope, `enter_project(id)` also returns its applicable rules.
+2. **Standing rules are binding — and the ones you were handed are not all of
+   them.** Load the resident set via `list_always_on_rules()` at session start
+   (see "Do this first"); treat every one as binding. Pull a rule's full
+   statement with `get_rule(id)` when it's about to bite. When a project is in
+   scope, `enter_project(id)` also returns its applicable rules.
+
+   Rules come in two tiers. **Always-on** rules are delivered — they arrive
+   whether or not you ask. **Conditional** rules are RETRIEVED, and one binds
+   just as hard for never having been handed to you. So before a consequential
+   act, `search(content_type="rule")` on what you are about to do. An empty
+   loaded set is not evidence that no rule applies; it is only evidence that
+   none was pushed, and those are different claims.
+
+   The tier split exists because delivery does not scale: every resident rule
+   costs tokens in every session forever, so a rulebook that grows past a few
+   dozen either stops growing or stops fitting. Retrieval is what lets the
+   rulebook keep growing — but retrieval only fires if something asks.
+
+   **Ask hardest where you feel most certain.** Rules about which TOOL to reach
+   for — use the forge's MCP client rather than curling its API, don't stand up
+   a local stack, don't run the suite CI owns — govern moves that feel like
+   mechanics rather than decisions. A reflex raises no doubt, so it generates
+   no query, so the rule that would have stopped it is never retrieved. That is
+   the failure this instruction exists to prevent, and confidence is its only
+   warning sign.
 
 3. **Update over duplicate.** When recording, prefer updating an existing
    note/rule/task over creating a new one. Search first; revise what's there.

src/scribe/mcp/server.py

@@ -45,6 +45,31 @@ from quart import Quart
 # The accepted cost: an agent that never opens create_note's docstring never
 # learns the field exists. Guidance lives in the create_note / update_note
 # docstrings and the using-scribe skill instead.
+#
+# Milestone 333 step 3 (2026-09-04) bought the HOW bullet's second clause —
+# search(content_type="rule") before a consequential act — by TRADING OUT
+# "Processes are saved procedures (follow verbatim)" and "Deletes are
+# trash-recoverable". Recorded so the trade is not silently reversed:
+#   - Both were already in test_instruction_surfaces_agree's DISPLACED_TOPICS
+#     and already stated on a delivered surface, so nothing fell off: the
+#     process reflex is in every scribe-proc-* skill listing (each says the
+#     process governs and is followed verbatim), and trash recovery is in the
+#     delete_*/list_trash/restore docstrings, which is where per-tool guidance
+#     belongs by this block's own doctrine.
+#   - What it bought is not per-tool guidance and has nowhere else to live at
+#     session-start altitude. Rules were retrievable only by RESIDENCY: the
+#     always-on preload put them in front of the agent, and nothing told a
+#     session to go looking for one it had not been handed. The tier split is
+#     therefore load-bearing on ANY install (rule 115): a delivered rule costs
+#     tokens in every session forever, so a rulebook that only delivers cannot
+#     grow past what one session can hold, and every rule worth keeping has to
+#     become resident to bind at all. Retrieval is what lets it keep growing —
+#     and retrieval fires only if something asks, which nothing told a session
+#     to do. A tool-choice reflex asks least of all (#3476, #161).
+#   - This states the PULL for conditional rules, exactly as the surrounding
+#     line states it for always-on ones. Rule 119 makes these surfaces the
+#     specification, so the same sentence lands on all three session-start
+#     surfaces, and test_instruction_surfaces_agree pins it.
 _INSTRUCTIONS = """
 Scribe is the operator's self-hosted second brain and system of record — and
 yours: recall from it before acting, record as you go. Keep no parallel copy
@@ -63,13 +88,13 @@ Hierarchy: Project -> Milestone -> Task/Note. The map, by purpose:
   active project_id to stay in scope.
 - WHERE work happens: Systems. Tag records with system_ids as you write;
   create_system when the area is unmodelled.
-- HOW: rules are binding — list_always_on_rules() at session start.
+- HOW: rules bind. list_always_on_rules() at start; before a consequential
+  act, search(content_type="rule") — the resident set is not all of them.
 - UI: the project's design system is binding — resolve_design_system /
   get_design_system_stylesheet before hand-writing a value.
 - REUSE: search snippets before writing a helper; record what you build with
   create_snippet; classify shapes against canon (classify_shapes) — a
-  consumer map is rows, never prose. Processes are saved procedures (follow
-  verbatim). Deletes are trash-recoverable.
+  consumer map is rows, never prose.
 
 A task is a note with status (*_note vs *_task tools).
 Creates are duplicate-gated: a near-match BLOCKS and returns the existing

src/scribe/mcp/tools/rulebooks.py

@@ -315,6 +315,61 @@ async def create_rule(
 ) -> dict:
     """Create a new rule in a rulebook (a SHARED rule — keep it general).
 
+    PROPOSE RULES READILY, AND WRITE ONE WHEN THE OPERATOR SAYS YES. Noticing
+    that something has hardened into a standing instruction is valuable work,
+    and a session that notices it and says nothing has thrown the observation
+    away. So raise it whenever you see one. The single step that belongs
+    between noticing and writing is the operator's yes: a rule binds every
+    future session, and they are the person it binds.
+
+    Their yes is also the only moment the rule is reliably IN FRONT of them.
+    After the write it may not be again for months — a conditional rule is not
+    read aloud at session start, and a project-scoped one does not appear in
+    an unfiltered list_rules() at all. So the proposal is the review.
+
+    When the operator asks for a rule in so many words, that IS the yes —
+    write it and move on. The loop below is for the rule you thought of.
+
+    A PROPOSAL CARRIES FOUR THINGS, and the fourth is the one that decides it:
+
+      1. WHAT it would require — the statement, in the words it would carry,
+         not a gloss of them. The operator is agreeing to text.
+      2. INTENT — what it changes about how work gets done, and what goes
+         wrong today without it. "Be careful about X" is not an intent; the
+         behaviour that would differ tomorrow is.
+      3. WHY NOW — the incident, observation or decision behind it. Pass that
+         record as arose_from_id, and say it in the conversation too: the
+         field is for the reader six months out, the sentence is for the
+         person deciding.
+      4. HOW IT WOULD BE ENFORCED — a test, a CI check, a hook, a schema
+         constraint, a duplicate gate, a review step... or nothing, in which
+         case say so plainly: "nothing — this is prose a session has to
+         remember." Answer this one honestly and it will sometimes dissolve
+         the rule, which is the point rather than a side effect. What a test
+         can assert should BE that test; a rule is what remains when nothing
+         mechanical can hold the thing. A rulebook grows by default and
+         shrinks only on purpose, so a question that prevents a rule is worth
+         more than any question that improves one's wording.
+
+    THEN CLOSE WITH A QUESTION THEY CAN ANSWER IN ONE WORD. Offer three
+    answers, and make the middle one the easy one:
+
+      * "Approve it AS WRITTEN" — you create it with the statement exactly as
+        shown. This is what makes element 1 load-bearing: they approved TEXT,
+        so that text is what gets stored, verbatim.
+      * "LET'S TALK ABOUT IT" — the wording, the scope, the tier, whether it
+        wants to be a rule at all. Most good rules arrive this way, so treat
+        this answer as the expected one rather than a setback.
+      * "NO" — let it go. If the observation is still worth keeping, it is a
+        note (create_note): recorded, findable, and binding on nobody.
+
+    Where the interface offers structured choices, ask it that way — a
+    question with named options is answered in a click, while the same
+    question inside a paragraph is answered by scrolling past. Where it does
+    not, write the three options out as three options. Either way ask once
+    and let the answer stand; re-raising a declined proposal argues a rule
+    into existence, which is the thing this whole loop exists to prevent.
+
     A rulebook rule is shared by every project that gets the rulebook: an
     always_on rulebook binds ALL your projects; a subscribed rulebook binds the
     projects that opt in. So a rulebook rule must read as a general standard —
@@ -433,6 +488,16 @@ async def create_project_rule(
     the rule is returned in get_project's applicable_rules (under
     project_rules) and in list_rules(project_id=...).
 
+    PROPOSE, THEN WRITE ON A YES — create_rule's opening carries the whole
+    loop: the four things a proposal states (what it would require, its
+    intent, why now, and how it would be enforced) and the one-word question
+    that closes it (approve as written / talk about it / no). All of it
+    applies here unchanged. Reach for that loop MORE readily on this surface,
+    not less: a project rule stays out of an unfiltered list_rules(), and a
+    conditional one stays out of session start too, so the operator's yes is
+    the one moment this rule is certain to have been seen by the person it
+    binds.
+
     Check first whether a rule is the right shape at all — create_rule's
     opening asks that question and it applies identically here. A visual
     standard is a design system; a procedure is a process (create_process);

src/scribe/mcp/tools/search.py

@@ -168,6 +168,20 @@ async def retrieval_telemetry(days: int = 30) -> dict:
     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.
 
+    READ `cleared_threshold` AND `zero_result_calls` TOGETHER, and check
+    `suppression` before concluding anything from either. A zero-result call is
+    two different events wearing one number: the ranker found nothing above the
+    bar, or it found only what this session had already been shown. Just the
+    first is evidence the bar is too high. `suppression` splits them where the
+    surface can tell — `zero_because_already_shown` comes off
+    `zero_result_calls` to leave the true ranker declines.
+
+    `suppression` is `null` when NO row in the window reported it, and that is
+    "not measured here", NOT "none suppressed". Surfaces that pass their
+    exclusions into the search never see what was dropped, so they cannot say.
+    Do not read a null as a zero: reading an artifact as a measurement is how
+    this surface got mis-scoped once already (#3311, #3497).
+
     `usage` — NOTES ONLY, 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
@@ -223,10 +237,37 @@ It is an UPPER BOUND per surface: a pull records the door it came
     those same rules over time: a resident set surfaced thousands of times and
     opened never is the dead-weight signal, one tier up.
 
-    Read it against `sources["write_path_rule"]`. That surface has never once
-    declined to fire, and until this block existed there was no way to tell a
-    well-tuned arm from a bar it cannot fail to clear (#3311). `pull_through`
-    is the number that tells them apart.
+    Read it against `sources["write_path_rule"]`. That arm was once believed
+    never to decline — the reading that scoped #3311 — but it was the arm's
+    `retrieval_logs` row being written only on calls that FOUND something, so
+    the zeros were missing rather than absent (#3497). Measured since, it
+    declines the large majority of its calls like any other surface.
+
+    EVERY COUNTER BLOCK CARRIES ITS OWN COVERAGE — `complete_from` and
+    `covers_window`. `complete_from` is when the number became trustworthy:
+    for one source, its first recorded row; for a section that sums several,
+    the LATEST of theirs, because a total is complete only once every
+    contributor was being written. `covers_window: false` means the window
+    reaches back further than the recording does, so the count is a fraction
+    of the period it appears to describe.
+
+    READ IT BEFORE COMPARING TWO NUMBERS, and especially before comparing
+    across a deploy. A counter added last week, read over a 30-day window,
+    reports a real count against an imagined denominator — and the result is
+    a plausible fraction rather than an obvious zero, which is what makes it
+    dangerous. That reading cost milestone #379 five steps aimed at a defect
+    that did not exist.
+
+    `covers_window` is null, never false, when nothing was ever recorded:
+    "no measurement" is not "partial measurement", the same distinction
+    `suppression`'s null carries a few paragraphs up.
+
+    A SOURCE SHOWING `calls: 0` WAS RECORDING AND MADE NO CALLS. `sources`
+    lists every source the table has ever held, not only those active in the
+    window, so a surface that stopped firing stays visible rather than
+    disappearing — being absent is reserved for a source that has never
+    recorded at all. Its score fields are null, not zero: the calls are a
+    real observation, the distribution is not one.
 
     `rule_usage_failed: true` means that read failed while the rest of the
     readout stood. The counts are still present so a caller can render, but

src/scribe/models/retrieval_log.py

@@ -42,6 +42,16 @@ class RetrievalLog(Base):
     # False=notes, NULL=any.
     is_task: Mapped[bool | None] = mapped_column(Boolean, nullable=True)
     result_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    # How many scored hits this call DROPPED because the session had already
+    # been shown them. NULLABLE, and the null is load-bearing: it means "this
+    # surface does not report suppression", which must not read as "nothing was
+    # suppressed". `result_count == 0` alone conflates two different events —
+    # the ranker found nothing above threshold, and the ranker found something
+    # the reader already had — and only the first says a threshold is too high.
+    # Reading a zero as a ranker decline is how #3311 mis-scoped a milestone;
+    # an unmeasured value that renders as 0 is the same mistake with a nicer
+    # face, so surfaces that filter INSIDE the search leave this null.
+    suppressed_count: Mapped[int | None] = mapped_column(Integer, nullable=True)
     top_score: Mapped[float | None] = mapped_column(Float, nullable=True)
     min_score: Mapped[float | None] = mapped_column(Float, nullable=True)
     # [{"id": int, "score": float, "rank": int}, ...], highest-first.
@@ -67,6 +77,7 @@ class RetrievalLog(Base):
             "project_id": self.project_id,
             "is_task": self.is_task,
             "result_count": self.result_count,
+            "suppressed_count": self.suppressed_count,
             "top_score": self.top_score,
             "min_score": self.min_score,
             "result_ids": self.result_ids,

src/scribe/services/embeddings.py

@@ -774,10 +774,17 @@ async def semantic_search_rules(
     is the surfacing question, and it has its own machinery
     (get_applicable_rules) rather than a second, subtly different copy here.
 
-    `tier` narrows to one tier. The write-path hint passes "conditional",
-    because an always-on rule is ALREADY in the session — surfacing it again as
-    a suggestion is pure noise, and noise on a hint that fires on every write
-    is how a hint gets ignored.
+    `tier` narrows to one tier, and NONE is the ordinary case. The write-path
+    and pre-tool hints deliberately pass nothing: an always-on rule is already
+    in the session, but being in a list from turn zero is not the same as being
+    in front of the reader when the action it governs is taken, and filtering
+    on tier made a whole class of rules permanently ineligible for the one
+    mechanism that surfaces a rule AT the moment. Relevance is the threshold's
+    job; see the block above RULEHINT_LIMIT in services/plugin_context.py for
+    the argument and for what the resulting scores are being read against.
+
+    Pass a tier when a caller genuinely wants one class — a listing, an audit,
+    a UI that renders the tiers apart. Not to approximate relevance.
 
     Collapses to best-chunk-per-rule like the note search, so a long rule split
     across chunks competes once rather than crowding the results with itself.

src/scribe/services/plugin_context.py

@@ -94,13 +94,17 @@ WRITEPATH_DEFAULT_THRESHOLD = 0.68
 # THE STRUCTURAL ARGUMENT, which is the only kind admissible here (rule 115).
 # Two facts hold on any install, including one with six rules and no telemetry:
 #
-#   1. The eligible corpus is TINY. The arm searches `tier="conditional"`
-#      rules only — a handful to a few dozen documents against thousands of
-#      notes. A top-k over forty candidates always returns something, so
-#      "the best match cleared the bar" stops meaning "a good match exists"
-#      and starts meaning "forty things were ranked". A bar calibrated for
-#      best-of-thousands is cleared by best-of-forty as arithmetic, not
-#      relevance.
+#   1. The eligible corpus is SMALL — every rule an install owns, still only
+#      a few dozen documents against thousands of notes. A top-k over a small
+#      pool always returns something, so "the best match cleared the bar"
+#      drifts from "a good match exists" toward "N things were ranked". A bar
+#      calibrated for best-of-thousands is cleared by best-of-forty as
+#      arithmetic rather than relevance.
+#      This argument WEAKENED when the arms stopped filtering to one tier
+#      (see the note on that below): a larger pool makes clearing the bar
+#      mean more, not less. The threshold was deliberately left where it was
+#      anyway — moving two variables at once would make the resulting
+#      distribution unreadable, and this one errs toward silence on purpose.
 #   2. Rules are short imperative technical English — a far more HOMOGENEOUS
 #      corpus than note prose. #2223 measured the floor for code against prose
 #      at 0.55-0.63 and set 0.68 above it. A more homogeneous corpus has a
@@ -128,9 +132,14 @@ RULEHINT_DEFAULT_THRESHOLD = 0.72
 # ONE rule per write, not two — and this is deliberately NOT a knob.
 #
 # With a corpus this small, top-k does as much damage as the threshold: k=2
-# over forty candidates means the second line is almost always the second-best
-# noise, arriving with the same confident framing as the first. Halving k
-# halves that regardless of where the bar sits.
+# over a few dozen candidates means the second line is almost always the
+# second-best noise, arriving with the same confident framing as the first.
+# Halving k halves that regardless of where the bar sits.
+#
+# It also BOUNDS the blast radius of widening the pool (below): with k=1 a
+# wider corpus can change WHICH rule surfaces and how often one does, but it
+# can never make a single hint longer. The loudness of one hint and the
+# eligibility of a rule are separate controls, and only one of them moved.
 #
 # It stays a constant because it is a decision about how LOUD one hint may be,
 # not a per-install tuning question. The hint already carries prior art, shape
@@ -140,6 +149,45 @@ RULEHINT_DEFAULT_THRESHOLD = 0.72
 # adds a way to misconfigure the surface (rule 25 cuts both ways).
 RULEHINT_LIMIT = 1
 
+# WHY THE ARMS NO LONGER FILTER TO ONE TIER (#3702).
+#
+# Both arms used to pass `tier="conditional"`, on the reasoning that an
+# always-on rule is already in the session, so surfacing it again is pure
+# noise. That reasoning conflates two different things:
+#
+#   PRESENT IN CONTEXT  — the rule was delivered at session start.
+#   SALIENT AT THE MOMENT  — the rule is in front of the reader when the
+#                            action it governs is about to be taken.
+#
+# A rule handed over in a list at turn zero is present while a session writes
+# a config value three hundred turns later. It is not surfaced. So the filter
+# did not merely skip a redundant hint — it made a whole class of rules
+# permanently ineligible for the only mechanism that puts a rule in front of
+# an agent AT the moment, and the more important a rule is, the more likely
+# it was in that class.
+#
+# The deeper defect is that the filter was doing the THRESHOLD's job. Whether
+# a rule belongs in this hint is a relevance question, and a similarity bar is
+# the control for relevance. A categorical exclusion standing in for a
+# relevance judgment cannot be tuned, cannot be measured, and cannot be wrong
+# in a way anybody notices.
+#
+# THIS IS A MEASURED CHANGE, NOT A SETTLED ONE. The old comment's fear is
+# real — a hint that fires on every write and says obvious things teaches the
+# reader to skip the block, and the surface is then lost along with its true
+# positives. That fear had simply never been checked. `retrieval_logs` already
+# records top_score, result_count and the query for every call, so the
+# evidence now arrives on its own:
+#
+#   - rules clear the bar often and at high scores  -> the fear was justified,
+#     the filter was a crude proxy for a bar set too low, and the WORK IS THE
+#     BAR. Any reinstated filter should then carry a measured reason.
+#   - rules clear rarely, in a thin band near the bar -> the filter was never
+#     the right instrument and relevance was always sufficient.
+#
+# Only the eligibility moved. The bar and k=1 were both left exactly where
+# they were, so the resulting distribution has one cause.
+
 # How much of a command reaches the embedding (#3476). A shell call is not a
 # file: most are short, and the ones that are not are usually a heredoc or a
 # pasted script whose bulk says nothing about which rule applies. The VERB AND
@@ -1205,7 +1253,7 @@ async def build_write_path_hint(
         rule_t0 = time.perf_counter()
         hits = await semantic_search_rules(
             user_id, code or path, limit=RULEHINT_LIMIT,
-            threshold=cfg["rule_threshold"], tier="conditional",
+            threshold=cfg["rule_threshold"],
         )
         rule_ms = (time.perf_counter() - rule_t0) * 1000.0
         fresh = [(score, rule) for score, rule in hits if rule.id not in already]
@@ -1253,6 +1301,11 @@ async def build_write_path_hint(
             threshold=cfg["rule_threshold"], limit=RULEHINT_LIMIT,
             project_id=project_id,
             is_task=None, results=fresh, duration_ms=rule_ms,
+            # What the ranker found and this session had already been told.
+            # Without it a zero row cannot say whether the bar was too high or
+            # the reader was simply ahead of it — and only the first is a
+            # reason to move the threshold.
+            suppressed=len(hits) - len(fresh),
         )
         if fresh:
             # `rule_ids` is `fresh`, i.e. AFTER exclude_rule_ids. A rule the
@@ -1332,7 +1385,7 @@ async def build_tool_rule_hint(
         t0 = time.perf_counter()
         hits = await semantic_search_rules(
             user_id, query, limit=RULEHINT_LIMIT,
-            threshold=cfg["rule_threshold"], tier="conditional",
+            threshold=cfg["rule_threshold"],
         )
         duration_ms = (time.perf_counter() - t0) * 1000.0
 
@@ -1352,6 +1405,10 @@ async def build_tool_rule_hint(
             threshold=cfg["rule_threshold"], limit=RULEHINT_LIMIT,
             project_id=project_id,
             is_task=None, results=fresh, duration_ms=duration_ms,
+            # See the sibling arm. It matters more here: this arm fires on every
+            # Bash call, so a long session excludes its way to an all-zero row
+            # and the threshold looks wrong when nothing about it is.
+            suppressed=len(hits) - len(fresh),
         )
         if not fresh:
             return out

src/scribe/services/retrieval_telemetry.py

@@ -55,12 +55,18 @@ def _build_payload(
     is_task: bool | None,
     results: list[tuple[float, Note]],
     duration_ms: float | None,
+    suppressed: int | None = None,
 ) -> dict:
     """Reduce a retrieval call to a flat, JSON-safe RetrievalLog payload.
 
     Pure and synchronous (no DB, no event loop) so it is unit-testable and safe
     to run inline before scheduling the write. `results` is the
     `(score, Note)` list from semantic_search_notes, already highest-first.
+
+    `suppressed` is how many scored hits the caller dropped because the session
+    had already been shown them, and it stays None for callers that cannot
+    know. See the column's comment: None means "not measured here", which is a
+    different fact from 0 and must never render as one.
     """
     items = [
         {"id": int(note.id), "score": round(float(score), 5), "rank": rank}
@@ -76,6 +82,7 @@ def _build_payload(
         "project_id": project_id,
         "is_task": is_task,
         "result_count": len(items),
+        "suppressed_count": (None if suppressed is None else int(suppressed)),
         "top_score": (scores[0] if scores else None),
         "min_score": (scores[-1] if scores else None),
         "result_ids": items,
@@ -115,6 +122,7 @@ def record_retrieval(
     is_task: bool | None,
     results: list[tuple[float, Any]],
     duration_ms: float | None = None,
+    suppressed: int | None = None,
 ) -> None:
     """Fire-and-forget: record one retrieval call.
 
@@ -140,6 +148,7 @@ def record_retrieval(
             is_task=is_task,
             results=results,
             duration_ms=duration_ms,
+            suppressed=suppressed,
         )
     except Exception:
         logger.debug("retrieval telemetry payload build failed", exc_info=True)
@@ -166,7 +175,8 @@ def record_retrieval(
 
 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
+    (calls, zero, cleared, p10, p50, p90, lo, hi, avg_n, dur,
+     measured, supp_calls, supp_zero) = rows
     return {
         "calls": int(calls or 0),
         # A call that returned nothing is not a low-scoring call — it is a
@@ -178,6 +188,22 @@ def _bucket(rows: list) -> dict:
         # 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),
+        # Of the zeros above, which were the RANKER declining and which were
+        # the reader having seen it already? `zero_result_calls` cannot say,
+        # and only the first kind is evidence about the threshold.
+        #
+        # None — not a zeroed dict — when no row in the window reported it. A
+        # surface that filters inside the search genuinely does not know, and
+        # rendering that as `{"calls": 0}` would state a measurement nobody
+        # made. That substitution is the whole of #3311.
+        "suppression": (
+            None if not int(measured or 0) else {
+                "measured_calls": int(measured or 0),
+                "calls_with_suppression": int(supp_calls or 0),
+                # Subtract from zero_result_calls for the true ranker declines.
+                "zero_because_already_shown": int(supp_zero or 0),
+            }
+        ),
         "top_score": {
             "p10": _round(p10), "p50": _round(p50), "p90": _round(p90),
             "min": _round(lo), "max": _round(hi),
@@ -187,10 +213,70 @@ def _bucket(rows: list) -> dict:
     }
 
 
+# The aggregate row Postgres would have returned for a source with no rows in
+# the window: nothing counted, nothing scored. Positional, matching the SELECT
+# `_bucket` unpacks — calls, zero, cleared, p10, p50, p90, min, max, avg_n,
+# dur, measured, supp_calls, supp_zero. The three counts are 0 because zero
+# calls is a real observation; everything else is None because a distribution
+# nobody sampled has no value, and rendering it as 0.0 would state one.
+_NO_ROWS_IN_WINDOW = [0, 0, 0, None, None, None, None, None, None, None, 0, 0, 0]
+
+
 def _round(v, places: int = 4):
     return None if v is None else round(float(v), places)
 
 
+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
+        ),
+    }
+
+
 async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
     """What the retrieval telemetry says, per surface, over a window.
 
@@ -240,6 +326,14 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
         else_=0,
     )
     zero = case((RetrievalLog.result_count == 0, 1), else_=0)
+    # Three sums rather than one, because "not measured" and "measured as zero"
+    # are different answers and a single counter cannot hold both.
+    measured = case((RetrievalLog.suppressed_count.isnot(None), 1), else_=0)
+    supp_calls = case((RetrievalLog.suppressed_count > 0, 1), else_=0)
+    supp_zero = case(
+        ((RetrievalLog.result_count == 0) & (RetrievalLog.suppressed_count > 0), 1),
+        else_=0,
+    )
 
     def pct(p: float):
         return func.percentile_cont(p).within_group(RetrievalLog.top_score.asc())
@@ -249,6 +343,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:
@@ -266,6 +363,9 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
                         func.percentile_cont(0.9).within_group(
                             RetrievalLog.duration_ms.asc()
                         ),
+                        func.sum(measured).label("measured"),
+                        func.sum(supp_calls).label("supp_calls"),
+                        func.sum(supp_zero).label("supp_zero"),
                     )
                     .where(
                         RetrievalLog.created_at >= since,
@@ -274,8 +374,34 @@ 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
+
+            # A source with rows in the table but NONE in this window would
+            # otherwise be absent from the readout — and absent is exactly how
+            # a source that never existed renders, so a surface that WAS
+            # recording and went silent is unreadable (#3720). That is #2663
+            # one level up: the failure that looks like the correct answer.
+            #
+            # Zero here is a real measurement, not a manufactured one. The
+            # all-time query proves the source was recording, and it made no
+            # calls across a window it fully covers — which is why no
+            # `covers_window` special case is needed: a source whose first row
+            # fell after `since` would have that row IN the window and already
+            # hold a bucket, so anything reaching here began before it.
+            for src, first_row in log_complete.items():
+                if src == "*" or first_row is None or src in out["sources"]:
+                    continue
+                quiet = _bucket(list(_NO_ROWS_IN_WINDOW))
+                quiet.update(_coverage(first_row, since))
+                out["sources"][src] = quiet
 
             # The corpus side, at its own grain. `ambient` mirrors
             # note_usage.usage_for_notes: an ambient surfacing was not a scored
@@ -302,6 +428,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
@@ -429,6 +556,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.
@@ -538,6 +668,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 ────────────────────────────
@@ -615,6 +749,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

tests/test_instruction_surfaces_agree.py

@@ -195,6 +195,46 @@ def test_displaced_topics_live_on_a_delivered_surface():
     )
 
 
+# The SECOND pull (milestone 333 step 3). `list_always_on_rules()` fetches the
+# resident tier; this one says that tier is not all of them, and that a
+# conditional rule has to be gone looking for. However a surface words the
+# surrounding prose, it names the call.
+RETRIEVE = 'content_type="rule"'
+
+
+def test_every_session_start_surface_states_the_conditional_retrieval():
+    """The push/pull asymmetry, one level in.
+
+    The tests above pin that a session PULLS the resident rules rather than
+    trusting the SessionStart push. This pins the same shape between the two
+    TIERS: an always-on rule is delivered, a conditional one is retrieved, and
+    a surface that states only the first leaves a session reading its loaded
+    set as the whole rulebook.
+
+    That reading is wrong in the direction that costs something. "Nothing was
+    pushed" and "no rule applies" are different claims, and only one of them
+    has been checked — the same asymmetry as #2198, now between tiers instead
+    of between channels.
+
+    It is also what made the always-on tier the only one that worked, on any
+    install rather than this one (rule 115). A rule nothing retrieves has to be
+    resident to bind at all, so every rule worth keeping becomes resident; and
+    a resident rule costs tokens in every session forever, so a rulebook that
+    only delivers cannot grow past what one session can hold. Retrieval is what
+    lifts that ceiling — and it only fires if something asks.
+    """
+    missing = []
+    for path in SESSION_START_SURFACES:
+        if RETRIEVE not in path.read_text():
+            missing.append(str(path.relative_to(ROOT)))
+    assert not missing, (
+        f"these surfaces state the always-on pull but never tell the agent to "
+        f"retrieve a conditional rule ({RETRIEVE}): {missing}. A session that "
+        f"reads its loaded set as the whole rulebook will act on \"I was not "
+        f"told\" as if it meant \"there is no rule\" (milestone 333 step 3)."
+    )
+
+
 def test_no_surface_names_the_push_without_stating_the_pull():
     """The exact shape #2497 took.
 

tests/test_rule_creation_asks_first.py

@@ -0,0 +1,120 @@
+"""Both rule-creation tools run the propose-then-approve loop (#3557).
+
+WHY THIS EXISTS
+
+Every other gate on `create_rule` and `create_project_rule` is about SHAPE:
+is this a rule or a process, is it one thing you could violate, is it general
+enough for a rulebook, is it a near-duplicate. All of those improve a rule
+someone has already decided to write. None of them asks the prior question —
+whether the person the rule will bind has agreed to be bound by it.
+
+That question belongs at the tool, because the tool is the last surface a
+caller reads before the write, and because the write is less reversible than
+it looks. The operator's yes is not merely consent; it is the one moment the
+rule is certainly IN FRONT of them. Afterwards it may not be again for
+months: a conditional rule is not read aloud at session start, and a
+project-scoped rule does not appear in an unfiltered `list_rules()` at all.
+The proposal IS the review, so there had better be one.
+
+WHY IT IS PHRASED AS A PRACTICE AND NOT A PROHIBITION
+
+The first cut of this guidance opened "NOT YOURS TO CALL UNPROMPTED." That is
+the wrong instrument, and the failure it invites is worse than the one it
+prevents: a caller reading a prohibition stops NOTICING rule-shaped things,
+rather than noticing them and asking. The wanted behaviour is more proposals,
+not fewer — spotting that something has hardened into a standing instruction
+is valuable work, and the only step that was ever missing came after it.
+
+So the docstrings describe what to DO: propose readily, state four things,
+close with a question the operator answers in one word. This test is written
+the same way — it asserts the parts of the loop are present, and has nothing
+to say about any wording that forbids.
+
+The fourth element — how the rule would be ENFORCED — is not ceremony. It is
+the part that sometimes dissolves the rule: a thing a test can assert should
+be that test, and a rule is what is left when nothing mechanical can hold it.
+A rulebook grows by default and shrinks only on purpose, so the question that
+prevents a rule earns more than any question that improves one's wording.
+
+WHAT THIS PINS, AND WHAT IT DOES NOT
+
+STRUCTURE, never wording — the same bargain the disambiguator guard (#3123)
+strikes next door. Each element matches a family of synonyms, so the prose
+stays free to be rewritten, reordered or sharpened; only DELETING one fails.
+Pinning phrasing would make every improvement a red build, and a test that
+punishes editing is a test someone deletes.
+
+It cannot tell whether an agent actually proposes. Nothing in a docstring
+can. It catches the regression that really happens: guidance tidied away in
+a later pass by someone who read it as throat-clearing in front of the Args.
+"""
+import pytest
+
+from tests.helpers import tool_doc as _doc
+
+# Both surfaces, because the one that needs it most is the one that looks
+# minor. A project rule is the least visible record the system can hold —
+# absent from an unfiltered list_rules(), and absent from session start too
+# whenever it is conditional — so the surface that writes one carries the
+# larger risk while reading as the smaller act.
+_SURFACES = [
+    ("scribe.mcp.tools.rulebooks", "create_rule"),
+    ("scribe.mcp.tools.rulebooks", "create_project_rule"),
+]
+
+# The loop, element by element, each as a family of ways to say it. A
+# docstring satisfies an element by containing ANY member — that is the room
+# left for rewriting. The families deliberately exclude bare words a
+# docstring would hold by accident ("why", "how", "reason", "rule"), which
+# would let the assertion pass on prose that says nothing of the kind.
+_ELEMENTS = {
+    "the invitation to propose": ("propose", "proposal"),
+    "the operator's approval": ("approve", "approval", "says yes", "a yes"),
+    "the rule's intent": ("intent", "what it changes about how work"),
+    "why it is being proposed now": (
+        "why now", "arose_from_id", "the incident", "prompted it",
+    ),
+    "how it would be enforced": ("enforc",),
+    "the answers offered back": ("as written", "talk about it", "discuss"),
+}
+
+
+@pytest.mark.parametrize(("module", "name"), _SURFACES)
+@pytest.mark.parametrize("element", sorted(_ELEMENTS))
+def test_a_rule_surface_carries_every_part_of_the_proposal_loop(
+    module, name, element
+):
+    """Each element of propose → state four things → ask survives."""
+    doc = _doc(module, name).lower()
+    assert any(token in doc for token in _ELEMENTS[element]), (
+        f"{name}'s docstring no longer mentions {element}. A caller reads "
+        f"this immediately before writing a rule that will bind every future "
+        f"session, and the proposal is the one moment that rule is certain to "
+        f"be seen by the operator. Say it in whatever words you like; this "
+        f"guard only checks it is still said. See create_rule's opening."
+    )
+
+
+@pytest.mark.parametrize(("module", "name"), _SURFACES)
+def test_the_proposal_loop_comes_before_the_parameter_contract(module, name):
+    """It has to be read to work, and the Args: block is where reading stops.
+
+    A caller who has decided to make the call skims down to the parameters.
+    Guidance parked below them — or folded into one argument's description —
+    arrives after the decision it was meant to inform, which is the same as
+    not being there.
+    """
+    doc = _doc(module, name).lower()
+    args_at = doc.find("args:")
+    assert args_at > 0, f"{name}'s docstring has no Args: block"
+    loop_at = min(
+        (doc.find(t) for t in _ELEMENTS["the invitation to propose"]
+         if doc.find(t) >= 0),
+        default=-1,
+    )
+    assert 0 <= loop_at < args_at, (
+        f"{name} introduces the proposal loop at or after its Args: block "
+        f"(loop {loop_at}, args {args_at}). Move it to the opening — a "
+        f"caller who has already decided to write the rule reads the "
+        f"parameters, not the prose under them."
+    )

tests/test_rule_usage_wiring.py

@@ -160,7 +160,15 @@ async def test_the_arm_searches_on_its_OWN_bar_not_the_code_one():
     kw = search.await_args.kwargs
     assert kw["threshold"] == 0.81, "the arm is still using the code threshold"
     assert kw["limit"] == pc.RULEHINT_LIMIT
-    assert kw["tier"] == "conditional"
+    # NO tier filter (#3702). The arms search every rule the caller owns,
+    # because "already in the session" is not the same as "in front of the
+    # reader at the moment it applies" — and relevance is the threshold's
+    # job, not a category's. If this assertion is failing because a tier
+    # argument came back, read the block above RULEHINT_LIMIT first: the
+    # filter may legitimately return, but only carrying a measured reason.
+    assert "tier" not in kw or kw["tier"] is None, (
+        "the arm is filtering the rule corpus by tier again"
+    )
 
 
 @pytest.mark.asyncio
@@ -702,3 +710,159 @@ def test_neither_rule_arm_logs_its_call_behind_a_results_guard():
         "the pre-tool arm returns before logging its call — a surface with no "
         "rows at all cannot be told apart from a hook that never fired"
     )
+
+
+# ── Suppression: which zeros were the ranker, which were repeats (#3497) ──
+#
+# Making the call log unconditional exposed a second ambiguity in the same row.
+# A zero-result rule call is two unrelated events: the ranker found nothing
+# above the bar, or it found only what this session already held. Only the
+# first says anything about the threshold, and a long session excludes its way
+# into the second — so without the split, the arm looks worse the longer it
+# runs correctly.
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_arm_reports_what_the_session_already_held():
+    log, rec = MagicMock(), MagicMock()
+    hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug")),
+            (0.70, fake_rule(id=157, title="A loop re-arms in a finally"))]
+    await _run_arm(hits, rec, retrieval_log=log, exclude_rule_ids=[156, 157])
+
+    row = next(c for c in log.call_args_list
+               if c.kwargs.get("source") == "write_path_rule")
+    assert row.kwargs["results"] == []
+    assert row.kwargs["suppressed"] == 2, (
+        "both hits were repeats, so this zero is not evidence about the bar"
+    )
+
+
+@pytest.mark.asyncio
+async def test_a_genuine_ranker_decline_reports_zero_suppression():
+    """Zero, not None. The arm filters in Python, so it always knows — and
+    'measured none' has to stay distinguishable from 'cannot measure'."""
+    log, rec = MagicMock(), MagicMock()
+    await _run_arm([], rec, retrieval_log=log)
+
+    row = next(c for c in log.call_args_list
+               if c.kwargs.get("source") == "write_path_rule")
+    assert row.kwargs["suppressed"] == 0
+    assert row.kwargs["suppressed"] is not None
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_reports_suppression_too():
+    log, rec = MagicMock(), MagicMock()
+    hits = [(0.75, fake_rule(id=161, title="Reach the forge through its MCP tools"))]
+    await _run_tool_arm(hits, rec, retrieval_log=log, exclude_rule_ids=[161])
+
+    assert log.call_args.kwargs["results"] == []
+    assert log.call_args.kwargs["suppressed"] == 1
+
+
+@pytest.mark.asyncio
+async def test_a_shown_hit_is_not_counted_as_suppressed():
+    """The obvious inverse, worth pinning: `suppressed` counts what was DROPPED,
+    not what came back. Off by one here and every zero row reads as a repeat."""
+    log, rec = MagicMock(), MagicMock()
+    hits = [(0.75, fake_rule(id=161, title="Reach the forge through its MCP tools")),
+            (0.70, fake_rule(id=12, title="Don't run a local stack unless asked"))]
+    out = await _run_tool_arm(hits, rec, retrieval_log=log, exclude_rule_ids=[12])
+
+    assert out["rule_ids"] == [161]
+    assert log.call_args.kwargs["suppressed"] == 1
+    assert len(log.call_args.kwargs["results"]) == 1
+
+
+# ── The identity that falsified this milestone (#3668) ─────────────────
+#
+# `rule_usage.surfaced` == `pre_tool_rule.cleared` + `write_path_rule.cleared`.
+# Milestone #379 was scoped on a reconstruction that put ~64% of ranked rule
+# surfacings as never reaching `rule_usage_events`. Five steps were planned
+# against it. One read of this identity — 17 = 17, then 39 = 39 on a second
+# window — falsified the whole thing: the gap was two counters that started
+# recording on different days, not a write path dropping rows.
+#
+# So the identity is not a nice-to-have. It is the cheapest true statement
+# available about this pair of tables, and its absence is what let a magnitude
+# that merely LOOKED wrong survive a code review and a five-step plan. An
+# identity that must hold exactly beats a magnitude that looks wrong.
+#
+# WHY THE ARM IS THE RIGHT PLACE TO PIN IT, and the readout is not. Inside an
+# arm, one `fresh` list feeds both recorders in one function, so the counts
+# cannot legitimately differ — at any limit. The readout-level form is weaker
+# than it looks: `cleared_threshold` counts CALLS that beat the bar while
+# `surfaced` counts RULES, and those coincide only while `RULEHINT_LIMIT` is 1.
+# Raise the limit and the readout identity breaks while nothing is wrong.
+# `RULEHINT_LIMIT` has already moved once (2 → 1, `2385100`), and that move is
+# half of why the original reconstruction misread its own numbers.
+#
+# Hence three hits below, where production currently returns at most one. The
+# test is deliberately in a state the limit does not permit today, because what
+# is being pinned is that the two recorders read the same list — not that the
+# list happens to be short.
+
+_THREE_HITS = [
+    (0.81, fake_rule(id=156, title="A wait with no deadline is a bug")),
+    (0.77, fake_rule(id=157, title="A loop re-arms in a finally")),
+    (0.74, fake_rule(id=161, title="Reach the forge through its MCP tools")),
+]
+
+_ARMS = [("write_path_rule", _run_arm), ("pre_tool_rule", _run_tool_arm)]
+
+
+def _both_ends(log, rec, source):
+    """What the two recorders said about one call, at the same grain.
+
+    Ids rather than counts. Equal counts drawn from different lists is a real
+    way for this to break — an off-by-one slice, or one recorder reading `hits`
+    where the other reads `fresh` in a window where the exclusion happened to
+    remove as many as it added — and a count comparison would call that agreement.
+    """
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == source]
+    assert len(rows) == 1, (
+        f"expected exactly one {source} call row, got {len(rows)} — the "
+        f"identity is per call and cannot be read across several"
+    )
+    logged = [rule.id for _score, rule in rows[0].kwargs["results"]]
+    surfaced = [
+        rid
+        for c in rec.call_args_list if c.kwargs.get("source") == source
+        for rid in c.kwargs["rule_ids"]
+    ]
+    return logged, surfaced
+
+
+@pytest.mark.parametrize(("source", "run"), _ARMS, ids=["write_path", "pre_tool"])
+@pytest.mark.parametrize(
+    ("excluded", "expected"),
+    [([], 3), ([157], 2), ([156, 157, 161], 0)],
+    ids=["nothing-held", "one-already-held", "all-already-held"],
+)
+@pytest.mark.asyncio
+async def test_both_recorders_report_the_same_rules_for_one_call(
+    source, run, excluded, expected
+):
+    """One list, two tables, no room to disagree.
+
+    The middle case is the one that discriminates. With nothing excluded both
+    recorders see the same three rules however wrongly they are wired, so an
+    arm logging `hits` to the call log and `fresh` to the surfacing log passes
+    that case and fails this one — and logging `hits` is exactly the divergence
+    that would manufacture an apparent write loss out of a correct system.
+    """
+    log, rec = MagicMock(), MagicMock()
+    await run(list(_THREE_HITS), rec, retrieval_log=log, exclude_rule_ids=excluded)
+
+    logged, surfaced = _both_ends(log, rec, source)
+    assert surfaced == logged, (
+        f"{source} told its two tables different stories about one call: the "
+        f"call log recorded {logged} and the surfacing log recorded {surfaced}. "
+        f"Both come from `fresh`, in one function, so any difference is a bug "
+        f"in the wiring — and it is the shape that reads as a lost write when "
+        f"the two tables are later compared in aggregate (#3668)."
+    )
+    assert len(logged) == expected, (
+        "the fixture stopped exercising what it claims to; check the exclusion "
+        "filter still runs before both recorders"
+    )

tests/test_services_retrieval_telemetry.py

@@ -57,6 +57,68 @@ def test_build_payload_rounds_scores_to_5dp():
     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, cleared, p10, p50, p90, min, max, avg_n, dur,
+    # measured, supp_calls, supp_zero
+    unmeasured = _bucket([326, 114, 212, 0.6, 0.68, 0.77, 0.55, 0.85, 1.7, 130.9,
+                          0, 0, 0])
+    assert unmeasured["suppression"] is None
+
+    measured = _bucket([35, 34, 1, 0.75, 0.75, 0.75, 0.75, 0.75, 0.03, 51.9,
+                        35, 9, 9])
+    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."""
@@ -576,3 +638,212 @@ async def test_ambient_alone_reports_no_ratio(_dispose_engine):
         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()