Merge pull request 'feat(rules): rule creation becomes propose-then-approve (#3557)' (#141) from dev into main

The first cut opened "NOT YOURS TO CALL UNPROMPTED", and that is the wrong
instrument. A caller reading a prohibition stops NOTICING rule-shaped things
rather than noticing them and asking — which trades a small failure for a
larger one. The wanted behaviour is more proposals, not fewer.

So both docstrings now describe the practice: propose readily, state the
four things, and close with a question the operator answers in one word —
approve it as written / let's talk about it / no. Named options where the
interface has them, three written-out options where it does not.

"Approve it as written" is what makes element 1 load-bearing: they approved
TEXT, so that text is stored verbatim. "Let's talk about it" is framed as
the expected answer rather than a setback. "No" routes the observation to
create_note, which records without binding.

The argument for asking is also better than consent. The operator's yes is
the one moment the rule is certainly in front of them: afterwards a
conditional rule is not read aloud at session start, and a project rule is
absent from an unfiltered list_rules(). The proposal IS the review.

Guard gains the answers-offered-back element and drops the wording that
forbade; its header records why the framing changed, so the prohibition does
not get reintroduced as a tidy-up.

Co-Authored-By: Claude Opus 5 (1M context) <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.02.0438",
+  "version": "2026.09.04.0140",
   "author": {
     "name": "Bryan Van Deusen"
   },

plugin/hooks/hooks.json

@@ -33,6 +33,15 @@
             "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_prior_art.sh\""
           }
         ]
+      },
+      {
+        "matcher": "Bash",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_tool_rules.sh\""
+          }
+        ]
       }
     ],
     "PostToolUse": [

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/hooks/scribe_tool_rules.sh

@@ -0,0 +1,116 @@
+#!/usr/bin/env bash
+# Scribe — PreToolUse rule arm for ACTIONS (#3476).
+#
+# The sibling of scribe_prior_art.sh. That hook is registered on Write|Edit and
+# asks "what is recorded about the file being written". This one asks "does a
+# standing rule speak to the command about to be run" — the question nothing
+# could ask before, and the reason every rule about which tool to reach for had
+# to live in the always-on preload instead.
+#
+# WHY A HOOK AND NOT AN INSTRUCTION. A reflex generates no query (note #3089):
+# you reach for `curl` confidently, with no moment of doubt, so a surface that
+# waits to be asked never fires. Here nothing is asked — the tool call IS the
+# query, and the reflex has to become a tool call before it can do anything.
+#
+# SILENT ON OUTAGE, deliberately, unlike the prior-art hook. A write is
+# occasional; a Bash call is not, and an "instance did not answer" line before
+# every command is the noise that gets a channel muted. scribe_prior_art.sh
+# still speaks for both when the instance is down.
+#
+# Env:
+#   SCRIBE_URL / SCRIBE_TOKEN override for the settings.json dogfooding path.
+
+command -v jq   >/dev/null 2>&1 || exit 0
+command -v curl >/dev/null 2>&1 || exit 0
+
+# PreToolUse delivers { session_id, cwd, tool_name, tool_input: {...}, ... }
+event=$(cat 2>/dev/null || true)
+tool_name=$(printf '%s' "$event" | jq -r '.tool_name // empty' 2>/dev/null) || exit 0
+session_id=$(printf '%s' "$event" | jq -r '.session_id // empty' 2>/dev/null) || session_id=""
+event_cwd=$(printf '%s' "$event" | jq -r '.cwd // empty' 2>/dev/null) || event_cwd=""
+
+[ -n "$tool_name" ] || exit 0
+
+# The action, as text. `.command` is Bash's field; the fallbacks let the matcher
+# in hooks.json widen to other tools without this script changing — which is the
+# whole reason the server side takes a name and a string rather than a schema.
+command_text=$(printf '%s' "$event" | jq -r '
+  .tool_input.command //
+  .tool_input.url //
+  .tool_input.prompt //
+  empty' 2>/dev/null) || command_text=""
+
+[ -n "$command_text" ] || exit 0
+
+# shellcheck source=plugin/hooks/scribe_defs.sh
+. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
+
+# scribe_config, not a hand-rolled pair of parameter expansions: it also treats
+# an UNEXPANDED `${...}` placeholder as unset, which would otherwise be sent as
+# a garbage Bearer token and 401 on every call (#2198's class).
+scribe_config || exit 0
+
+# Bounded before encoding: a heredoc or a pasted script can be enormous, and
+# the verb and its target — the part a rule is about — sit at the front. The
+# server bounds it again; this keeps a huge payload off the wire in the first
+# place. `head -c`, never `cut -c`: cut truncates each LINE and caps nothing.
+command_text=$(printf '%s' "$command_text" | head -c 2000)
+
+# -sRr, never -rR: jq -R without -s reads LINE BY LINE, so a multi-line command
+# would encode per line and join with raw newlines — an invalid URL.
+cmd_enc=$(printf '%s' "$command_text" | jq -sRr '@uri' 2>/dev/null) || exit 0
+tool_enc=$(printf '%s' "$tool_name" | jq -sRr '@uri' 2>/dev/null) || exit 0
+
+repo_q=""
+lookup_dir=${event_cwd:-${CLAUDE_PROJECT_DIR:-$PWD}}
+repo_remote=$(git -C "$lookup_dir" remote get-url origin 2>/dev/null || true)
+if [ -n "$repo_remote" ]; then
+  repo_enc=$(printf '%s' "$repo_remote" | jq -sRr '@uri' 2>/dev/null) || repo_enc=""
+  [ -n "$repo_enc" ] && repo_q="&repo=${repo_enc}"
+fi
+
+# THE SHARED SESSION LEDGER, and the thing most worth getting right here.
+#
+# scribe_prior_art.sh keeps the rules it has already named in
+# <state>/<sid>.rules.ids and passes them as exclude_rule_ids. This hook reads
+# and appends to that SAME file rather than keeping its own: two ledgers would
+# mean a rule named by one arm gets re-offered by the other, and the hint that
+# fires most often is exactly the one that must not repeat itself.
+#
+# The directory keeps the prior-art name on purpose — renaming it would orphan
+# every live session's state for a cosmetic gain.
+state_dir="${TMPDIR:-/tmp}/scribe-priorart"
+mkdir -p "$state_dir" 2>/dev/null || true
+rulefile=""
+rule_exclude_q=""
+if [ -n "$session_id" ]; then
+  safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
+  rulefile="$state_dir/${safe_sid}.rules.ids"
+  if [ -f "$rulefile" ]; then
+    rule_seen=$(tr '\n' ',' < "$rulefile" 2>/dev/null | sed 's/,$//')
+    [ -n "$rule_seen" ] && rule_exclude_q="&exclude_rule_ids=${rule_seen}"
+  fi
+fi
+
+# `|| exit 0` here, unlike the prior-art hook: there is no local arm whose
+# finding would be discarded, and an outage line before every command is worse
+# than silence. See the header.
+body=$(curl -fsS --max-time 5 \
+  -H "Authorization: Bearer ${token}" \
+  "${url%/}/api/plugin/tool-rules?tool=${tool_enc}&command=${cmd_enc}${repo_q}${rule_exclude_q}" 2>/dev/null) || exit 0
+
+context=$(printf '%s' "$body" | jq -r '.context // empty' 2>/dev/null) || exit 0
+[ -n "$context" ] || exit 0
+
+# Remember what was named so it is not repeated this session.
+if [ -n "$rulefile" ]; then
+  printf '%s' "$body" | jq -r '(.rule_ids // [])[]?' 2>/dev/null >> "$rulefile" || true
+fi
+
+jq -cn --arg ctx "$context" '{
+  hookSpecificOutput: {
+    hookEventName: "PreToolUse",
+    additionalContext: $ctx
+  }
+}' 2>/dev/null || true
+exit 0

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.

scripts/check_plugin.py

@@ -305,6 +305,16 @@ SMOKE_EVENTS: dict[str, str] = {
          "tool_input": {"file_path": "src/x.py",
                         "new_string": f"def {_ABSENT_SYM}():\n    pass\n"}}
     ),
+    # The pre-tool rule arm (#3476). A real Bash call, and one whose whole
+    # point is that it looks harmless: reaching for curl against the forge API
+    # is the reflex the arm exists to catch. With no instance it must stay
+    # SILENT — it is deliberately not an OUTAGE_SPEAKER, because a Bash call is
+    # not occasional and an outage line before every command gets the channel
+    # muted.
+    "scribe_tool_rules.sh": json.dumps(
+        {"session_id": "smoke", "cwd": ".", "tool_name": "Bash",
+         "tool_input": {"command": "curl -s https://example.invalid/api/v1/runs"}}
+    ),
     "scribe_sync_processes.sh": json.dumps({"source": "startup"}),
     "scribe_session_context.sh": json.dumps({"source": "startup"}),
     # The after-write hook (#2901) diffs the working tree; on CI's clean

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/milestones.py

@@ -57,7 +57,7 @@ async def get_milestone(milestone_id: int) -> dict:
     return {
         "milestone": out,
         "steps": [t.to_dict() for t in steps],
-        **rulebooks_svc.rules_payload(applicable),
+        **rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_milestone"),
     }
 
 

src/scribe/mcp/tools/projects.py

@@ -207,7 +207,7 @@ async def enter_project(project_id: int) -> dict:
         ],
         "design_system": design_system,
         "milestone_summary": milestone_summary,
-        **rulebooks_svc.rules_payload(applicable),
+        **rulebooks_svc.rules_payload(applicable, user_id=uid, source="enter_project"),
         "open_tasks": [
             {
                 "id": t.id, "title": t.title, "status": t.status,
@@ -251,7 +251,7 @@ async def get_project(project_id: int) -> dict:
     applicable = await rulebooks_svc.get_applicable_rules(
         project_id=project_id, user_id=uid,
     )
-    data.update(rulebooks_svc.rules_payload(applicable))
+    data.update(rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_project"))
     return data
 
 

src/scribe/mcp/tools/rulebooks.py

@@ -18,7 +18,7 @@ from scribe.mcp._context import current_user_id
 from scribe.services import dedup as dedup_svc
 from scribe.services import rulebooks as rulebooks_svc
 from scribe.services import trash as trash_svc
-from scribe.services.rule_usage import record_rule_pulled
+from scribe.services.rule_usage import record_rule_pulled, record_rule_surfaced
 
 
 # ── Rulebook CRUD ───────────────────────────────────────────────────────
@@ -265,6 +265,15 @@ async def list_always_on_rules(project_id: int = 0) -> dict:
     """
     uid = current_user_id()
     rules = await rulebooks_svc.list_always_on_rules(uid, project_id=project_id)
+    # AMBIENT source: the resident set, handed over whole. No ranker chose
+    # these, so they must not land in the pull-through numerator's denominator
+    # — but they must land SOMEWHERE, or the largest rule surface in the
+    # product stays the one surface its own scoreboard cannot see (#3473).
+    record_rule_surfaced(
+        user_id=uid,
+        rule_ids=[r.id for r in rules],
+        source="list_always_on_rules",
+    )
     return {
         "rules": [_rule_summary(r) for r in rules],
         "total": len(rules),
@@ -306,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 —
@@ -424,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
@@ -197,17 +211,31 @@ It is an UPPER BOUND per surface: a pull records the door it came
     query failed while the rest of the readout stood.
 
     `rule_usage` — the same question for RULES, from `rule_usage_events`:
-    `surfaced`, `pulled` split into `pulled_by_agent` / `pulled_by_human`, the
-    distinct-rule counts, and `pull_through` on the same definition (agent
-    pulls over surfacings).
+    `surfaced` and `ambient`, `pulled` split into `pulled_by_agent` /
+    `pulled_by_human`, the distinct-rule counts, and `pull_through` on the same
+    definition (agent pulls over RANKED surfacings).
 
     A SEPARATE BLOCK, not folded into `usage`, and reading it as one number
     with that is the mistake to avoid. The corpora differ by orders of
     magnitude — a few dozen eligible rules against thousands of notes — so a
     blended ratio would be the note ratio with noise on it and would hide the
-    rule arm entirely. It also has no `ambient` key, because nothing surfaces a
-    rule un-ranked: `list_always_on_rules` and `enter_project` hand over rules
-    wholesale but emit no event, so there is no ambient class to separate.
+    rule arm entirely.
+
+    `surfaced` VS `ambient` IS THE READING THAT MATTERS HERE. `surfaced` counts
+    rules a ranker chose — today only the write-path arm — and those are claims
+    a pull can settle. `ambient` counts BULK DELIVERIES: the SessionStart
+    preload, `list_always_on_rules`, and the `rules_payload` surfaces
+    (`enter_project`, `get_project`, `get_milestone`, `start_planning`,
+    `get_task`), which hand over the whole applicable set at once with nobody
+    choosing anything. A large `ambient` says the resident set is big and
+    arrives often — never that it is useful, and never that it is read.
+
+    `pull_through` therefore divides by `surfaced` alone. Fold the preload in
+    and growing the always-on set would depress the arm's measured precision
+    while trimming it would flatter it, for reasons having nothing to do with
+    the arm. To judge the PRELOAD instead, compare `ambient` against pulls of
+    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

src/scribe/mcp/tools/tasks.py

@@ -103,7 +103,7 @@ async def get_task(task_id: int) -> dict:
         applicable = await rulebooks_svc.get_applicable_rules(
             project_id=note.project_id, user_id=uid,
         )
-        data.update(rulebooks_svc.rules_payload(applicable))
+        data.update(rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_task"))
     data.update(await access_svc.describe_provenance(uid, note))
     # Same reasoning as get_note's record_pulled, and this is the tool where it
     # matters MOST: auto-inject ranks kind-blind over a corpus that is

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/routes/plugin.py

@@ -101,6 +101,46 @@ async def autoinject_retrieve():
     return jsonify(result)
 
 
+@plugin_bp.get("/tool-rules")
+@login_required
+async def pre_tool_rules():
+    """Standing rules for the plugin's PreToolUse hook on ACTIONS (#3476).
+
+    Answers "does a recorded rule speak to the command about to be run?" — the
+    sibling of /prior-art, which can only answer that question about a code
+    write. Rules about which tool to reach for (don't curl the forge, don't
+    stand up a stack, don't run the suite locally) had no retrieval surface at
+    all before this, which is why they all had to live in the resident preload.
+
+    Titles + trigger only, never the statement: the hint says a rule may apply
+    and hands over `get_rule(id)`. One rule at most (RULEHINT_LIMIT), and empty
+    most of the time.
+
+    Query:
+        tool (str)              — the tool about to run, e.g. `Bash`. Used in
+                                  the hint's wording, not in the search: a
+                                  rule is about the action, not the harness.
+        command (str)           — the command about to run; the semantic query.
+                                  Absent or blank → empty, no search.
+        repo (optional)         — working repo remote, resolved to the bound
+                                  project exactly as /retrieve and /prior-art.
+        exclude_rule_ids (opt)  — comma-separated rule ids already surfaced
+                                  this session. SHARED with /prior-art's
+                                  ledger on purpose: one session keeps one
+                                  list, so a rule named by either arm is not
+                                  re-offered by the other.
+    """
+    tool = (request.args.get("tool") or "tool").strip()
+    command = request.args.get("command") or ""
+    project_id, _repo, _unbound = await _project_scope()
+    exclude_rule_ids = _int_list(request.args.get("exclude_rule_ids"))
+    result = await plugin_ctx_svc.build_tool_rule_hint(
+        g.user.id, tool, command,
+        project_id=project_id, exclude_rule_ids=exclude_rule_ids,
+    )
+    return jsonify(result)
+
+
 @plugin_bp.get("/prior-art")
 @login_required
 async def write_path_prior_art():

src/scribe/services/planning.py

@@ -60,7 +60,7 @@ async def start_planning(user_id: int, project_id: int, title: str) -> dict:
 
     return {
         "milestone": milestone.to_dict(),
-        **rulebooks_svc.rules_payload(applicable),
+        **rulebooks_svc.rules_payload(applicable, user_id=user_id, source="start_planning"),
         "project_goal": getattr(project, "goal", "") or "",
         "open_task_count": open_count,
     }

src/scribe/services/plugin_context.py

@@ -140,6 +140,15 @@ RULEHINT_DEFAULT_THRESHOLD = 0.72
 # adds a way to misconfigure the surface (rule 25 cuts both ways).
 RULEHINT_LIMIT = 1
 
+# 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
+# ITS TARGET sit at the front — `curl https://git.fabledsword.com/api/...`,
+# `docker compose up`, `git checkout -b` — and that head is the whole signal.
+# Sending the tail as well would push it out of a 512-token window and let a
+# heredoc's prose decide the match.
+_TOOL_QUERY_CHARS = 400
+
 # Minimum SUBSTANCE (non-whitespace chars) a payload must carry before the
 # semantic arm will run at all — the cheap half of the operator's #89 idea
 # ("a sliding scale between number of characters and semantic threshold").
@@ -1209,24 +1218,48 @@ async def build_write_path_hint(
                   "does not apply; it is not in this session's loaded set."
             )
             rule_ids.append(rule.id)
+        # TWO tables, and the split is not arbitrary. retrieval_logs is one
+        # row per CALL, keyed on the score distribution a threshold is tuned
+        # from. rule_usage_events is one row per RULE per event, which is the
+        # grain "was this hint ever acted on" needs and the grain a JSONB
+        # result_ids array cannot be indexed at.
+        #
+        # This comment used to say rule ids had nowhere to go — that
+        # note_usage_events remaps ids on restore, so a rule id there would
+        # return attached to whatever note took that number. That is still
+        # true of the NOTE table, and it is exactly why rule_usage_events is
+        # its own (milestone 333 step 1). The gap it described is closed.
+        #
+        # THE CALL LOG IS UNCONDITIONAL; THE SURFACING LOG IS NOT, and the
+        # asymmetry is the correction #3497 exists to make. Both used to sit
+        # inside an `if fresh:`, which is how this arm came to report
+        # `zero_result_calls: 0` and `cleared_threshold: 133/133` — not a
+        # perfectly tuned surface but one structurally unable to record its
+        # own misses. #3311 read that artifact as a measurement and a whole
+        # milestone was scoped on it. A call that found nothing is the ONLY
+        # evidence a threshold is set too high, and it is the row every note
+        # surface has always written (write_path: 421 zeroes of 613 calls;
+        # auto_inject: 114 of 326). A SURFACING is different in kind: nothing
+        # was shown, so no such event occurred, and its log stays guarded.
+        #
+        # `results=fresh`, not `hits`: the note arms pass their exclusions
+        # INTO semantic_search_notes, so what they log is already
+        # post-exclusion. semantic_search_rules takes no such parameter and
+        # this filter is where the equivalent happens — logging `hits` would
+        # quietly make this row mean something other than every other row in
+        # the same readout.
+        record_retrieval(
+            user_id=user_id, source="write_path_rule", query=code or path,
+            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:
-            # TWO tables, and the split is not arbitrary. retrieval_logs is one
-            # row per CALL, keyed on the score distribution a threshold is
-            # tuned from. rule_usage_events is one row per RULE per event,
-            # which is the grain "was this hint ever acted on" needs and the
-            # grain a JSONB result_ids array cannot be indexed at.
-            #
-            # This comment used to say rule ids had nowhere to go — that
-            # note_usage_events remaps ids on restore, so a rule id there would
-            # return attached to whatever note took that number. That is still
-            # true of the NOTE table, and it is exactly why rule_usage_events
-            # is its own (milestone 333 step 1). The gap it described is closed.
-            record_retrieval(
-                user_id=user_id, source="write_path_rule", query=code or path,
-                threshold=cfg["rule_threshold"], limit=RULEHINT_LIMIT,
-                project_id=project_id,
-                is_task=None, results=fresh, duration_ms=rule_ms,
-            )
             # `rule_ids` is `fresh`, i.e. AFTER exclude_rule_ids. A rule the
             # session already holds was considered and not shown, and counting
             # it would inflate the denominator with claims the agent never saw
@@ -1250,6 +1283,114 @@ async def build_write_path_hint(
     }
 
 
+async def build_tool_rule_hint(
+    user_id: int,
+    tool_name: str,
+    command: str,
+    *,
+    project_id: int = 0,
+    exclude_rule_ids: list[int] | None = None,
+) -> dict:
+    """Standing rules that may apply to the ACTION about to be taken (#3476).
+
+    The sibling of the write-path rule arm, and the surface that was missing.
+    That arm is keyed on `code or path`, so a rule can only be retrieved at the
+    moment of a code WRITE. Every rule about which tool to reach for — don't
+    curl the forge, don't stand up a stack, don't run the suite locally, don't
+    branch — was therefore unreachable at the moment it mattered, and residency
+    in the always-on preload was the only surface it had.
+
+    WHY A MECHANICAL TRIGGER AND NOT AN INSTRUCTION. Note #3089's finding is
+    that a reflex generates no query: you reach for `curl` confidently, with no
+    moment of doubt, so any surface that waits to be asked never fires. Here
+    nothing has to be asked — the tool call IS the query, and the reflex has to
+    become a tool call before it can do anything.
+
+    Deliberately TOOL-AGNOSTIC: takes a name and a string. The hook decides
+    which tools it watches, so widening the matcher is a `hooks.json` edit with
+    no change here.
+
+    CONDITIONAL ONLY, exactly as the write-path arm — an always-on rule is
+    already resident and repeating it is noise. That filter is also the
+    transition this arm exists to enable: re-tier a rule to `conditional` and
+    it starts arriving here instead of in every session's preamble.
+
+    Fails open and returns an empty context on any error: a recall aid may
+    never break the operator's action.
+    """
+    out: dict = {"context": "", "rule_ids": []}
+    command = (command or "").strip()
+    if not command:
+        return out
+
+    try:
+        cfg = await get_writepath_config(user_id)
+        if not cfg.get("enabled"):
+            return out
+
+        # The command text is the query. A long heredoc or a pasted script
+        # would otherwise push the meaningful head of the command out of the
+        # embedding window, so it is bounded — the verb and its target sit at
+        # the front, which is the part a rule is about.
+        query = command[:_TOOL_QUERY_CHARS]
+
+        t0 = time.perf_counter()
+        hits = await semantic_search_rules(
+            user_id, query, limit=RULEHINT_LIMIT,
+            threshold=cfg["rule_threshold"], tier="conditional",
+        )
+        duration_ms = (time.perf_counter() - t0) * 1000.0
+
+        already = set(exclude_rule_ids or [])
+        fresh = [(score, rule) for score, rule in hits if rule.id not in already]
+
+        # Logged BEFORE the early return, for the reason spelled out at length
+        # on the write-path arm above: a call that found nothing is the only
+        # evidence a threshold is too high, and an arm that logs only the calls
+        # it liked reports a flawless clear-rate however badly it is tuned.
+        # This arm shipped with the same defect inherited from its sibling, and
+        # it mattered more here — a surface with no rows at all cannot be told
+        # apart from a hook that never fired, which is precisely the silent
+        # failure the arm was built to stop.
+        record_retrieval(
+            user_id=user_id, source="pre_tool_rule", query=query,
+            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
+
+        lines: list[str] = []
+        rule_ids: list[int] = []
+        for _score, rule in fresh:
+            trigger = (rule.when_to_apply or "").strip()
+            lines.append(
+                f"Standing rule that may apply to this {tool_name} call — "
+                f"“{rule.title}”"
+                + (f" ({trigger})" if trigger else "")
+                + f". Read it with get_rule({rule.id}) before deciding it "
+                  "does not apply; it is not in this session's loaded set."
+            )
+            rule_ids.append(rule.id)
+
+        # RANKED, not ambient: this arm chose what it showed, so a pull can
+        # settle whether the choice was any good. `rule_usage.RANKED_SOURCES`
+        # carries the same name.
+        record_rule_surfaced(
+            user_id=user_id, rule_ids=rule_ids, source="pre_tool_rule",
+        )
+        out["context"] = "\n".join(lines)
+        out["rule_ids"] = rule_ids
+    except Exception:
+        logger.debug("pre-tool rule arm failed", exc_info=True)
+    return out
+
+
 def _derive_line(path: str, derive: list[dict]) -> str:
     """The ledger's word on the names being written (#2900): a duplicate
     family to derive, or a canon to reuse — said at the write."""
@@ -1375,6 +1516,24 @@ async def build_session_context(
     # exclusion (milestone 297) takes a rulebook out of this block, and is
     # named below so the departure is visible rather than silent.
     rules = await rulebooks_svc.list_always_on_rules(user_id, project_id=project_id)
+    # AMBIENT source, and the one that matters most: this is the preload — the
+    # block every session opens with, chosen by nobody, paid for every turn.
+    #
+    # It emitted nothing until 2026-09-03, which made the resident set's cost
+    # certain and its usefulness unfalsifiable at the same time (#3473). Note
+    # #3089 is the argument this measurement finally lets someone test: that a
+    # rule arriving with thirty others, none of them relevant, is read as
+    # preamble rather than as a claim — so presence is not surfacing, and a
+    # tier-1 set can grow without anybody noticing it stopped working.
+    #
+    # Recorded even when the hook truncates the block below: the rules WERE
+    # delivered, and counting only the untruncated ones would quietly shrink
+    # the denominator exactly where the set is too big to read.
+    record_rule_surfaced(
+        user_id=user_id,
+        rule_ids=[r.id for r in rules],
+        source="session_start",
+    )
     excluded = (
         await rulebooks_svc.excluded_always_on_rulebooks(user_id, project_id)
         if project_id else []

src/scribe/services/retrieval_telemetry.py

@@ -30,6 +30,7 @@ from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
 from scribe.models.rule_usage import PULLED as RULE_PULLED
 from scribe.models.rule_usage import SURFACED as RULE_SURFACED
 from scribe.models.rule_usage import RuleUsageEvent
+from scribe.services.rule_usage import is_ambient
 from scribe.models.retrieval_log import RetrievalLog
 
 logger = logging.getLogger(__name__)
@@ -54,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}
@@ -75,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,
@@ -114,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.
 
@@ -139,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)
@@ -165,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
@@ -177,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),
@@ -239,6 +266,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())
@@ -265,6 +300,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,
@@ -428,10 +466,15 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
                         .group_by(RuleUsageEvent.event, RuleUsageEvent.source)
                     )
                 ).all()
-                # No AMBIENT exclusion here, unlike the note twin: nothing
-                # surfaces a rule un-ranked yet. `list_always_on_rules` and
-                # `enter_project` deliver rules wholesale but emit no event, so
-                # there is no ambient class to subtract (milestone 333 step 1).
+                # 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.
+                #
+                # `distinct_rules_surfaced` deliberately counts BOTH classes. It
+                # answers "how many distinct rules did this install put in front
+                # of an agent at all", which is the denominator for dead weight
+                # — and a rule delivered by the preload a hundred times and
+                # never opened is the most important case that question has.
                 distinct_rules_surfaced = (
                     await session.execute(
                         select(func.count(func.distinct(RuleUsageEvent.rule_id)))
@@ -545,11 +588,18 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
     # been comparing across windows, without telling them it now measures
     # something else.
     #
-    # No `ambient` key, unlike its twin. Nothing surfaces a rule un-ranked yet;
-    # the absence is a fact about the data rather than an oversight, and it
-    # returns the moment a bulk loader starts emitting.
+    # `ambient` now carries the bulk deliveries — the SessionStart preload,
+    # `list_always_on_rules`, and every `rules_payload` surface (#3473). Before
+    # they emitted, this block had no ambient key and said the absence was a
+    # fact about the data. It was, and it was also the thing that made the
+    # always-on set impossible to judge: the largest rule surface in the
+    # product was the one surface its own scoreboard could not see.
+    #
+    # READ THE TWO SEPARATELY, ALWAYS. `surfaced` is a claim a ranker made and
+    # a pull can settle. `ambient` is a delivery nobody chose, so a high count
+    # says the set is large and resident, never that it is useful.
     rule_usage = {
-        "surfaced": 0,
+        "surfaced": 0, "ambient": 0,
         "pulled": 0, "pulled_by_agent": 0, "pulled_by_human": 0,
         "distinct_rules_surfaced": int(distinct_rules_surfaced or 0),
         "distinct_rules_pulled": int(distinct_rules_pulled or 0),
@@ -565,7 +615,14 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
         for event, source, n in rule_rows:
             n = int(n)
             if event == RULE_SURFACED:
-                rule_usage["surfaced"] += n
+                # One definition of ranked-vs-ambient, imported rather than
+                # restated — the per-rule badge readout reads the same
+                # predicate, and two spellings of "what counts as surfaced" is
+                # precisely the uneven wiring #3246 found across this system.
+                if is_ambient(source):
+                    rule_usage["ambient"] += n
+                else:
+                    rule_usage["surfaced"] += n
             elif event == RULE_PULLED:
                 rule_usage["pulled"] += n
                 # Same split, and it carries MORE weight here than for notes.
@@ -582,6 +639,15 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
     # empty numerator AND denominator that is a claim the data does not
     # support, and it is the reading that would make a brand-new install look
     # like a broken one.
+    #
+    # RANKED SURFACINGS ONLY in the denominator, and this is the load-bearing
+    # line of the whole change. Pull-through asks "was that hint any use", and
+    # only a surface that CHOSE what it showed can be judged by it. Folding the
+    # preload in would divide the same pulls by a number that grows with every
+    # session and every rule added to the resident set — so enlarging the
+    # always-on set would DEPRESS the arm's measured precision, and trimming it
+    # would flatter it, neither for any reason to do with the arm. The ambient
+    # count sits beside it, unaveraged, and is read as size rather than skill.
     rule_usage["pull_through"] = (
         round(rule_usage["pulled_by_agent"] / rule_usage["surfaced"], 4)
         if rule_usage["surfaced"] else None

src/scribe/services/rule_usage.py

@@ -12,11 +12,23 @@ Two event streams, deliberately independent:
 
 WHY THIS ARM AND NOT ANOTHER. Every other surface declines most of the time —
 `write_path` returns nothing on 78% of calls, `reuse_slot` on 79%, auto-inject
-on 39%. The rule arm has never once returned nothing (#3311). That is either a
-perfectly tuned surface or a bar it cannot fail to clear, and `retrieval_logs`
-cannot tell the two apart: it records what the ranker scored, never whether the
-hint was any use. The ratio these two streams produce is the missing half, and
-without it any threshold change is a number picked off a histogram.
+on 39%. The rule arm APPEARED never to have returned nothing (#3311), and this
+docstring used to put that forward as the puzzle worth measuring: "either a
+perfectly tuned surface or a bar it cannot fail to clear".
+
+It was neither, and the correction belongs here rather than being quietly
+deleted. The arm wrote its `retrieval_logs` row only on calls that FOUND
+something (#3497), so `zero_result_calls` sat at 0 and `cleared_threshold` at
+`calls` because of the shape of the code — at any threshold whatsoever. A
+statistic that could not vary was read as a finding about the corpus. It is the
+#2663 failure mode one level up: there the broken readout was a zero, here it
+was a hundred percent, which is far better camouflage.
+
+The reason to measure this arm survives the correction, and is stronger for it.
+`retrieval_logs` records what the ranker scored, never whether the hint was any
+use, so even an honest clear-rate would not settle the question. The ratio these
+two streams produce is the missing half, and without it any threshold change is
+a number picked off a histogram.
 
 Design notes, mirroring `note_usage`:
   - Writes are fire-and-forget through `background.spawn`, so telemetry never
@@ -30,23 +42,46 @@ Design notes, mirroring `note_usage`:
   - Reads (`usage_for_rules`) are awaited and aggregated in one round-trip for
     a whole page, never per row.
 
-NO AMBIENT BUCKET, YET — and that is a decision, not an omission. The note twin
-splits ranked surfacings from ambient ones because `enter_project` and the
-skill sync put records in front of the agent without choosing them, and
-counting those as surfacings makes recency read as popularity (#2477). Rules
-have the same shape of problem waiting: `list_always_on_rules` and
-`enter_project` load rules wholesale on every session. They do not emit here
-today, so there is nothing to bucket, and an empty `AMBIENT_SOURCES` would be
-machinery pretending to a distinction the data does not yet contain. When a
-bulk surface starts emitting, the split is a readout-level change — a tuple and
-a `case()`, exactly as in the twin — and needs no migration. Keep it that way:
-`source` stays granular so the choice remains available.
+AMBIENT VS RANKED. The note twin splits ranked surfacings from ambient ones
+because `enter_project` and the skill sync put records in front of the agent
+without choosing them, and counting those as surfacings makes recency read as
+popularity (#2477). Rules have exactly that shape: the SessionStart preload,
+`list_always_on_rules`, and every `rules_payload` surface hand over the whole
+applicable set at once, chosen by nobody.
+
+Until 2026-09-03 those bulk surfaces emitted nothing, and this module said so —
+"an empty `AMBIENT_SOURCES` would be machinery pretending to a distinction the
+data does not yet contain". True as far as it went, but it had a consequence
+worth naming, because it is the reason the bucket exists now: the always-on
+set's token cost was certain and its usefulness was UNFALSIFIABLE, permanently
+and by construction. The one surface whose value was actually in question was
+the one surface exempt from the scoreboard that judges every other.
+
+They emit now. The split is the readout-level change the old note promised — a
+`case()`, no migration, because `event` and `source` are plain Text with no
+CHECK constraint. `source` stays granular so a reader can still tell the
+preload from `enter_project` from the ranked arm.
+
+WHY THIS NAMES THE RANKED SOURCES AND THE TWIN NAMES THE AMBIENT ONES. A
+deliberate divergence, on the failure mode rather than on symmetry. Both shapes
+fail silently when someone adds a surface and forgets the list, so the question
+is which list changes more often — and here it is emphatically the ambient one:
+there are TWO ranked rule sources (the write-path arm and the pre-tool arm)
+against the seven bulk ones the preload alone contributes. Ranked sources are
+added when somebody builds a ranker, which is rare and deliberate; bulk ones
+appear whenever a surface hands rules over, which is most of them. Naming the
+rare, slow-moving half means a newly-added bulk surface defaults to
+`ambient`, which merely under-counts it, instead of defaulting to `ranked`,
+which would quietly pad the pull-through denominator with surfacings nobody
+chose and make the arm look imprecise. Same argument #3191 and #3430 make
+against hand-kept lists: keep the list that must be remembered as short and as
+slow-moving as possible.
 """
 from __future__ import annotations
 
 import logging
 
-from sqlalchemy import func, select
+from sqlalchemy import case, func, select
 
 from scribe.models import async_session
 from scribe.models.base import iso
@@ -55,6 +90,31 @@ from scribe.services.background import report_telemetry_failure, spawn
 
 logger = logging.getLogger(__name__)
 
+# The surfaces that CHOSE the rules they showed. Everything else is ambient —
+# see the module docstring for why the rare half is the half that gets named.
+#
+# Membership is the whole definition of the pull-through denominator: a ranked
+# surfacing is a claim ("this rule may apply to what you are doing") that a pull
+# can confirm or refute, while an ambient one is a delivery nobody decided on.
+# Add a source here only when a ranker picked it.
+RANKED_SOURCES = ("write_path_rule", "pre_tool_rule")
+
+
+def is_ambient(source: str) -> bool:
+    """Was this surfacing a bulk delivery rather than a ranked choice?
+
+    One definition, read by both the per-rule badge readout and the aggregate
+    in `retrieval_telemetry` — the two used to be able to disagree about what
+    "surfaced" counted, which is the class of drift #3246 found across the
+    rules system.
+
+    Sync and pure, per the service canon (#2860), but deliberately PUBLIC where
+    that canon says such helpers stay `_private`. The departure is the point:
+    a module-private copy in each caller is exactly the second definition this
+    exists to prevent.
+    """
+    return source not in RANKED_SOURCES
+
 
 async def _report_failure(site: str) -> None:
     await report_telemetry_failure("rule_usage", site)
@@ -81,14 +141,21 @@ def record_rule_surfaced(
 ) -> None:
     """Fire-and-forget: record that these rules were shown to the agent.
 
-    Takes the whole hint at once — one insert per surfacing event, not per rule
-    — because a hint is a single decision and its rows should land together.
+    Takes the whole delivery at once — one insert per surfacing event, not per
+    rule — because a hint is a single decision and its rows should land
+    together.
 
-    Record the RANKED hits only. The arm filters candidates before it speaks
-    (`exclude_rule_ids` drops what the session already holds), and a rule that
-    was considered and not shown was not surfaced. Counting those would inflate
-    the denominator with claims the agent never saw, which reads as a precision
-    problem the arm does not have.
+    Record what was actually SHOWN, never what was considered. For the ranked
+    arm that means the post-filter hits: it drops what the session already
+    holds (`exclude_rule_ids`) before it speaks, and a rule considered and not
+    shown was not surfaced. Counting those would inflate the denominator with
+    claims the agent never saw, which reads as a precision problem the arm does
+    not have.
+
+    Bulk surfaces pass their whole delivered set, which is the same rule read
+    from the other end — everything in a preload IS shown. `source` is what
+    separates the two afterwards (see `RANKED_SOURCES`); this function does not
+    care which kind it is recording.
     """
     try:
         rows = [
@@ -137,9 +204,17 @@ def empty_rule_usage() -> dict:
     distinction matters more here than for notes: every rule in an install
     predates this table, so for a while "no events" is the normal state and it
     must not look like a broken readout.
+
+    `surfaced_count` is RANKED surfacings only; `ambient_count` is the bulk
+    deliveries (see `RANKED_SOURCES`). The split is what keeps the badge's
+    "shown often, opened never → dead weight" reading honest: every rule in an
+    always-on set is delivered every session, so an unsplit counter would rank
+    the resident set as the most-surfaced rules in the install purely for being
+    resident.
     """
     return {
         "surfaced_count": 0,
+        "ambient_count": 0,
         "pull_count": 0,
         "last_surfaced_at": None,
         "last_pulled_at": None,
@@ -159,6 +234,19 @@ async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
     if not ids:
         return out
 
+    # Classified in SQL so the group stays small: per rule we get at most
+    # (surfaced-ranked, surfaced-ambient, pulled) rather than a row per distinct
+    # source. ONE labelled expression, bound to a variable and reused in the
+    # GROUP BY — a second `case()` instance there renders its own expanding-IN
+    # bind names under asyncpg, so the database sees two DIFFERENT expressions
+    # and rejects the query with a GroupingError. The note twin carries the
+    # same warning for the same reason, and #2663 is what it cost: the
+    # rejection was swallowed and every counter read zero in production while
+    # the writes were landing fine.
+    ambient = case(
+        (RuleUsageEvent.source.notin_(RANKED_SOURCES), True),
+        else_=False,
+    ).label("ambient")
     try:
         async with async_session() as session:
             rows = (
@@ -168,9 +256,14 @@ async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
                         RuleUsageEvent.event,
                         func.count().label("n"),
                         func.max(RuleUsageEvent.created_at).label("last_at"),
+                        ambient,
                     )
                     .where(RuleUsageEvent.rule_id.in_(ids))
-                    .group_by(RuleUsageEvent.rule_id, RuleUsageEvent.event)
+                    .group_by(
+                        RuleUsageEvent.rule_id,
+                        RuleUsageEvent.event,
+                        ambient,
+                    )
                 )
             ).all()
     except Exception:
@@ -180,14 +273,23 @@ async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
         await _report_failure("readout")
         return out
 
-    for rule_id, event, n, last_at in rows:
+    for rule_id, event, n, last_at, is_amb in rows:
         slot = out.get(int(rule_id))
         if slot is None:
             continue
-        if event == SURFACED:
+        if event == SURFACED and is_amb:
+            slot["ambient_count"] = int(n)
+        elif event == SURFACED:
             slot["surfaced_count"] = int(n)
             slot["last_surfaced_at"] = iso(last_at)
         elif event == PULLED:
-            slot["pull_count"] = int(n)
-            slot["last_pulled_at"] = iso(last_at)
+            # Pulls are pulls regardless of what surfaced the rule — "did
+            # anyone ever open this?" does not depend on how it was found. Both
+            # halves accumulate, so this ADDS rather than assigns: a rule can
+            # now be pulled after a ranked hint and after a preload, and the
+            # split arrives as two rows.
+            slot["pull_count"] = slot["pull_count"] + int(n)
+            latest = iso(last_at)
+            if latest and (slot["last_pulled_at"] or "") < latest:
+                slot["last_pulled_at"] = latest
     return out

src/scribe/services/rulebooks.py

@@ -23,6 +23,7 @@ from scribe.services.verification import (
 )
 from scribe.services import rule_versions
 from scribe.models.rule_version import RuleVersion
+from scribe.services.rule_usage import record_rule_surfaced
 
 logger = logging.getLogger(__name__)
 
@@ -1395,7 +1396,7 @@ async def get_applicable_rules(
     }
 
 
-def rules_payload(applicable: dict) -> dict:
+def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict:
     """The caller-facing shape of a get_applicable_rules() result.
 
     Every surface that hands rules to an agent (enter_project, get_project,
@@ -1406,7 +1407,34 @@ def rules_payload(applicable: dict) -> dict:
     `excluded_always_on` (milestone 297) names the always-on rulebooks this
     project decided NOT to inherit, so the departure is visible wherever the
     rules are.
+
+    IT ALSO RECORDS THE SURFACING, which is why it now takes a caller and a
+    source. Every one of those surfaces is a bulk delivery — the applicable set
+    handed over whole, chosen by nobody — so this is the one place that has to
+    emit for all of them. Doing it per-caller instead would be five sites to
+    remember, and #3430 gap 2 is what that costs: the process→skill sync went
+    un-emitted through an entire dedicated telemetry survey because nothing
+    forced its surface to be accounted for.
+
+    `source` stays the CALLER's name rather than a constant, so the readout can
+    still separate the session handshake from a mid-session milestone read;
+    `RANKED_SOURCES` in `rule_usage` is what folds them back together.
+
+    Emitting from here is safe in a way emitting from `get_applicable_rules`
+    would not be: this function is only ever called to BUILD A REPLY. The two
+    other callers of the rules machinery — the write-path etag arm
+    (`plugin_context`) and `rules_etag_for` — compute a marker and show nobody
+    anything, and counting those would put rules in the denominator that no
+    agent ever saw.
     """
+    record_rule_surfaced(
+        user_id=user_id,
+        rule_ids=(
+            [r["id"] for r in applicable.get("rules", [])]
+            + [r["id"] for r in applicable.get("project_rules", [])]
+        ),
+        source=source,
+    )
     return {
         "applicable_rules": applicable["rules"],
         "applicable_rules_truncated": applicable["truncated"],

tests/test_inception_rules.py

@@ -15,14 +15,17 @@ def test_rules_payload_carries_excluded_always_on_as_the_seventh_key():
     out = rules_payload({
         "rules": [], "truncated": False, "subscribed_rulebooks": [],
         "excluded_always_on": [{"id": 1, "title": "Family"}],
-    })
+    }, user_id=1, source="enter_project")
     assert set(out) == {
         "applicable_rules", "applicable_rules_truncated", "subscribed_rulebooks",
         "project_rules", "suppressed_rules", "suppressed_topics", "excluded_always_on",
     }
     assert out["excluded_always_on"] == [{"id": 1, "title": "Family"}]
     # An older applicable dict without the key still renders (empty list).
-    assert rules_payload({"rules": [], "truncated": False, "subscribed_rulebooks": []})["excluded_always_on"] == []
+    assert rules_payload(
+        {"rules": [], "truncated": False, "subscribed_rulebooks": []},
+        user_id=1, source="enter_project",
+    )["excluded_always_on"] == []
 
 
 def test_list_always_on_rules_service_and_tool_take_a_project_id():

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

@@ -47,12 +47,14 @@ _PRIOR_ART = [(0.72, fake_note(id=9, title="debounce helper", user_id=1,
                                note_type="snippet"))]
 
 
-def _arm_patches(pc, hits, recorder, prior_art=None, cfg=None, rule_search=None):
+def _arm_patches(pc, hits, recorder, prior_art=None, cfg=None, rule_search=None,
+                 retrieval_log=None):
     """The minimum stubbing that lets the rule arm run and nothing else.
 
-    `cfg` and `rule_search` are overridable so a caller can inspect what the
-    arm ASKED for rather than only what it did with the answer — patching them
-    a second time on top would work, but reads as an accident.
+    `cfg`, `rule_search` and `retrieval_log` are overridable so a caller can
+    inspect what the arm ASKED for, and what it told the CALL log, rather than
+    only what it did with the answer — patching them a second time on top would
+    work, but reads as an accident.
     """
     return (
         patch.object(pc, "get_writepath_config",
@@ -66,7 +68,7 @@ def _arm_patches(pc, hits, recorder, prior_art=None, cfg=None, rule_search=None)
                                else prior_art)),
         patch.object(pc, "semantic_search_rules",
                      rule_search or AsyncMock(return_value=hits)),
-        patch.object(pc, "record_retrieval", MagicMock()),
+        patch.object(pc, "record_retrieval", retrieval_log or MagicMock()),
         patch.object(pc, "record_surfaced", MagicMock()),
         patch.object(pc, "record_rule_surfaced", recorder),
         patch.object(pc, "owner_names_for", AsyncMock(return_value={})),
@@ -74,10 +76,11 @@ def _arm_patches(pc, hits, recorder, prior_art=None, cfg=None, rule_search=None)
     )
 
 
-async def _run_arm(hits, recorder, prior_art=None, **kwargs):
+async def _run_arm(hits, recorder, prior_art=None, retrieval_log=None, **kwargs):
     from scribe.services import plugin_context as pc
     with ExitStack() as stack:
-        for ctx in _arm_patches(pc, hits, recorder, prior_art):
+        for ctx in _arm_patches(pc, hits, recorder, prior_art,
+                                retrieval_log=retrieval_log):
             stack.enter_context(ctx)
         return await pc.build_write_path_hint(
             1, "frontend/src/api/client.ts", code="x" * 400, **kwargs
@@ -274,3 +277,490 @@ def test_the_bulk_loaders_are_not_counted_as_pulls():
                 "applicable rule at once — so counting it would drown the "
                 "surfaced:pulled ratio in ambient delivery."
             )
+
+
+# ── The AMBIENT end: bulk deliveries (#3473) ───────────────────────────
+#
+# The preload was the largest rule surface in the product and emitted nothing,
+# so its cost was certain and its usefulness unfalsifiable. These assert the
+# three delivery shapes now emit — and, just as importantly, that the two
+# lookalike call sites which show nobody anything do NOT.
+
+
+@pytest.mark.asyncio
+async def test_the_session_start_preload_records_what_it_delivered():
+    """The block every session opens with. Chosen by nobody, paid for every
+    turn — and until it emitted, invisible to the scoreboard that judges every
+    other surface."""
+    from scribe.services import plugin_context as pc
+
+    rec = MagicMock()
+    rules = [fake_rule(id=1, title="`dev` is home"),
+             fake_rule(id=2, title="`main` — never without explicit request")]
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch.object(pc.rulebooks_svc, "list_always_on_rules",
+                         AsyncMock(return_value=rules))
+        )
+        stack.enter_context(
+            patch.object(pc.rulebooks_svc, "excluded_always_on_rulebooks",
+                         AsyncMock(return_value=[]))
+        )
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
+        stack.enter_context(
+            patch.object(pc, "_topic_titles", AsyncMock(return_value={}))
+        )
+        await pc.build_session_context(1, project_id=0)
+
+    assert rec.call_count == 1, "the preload recorded nothing"
+    kw = rec.call_args.kwargs
+    assert kw["rule_ids"] == [1, 2]
+    assert kw["source"] == "session_start"
+
+
+@pytest.mark.asyncio
+async def test_the_always_on_tool_records_what_it_handed_over():
+    from scribe.mcp.tools import rulebooks as tools
+
+    rec = MagicMock()
+    rules = [fake_rule(id=3, title="No GitHub — Fabled-Git only")]
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch.object(tools.rulebooks_svc, "list_always_on_rules",
+                         AsyncMock(return_value=rules))
+        )
+        stack.enter_context(
+            patch.object(tools.rulebooks_svc, "rules_etag",
+                         MagicMock(return_value="etag"))
+        )
+        stack.enter_context(patch.object(tools, "record_rule_surfaced", rec))
+        await tools.list_always_on_rules()
+
+    assert rec.call_args.kwargs["rule_ids"] == [3]
+    assert rec.call_args.kwargs["source"] == "list_always_on_rules"
+
+
+def test_rules_payload_records_both_the_family_and_project_halves():
+    """One emit site for all five `rules_payload` surfaces.
+
+    Per-caller emission would be five sites to remember, and #3430 gap 2 is
+    what that costs: the process→skill sync went un-emitted through an entire
+    dedicated telemetry survey because nothing forced its surface to be
+    accounted for.
+    """
+    from scribe.services import rulebooks as svc
+
+    rec = MagicMock()
+    with patch.object(svc, "record_rule_surfaced", rec):
+        svc.rules_payload(
+            {
+                "rules": [{"id": 10}, {"id": 11}],
+                "project_rules": [{"id": 12}],
+                "truncated": False,
+                "subscribed_rulebooks": [],
+            },
+            user_id=1,
+            source="enter_project",
+        )
+
+    kw = rec.call_args.kwargs
+    assert kw["rule_ids"] == [10, 11, 12], "project-scoped rules were delivered too"
+    assert kw["source"] == "enter_project"
+
+
+def test_every_rules_payload_caller_names_itself():
+    """`source` is the CALLER's name, so the readout can still separate the
+    session handshake from a mid-session milestone read. A shared constant here
+    would collapse five distinguishable surfaces into one."""
+    import re
+
+    seen = set()
+    for path in Path("src/scribe").rglob("*.py"):
+        for m in re.finditer(r"rules_payload\([^)]*source=\"([a-z_]+)\"", path.read_text()):
+            seen.add(m.group(1))
+    assert seen == {
+        "enter_project", "get_project", "get_milestone",
+        "start_planning", "get_task",
+    }, f"a rules_payload caller is missing or misnamed: {sorted(seen)}"
+
+
+def test_the_marker_paths_stay_silent():
+    """The two call sites that read the rules and show NOBODY anything.
+
+    `rules_etag_for` and the write-path staleness arm both call
+    `list_always_on_rules` to build or compare a marker. Emitting there would
+    put rules in the denominator that no agent ever saw — the exact inflation
+    `record_rule_surfaced`'s docstring forbids, arriving from the one direction
+    nothing else guards.
+    """
+    svc_src = Path("src/scribe/services/rulebooks.py").read_text()
+    etag_fn = svc_src.split("async def rules_etag_for")[1].split("\ndef ")[0]
+    assert "record_rule_surfaced" not in etag_fn, (
+        "rules_etag_for emits a surfacing — it builds a marker, it shows nothing"
+    )
+
+    pc_src = Path("src/scribe/services/plugin_context.py").read_text()
+    staleness = pc_src.split("if rules_etag:")[1].split("# The guard sits BELOW")[0]
+    assert "record_rule_surfaced" not in staleness, (
+        "the staleness arm emits a surfacing — it compares a marker, it shows nothing"
+    )
+
+
+# ── The PRE-TOOL arm: rules keyed on the action (#3476) ────────────────
+#
+# The write-path arm can only be reached by a code write, so every rule about
+# which tool to reach for was unretrievable at the moment it mattered — which
+# is why they all had to be resident. These cover the surface that changes it.
+
+
+def _tool_patches(pc, hits, recorder, cfg=None, retrieval_log=None):
+    return (
+        patch.object(pc, "get_writepath_config",
+                     AsyncMock(return_value=cfg or {
+                         "enabled": True, "threshold": 0.6,
+                         "top_k": 3, "rule_threshold": 0.6,
+                     })),
+        patch.object(pc, "semantic_search_rules", AsyncMock(return_value=hits)),
+        patch.object(pc, "record_retrieval", retrieval_log or MagicMock()),
+        patch.object(pc, "record_rule_surfaced", recorder),
+    )
+
+
+async def _run_tool_arm(hits, recorder, command="curl -s https://git.example/api/v1/runs",
+                        tool="Bash", retrieval_log=None, **kwargs):
+    from scribe.services import plugin_context as pc
+    with ExitStack() as stack:
+        for ctx in _tool_patches(pc, hits, recorder, retrieval_log=retrieval_log):
+            stack.enter_context(ctx)
+        return await pc.build_tool_rule_hint(1, tool, command, **kwargs)
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_names_a_rule_for_the_command_about_to_run():
+    """The 2026-09-03 incident in one test: reaching for curl against the forge
+    API is a Bash call, and nothing watched Bash."""
+    rec = MagicMock()
+    hits = [(0.71, fake_rule(id=161,
+                             title="Reach the forge through its MCP tools, never curl",
+                             when_to_apply="whenever you need CI status"))]
+    out = await _run_tool_arm(hits, rec)
+
+    assert out["rule_ids"] == [161]
+    assert "Reach the forge through its MCP tools" in out["context"]
+    assert "get_rule(161)" in out["context"], "the hint must hand over the way to read it"
+    assert "Bash" in out["context"], "the hint names the tool it is about"
+    assert rec.call_args.kwargs["source"] == "pre_tool_rule"
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_is_a_ranked_source():
+    """It CHOSE what it showed, so a pull can settle whether the choice was any
+    good — unlike a preload, which chose nothing. If this drifts into the
+    ambient class the arm becomes unjudgeable, which is the state #3311
+    described and M333 existed to end."""
+    from scribe.services.rule_usage import is_ambient
+
+    assert not is_ambient("pre_tool_rule")
+
+
+@pytest.mark.asyncio
+async def test_a_rule_the_session_already_holds_is_not_re_offered():
+    rec = MagicMock()
+    hits = [(0.71, 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, exclude_rule_ids=[161])
+
+    assert out["rule_ids"] == [12]
+    assert "161" not in out["context"]
+
+
+@pytest.mark.asyncio
+async def test_an_empty_command_asks_the_ranker_nothing():
+    """Every Bash call reaches this. A blank payload must cost no embedding
+    query at all, not merely return nothing after paying for one."""
+    from scribe.services import plugin_context as pc
+
+    search = AsyncMock(return_value=[])
+    rec = MagicMock()
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "semantic_search_rules", search))
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
+        out = await pc.build_tool_rule_hint(1, "Bash", "   ")
+
+    assert out == {"context": "", "rule_ids": []}
+    search.assert_not_called()
+    rec.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_fails_open():
+    """A recall aid may never break the operator's action. A ranker that raises
+    must cost the hint, not the command."""
+    from scribe.services import plugin_context as pc
+
+    with ExitStack() as stack:
+        stack.enter_context(patch.object(pc, "get_writepath_config",
+                                         AsyncMock(side_effect=RuntimeError("boom"))))
+        out = await pc.build_tool_rule_hint(1, "Bash", "docker compose up -d")
+
+    assert out == {"context": "", "rule_ids": []}
+
+
+@pytest.mark.asyncio
+async def test_a_long_command_is_bounded_before_it_reaches_the_ranker():
+    """A heredoc or a pasted script would push the verb and its target — the
+    part a rule is about — out of the embedding window."""
+    from scribe.services import plugin_context as pc
+
+    search = AsyncMock(return_value=[])
+    with ExitStack() as stack:
+        for ctx in _tool_patches(pc, [], MagicMock()):
+            stack.enter_context(ctx)
+        stack.enter_context(patch.object(pc, "semantic_search_rules", search))
+        await pc.build_tool_rule_hint(1, "Bash", "git tag v1 && " + "x" * 5000)
+
+    sent = search.call_args.args[1]
+    assert len(sent) <= pc._TOOL_QUERY_CHARS
+    assert sent.startswith("git tag v1"), "the head of the command is the signal"
+
+
+def test_the_two_pre_tool_arms_share_one_session_rule_ledger():
+    """The integration point most worth guarding.
+
+    Two ledgers would mean a rule named by the write arm gets re-offered by the
+    tool arm — and the hint that fires most often is exactly the one that must
+    not repeat itself. Asserted on the FILENAME both scripts build, because
+    that is the shared thing; a copy of the path in each is how they drift.
+    """
+    prior = Path("plugin/hooks/scribe_prior_art.sh").read_text()
+    tool = Path("plugin/hooks/scribe_tool_rules.sh").read_text()
+
+    for src, name in ((prior, "scribe_prior_art.sh"), (tool, "scribe_tool_rules.sh")):
+        assert '"${TMPDIR:-/tmp}/scribe-priorart"' in src, f"{name}: state dir moved"
+        assert '.rules.ids' in src, f"{name}: rules ledger filename moved"
+        assert "exclude_rule_ids" in src, f"{name}: does not send the exclusion"
+
+
+def test_the_tool_arm_is_registered_on_bash():
+    """A hook that exists and is not registered runs never — and reads exactly
+    like a surface nobody needed."""
+    import json
+
+    manifest = json.loads(Path("plugin/hooks/hooks.json").read_text())
+    pre = manifest["hooks"]["PreToolUse"]
+    entries = {
+        m.get("matcher"): [h["command"] for h in m["hooks"]] for m in pre
+    }
+    assert "Bash" in entries, "nothing watches Bash — the reflex surface is unguarded"
+    assert any("scribe_tool_rules.sh" in c for c in entries["Bash"])
+    # The write arm keeps its own matcher; this is an addition, not a move.
+    assert any("scribe_prior_art.sh" in c for c in entries["Write|Edit"])
+
+
+def test_the_hook_and_the_route_agree_on_every_parameter_name():
+    """Rule 33, on a brand-new integration between layers.
+
+    The hook is shell and the route is Python; nothing but this test connects
+    them. A renamed query arg fails SILENTLY — the route reads an absent value,
+    the arm quietly searches nothing, and the surface looks like one that never
+    finds anything rather than one that is broken.
+    """
+    import re
+
+    hook = Path("plugin/hooks/scribe_tool_rules.sh").read_text()
+    route = Path("src/scribe/routes/plugin.py").read_text()
+    handler = route.split("async def pre_tool_rules")[1].split("\n@plugin_bp")[0]
+
+    sent = set(re.findall(r"[?&]([a-z_]+)=", hook))
+    assert sent == {"tool", "command", "repo", "exclude_rule_ids"}, sent
+
+    # `repo` is read by the shared _project_scope() helper, not inline.
+    assert "_project_scope()" in handler
+    for arg in ("tool", "command", "exclude_rule_ids"):
+        assert f'request.args.get("{arg}")' in handler, (
+            f"the hook sends {arg!r} and the route never reads it"
+        )
+
+
+# ── The CALL log is unconditional; the SURFACING log is not (#3497) ────
+#
+# Both arms used to write their retrieval_logs row inside a guard on having
+# results, so `zero_result_calls` was pinned at 0 and `cleared_threshold` at
+# `calls` by the shape of the code — at any threshold whatsoever. #3311 read
+# that as a measurement of the corpus and a milestone was scoped on it.
+#
+# The distinction these tests hold: a CALL happened whether or not it found
+# anything, and the calls that found nothing are the only evidence a threshold
+# is set too high. A SURFACING did not happen when nothing was shown.
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_arm_logs_the_call_that_found_nothing():
+    log, rec = MagicMock(), MagicMock()
+    await _run_arm([], rec, retrieval_log=log)
+
+    rule_calls = [c for c in log.call_args_list
+                  if c.kwargs.get("source") == "write_path_rule"]
+    assert len(rule_calls) == 1, (
+        "a rule call that found nothing wrote no row — `zero_result_calls` can "
+        "then only ever read 0, however badly the threshold is tuned"
+    )
+    assert rule_calls[0].kwargs["results"] == []
+    rec.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_write_path_arm_logs_a_call_whose_only_hit_was_already_shown():
+    """The subtler half. The ranker DID find something; the session had already
+    been told. That is a decline from the reader's side and must be logged as
+    one — the note arms get this for free by passing exclusions into the search,
+    so their zero-result rows already include this case."""
+    log, rec = MagicMock(), MagicMock()
+    hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug"))]
+    await _run_arm(hits, rec, retrieval_log=log, exclude_rule_ids=[156])
+
+    rule_calls = [c for c in log.call_args_list
+                  if c.kwargs.get("source") == "write_path_rule"]
+    assert len(rule_calls) == 1
+    assert rule_calls[0].kwargs["results"] == [], (
+        "the row must record what the arm could SHOW, so this row is comparable "
+        "with an auto_inject row, whose exclusions are applied by the search"
+    )
+    rec.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_logs_the_call_that_found_nothing():
+    """It matters more here than on the sibling. This arm fires on every Bash
+    call, so an empty `sources` row is the normal outcome — and with no row at
+    all, "the ranker declined" is indistinguishable from "the hook never fired",
+    which is exactly the silent failure the arm was built to stop (#3476)."""
+    log, rec = MagicMock(), MagicMock()
+    out = await _run_tool_arm([], rec, retrieval_log=log)
+
+    assert out == {"context": "", "rule_ids": []}
+    assert log.call_count == 1
+    assert log.call_args.kwargs["source"] == "pre_tool_rule"
+    assert log.call_args.kwargs["results"] == []
+    assert log.call_args.kwargs["query"] == "curl -s https://git.example/api/v1/runs", (
+        "the query is the point of the row: it is what a threshold is tuned against"
+    )
+    rec.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_the_tool_arm_logs_a_call_whose_only_hit_was_already_shown():
+    log, rec = MagicMock(), MagicMock()
+    hits = [(0.71, fake_rule(id=161, title="Reach the forge through its MCP tools"))]
+    out = await _run_tool_arm(hits, rec, retrieval_log=log, exclude_rule_ids=[161])
+
+    assert out["rule_ids"] == []
+    assert log.call_count == 1
+    assert log.call_args.kwargs["results"] == []
+    rec.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_a_command_the_arm_never_searched_writes_no_row_at_all():
+    """The one case that must stay silent, and the boundary of the rule above.
+
+    A blank command costs no embedding query, so there was no retrieval to log.
+    A row here would report a call that never happened and drag the clear-rate
+    down with phantom declines — the mirror of the defect, from the other side.
+    """
+    from scribe.services import plugin_context as pc
+
+    log = MagicMock()
+    with ExitStack() as stack:
+        for ctx in _tool_patches(pc, [], MagicMock(), retrieval_log=log):
+            stack.enter_context(ctx)
+        await pc.build_tool_rule_hint(1, "Bash", "   ")
+
+    log.assert_not_called()
+
+
+def test_neither_rule_arm_logs_its_call_behind_a_results_guard():
+    """Structural, on top of the behavioural pair above, because the defect was
+    one level of indentation and it appeared INDEPENDENTLY in two places — the
+    pre-tool arm inherited it by being modelled on its sibling. The third arm
+    modelled on either of them is the one this catches.
+    """
+    pc_src = Path("src/scribe/services/plugin_context.py").read_text()
+
+    # Write-path arm: what remains inside `if fresh:` is the SURFACING log only.
+    guarded = pc_src.split('source="write_path_rule", query=code or path')[1]
+    guarded = guarded.split("if fresh:")[1].split("except Exception:")[0]
+    assert "record_rule_surfaced" in guarded, "the surfacing log must stay guarded"
+    assert "record_retrieval" not in guarded, (
+        "the call log is back inside the results guard — a call that found "
+        "nothing is the only evidence a threshold is set too high"
+    )
+
+    # Pre-tool arm: the call log comes BEFORE the early return.
+    body = pc_src.split("async def build_tool_rule_hint")[1]
+    assert body.index('source="pre_tool_rule"') < body.index("if not fresh:"), (
+        "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

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."""
@@ -511,3 +573,68 @@ async def test_rule_usage_sees_only_its_own_users_events(_dispose_engine):
         assert (await retrieval_summary(990014, days=30))["rule_usage"]["surfaced"] == 1
     finally:
         await cleanup()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_the_preload_lands_in_ambient_and_never_in_the_ratio(_dispose_engine):
+    """The split that makes the always-on set judgeable (#3473).
+
+    Pull-through asks "was that hint any use", and only a surface that CHOSE
+    what it showed can be judged by it. If the preload counted toward the
+    denominator, growing the always-on set would DEPRESS the arm's measured
+    precision and trimming it would flatter it — neither for any reason to do
+    with the arm. So the resident deliveries are counted, reported, and kept
+    out of the ratio.
+    """
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    cleanup = await _rule_events(990012, [
+        # One rule the arm actually chose, and opened.
+        (5101, "surfaced", "write_path_rule"),
+        (5101, "pulled", "mcp_get_rule"),
+        # Four bulk deliveries across every shape of preload. Nobody chose any
+        # of them, and none may touch the denominator.
+        (5102, "surfaced", "session_start"),
+        (5103, "surfaced", "list_always_on_rules"),
+        (5104, "surfaced", "enter_project"),
+        (5105, "surfaced", "get_milestone"),
+    ])
+    try:
+        ru = (await retrieval_summary(990012, days=30))["rule_usage"]
+
+        assert ru["surfaced"] == 1, "only the arm chose a rule"
+        assert ru["ambient"] == 4, "the four bulk deliveries are reported, not dropped"
+
+        # 1 agent pull over 1 RANKED surfacing. Were the ambient four folded in
+        # the ratio would read 0.2 — the arm looking four times worse for
+        # having a large resident set beside it.
+        assert ru["pull_through"] == 1.0
+
+        # Dead-weight detection needs both classes: a rule delivered by the
+        # preload and never opened is the case that reading matters most for.
+        assert ru["distinct_rules_surfaced"] == 5
+        assert ru["distinct_rules_pulled"] == 1
+    finally:
+        await cleanup()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_ambient_alone_reports_no_ratio(_dispose_engine):
+    """A brand-new install loads rules every session and may never trigger the
+    arm. That must read as "no ranked surfacings yet", not as a precision of
+    zero — the reading that would make a working install look broken."""
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    cleanup = await _rule_events(990013, [
+        (5201, "surfaced", "session_start"),
+        (5202, "surfaced", "session_start"),
+    ])
+    try:
+        ru = (await retrieval_summary(990013, days=30))["rule_usage"]
+        assert ru["ambient"] == 2
+        assert ru["surfaced"] == 0
+        assert ru["pull_through"] is None
+    finally:
+        await cleanup()

tests/test_services_rule_usage.py

@@ -99,12 +99,32 @@ def test_the_zero_readout_names_every_key():
     — a missing key here would read as a broken readout on almost every row."""
     assert rule_usage.empty_rule_usage() == {
         "surfaced_count": 0,
+        "ambient_count": 0,
         "pull_count": 0,
         "last_surfaced_at": None,
         "last_pulled_at": None,
     }
 
 
+def test_only_a_ranker_counts_as_ranked():
+    """The bulk surfaces are ambient; the write-path arm is the only chooser.
+
+    Inverted against the note twin on purpose (see the module docstring): the
+    RARE half is the one that gets named, so a bulk surface added later and
+    forgotten defaults to ambient — under-counting it — instead of defaulting
+    to ranked and padding the pull-through denominator with surfacings nobody
+    chose.
+    """
+    assert not rule_usage.is_ambient("write_path_rule")
+    for bulk in (
+        "session_start", "list_always_on_rules", "enter_project",
+        "get_project", "get_milestone", "start_planning", "get_task",
+    ):
+        assert rule_usage.is_ambient(bulk), bulk
+    # The safe default is the whole point of the inversion.
+    assert rule_usage.is_ambient("some_surface_invented_next_year")
+
+
 def test_the_model_serialises_the_fields_the_ratio_needs():
     ev = RuleUsageEvent(
         user_id=7, rule_id=156, event=SURFACED, source="write_path_rule"
@@ -161,6 +181,10 @@ async def test_usage_for_rules_aggregates_per_rule(_dispose_engine):
         # never have to tell "no events" from "not in the result" — and on any
         # existing install that is nearly every rule.
         assert out[6003] == rule_usage.empty_rule_usage()
+
+        # Nothing ambient in this fixture, so the ambient bucket stays empty
+        # rather than absorbing the ranked hits.
+        assert out[6001]["ambient_count"] == 0
     finally:
         async with async_session() as s:
             await s.execute(
@@ -178,3 +202,56 @@ async def test_usage_for_rules_on_an_empty_id_list_asks_the_database_nothing(
     empty topic is nothing. An unguarded `IN ()` is both a pointless round trip
     and, on some drivers, a syntax error."""
     assert await rule_usage.usage_for_rules([]) == {}
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_a_preloaded_rule_does_not_read_as_a_ranked_surfacing(_dispose_engine):
+    """The split that makes the always-on set judgeable (#3473).
+
+    A resident rule is delivered every session by a surface that chose
+    nothing. Counting those as `surfaced_count` would rank the always-on set
+    as the most-surfaced rules in the install purely for being resident — and
+    the badge's "shown often, opened never → dead weight" reading, which is
+    the whole reason the counter exists, would then be exactly backwards.
+    """
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.rule_usage import RuleUsageEvent
+
+    async with async_session() as s:
+        s.add_all([
+            # Delivered by the preload three times over: ambient, all of it.
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="session_start"),
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="list_always_on_rules"),
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="enter_project"),
+            # ...and once by the arm, which DID choose it.
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="write_path_rule"),
+            # Opened once after a hint and once from the list: pulls are pulls
+            # however the rule was found, so both land in the one counter.
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=PULLED, source="mcp_get_rule"),
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=PULLED, source="rest_rule"),
+        ])
+        await s.commit()
+    try:
+        out = await rule_usage.usage_for_rules([6101])
+
+        assert out[6101]["surfaced_count"] == 1, "only the arm chose this rule"
+        assert out[6101]["ambient_count"] == 3, "three bulk deliveries"
+        # Both PULLED rows accumulate — the loop ADDS rather than assigns, so a
+        # rule opened after a hint and again from the list reports two, not one.
+        assert out[6101]["pull_count"] == 2
+        assert out[6101]["last_pulled_at"] is not None
+    finally:
+        async with async_session() as s:
+            await s.execute(
+                delete(RuleUsageEvent).where(RuleUsageEvent.user_id == 990021)
+            )
+            await s.commit()

tests/test_write_path_trigger.py

@@ -360,10 +360,18 @@ async def test_telemetry_uses_its_own_source():
          patch.object(pc, "record_retrieval", rec), \
          patch.object(pc, "owner_names_for", AsyncMock(return_value={})):
         await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE, project_id=4)
-    rec.assert_called_once()
-    assert rec.call_args.kwargs["source"] == "write_path"
-    assert rec.call_args.kwargs["source"] != "auto_inject"
-    assert rec.call_args.kwargs["project_id"] == 4
+    sources = [c.kwargs["source"] for c in rec.call_args_list]
+    assert sources.count("write_path") == 1, sources
+    assert "auto_inject" not in sources
+
+    note_arm = next(c for c in rec.call_args_list if c.kwargs["source"] == "write_path")
+    assert note_arm.kwargs["project_id"] == 4
+
+    # The rule arm rides along on the same hint and logs its own call even when
+    # it finds nothing (#3497). This assertion used to be `assert_called_once`,
+    # which passed only because that row was never written — the test encoded
+    # the defect. The second row is the point of having two sources.
+    assert "write_path_rule" in sources, sources
 
 
 @pytest.mark.asyncio