diff --git a/plugin/.claude-plugin/plugin.json b/plugin/.claude-plugin/plugin.json
index c462aa9..f0f1eaf 100644
--- a/plugin/.claude-plugin/plugin.json
+++ b/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.03.0329",
   "author": {
     "name": "Bryan Van Deusen"
   },
diff --git a/plugin/hooks/hooks.json b/plugin/hooks/hooks.json
index 93a9299..0bec8ea 100644
--- a/plugin/hooks/hooks.json
+++ b/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": [
diff --git a/plugin/hooks/scribe_tool_rules.sh b/plugin/hooks/scribe_tool_rules.sh
new file mode 100644
index 0000000..6803a78
--- /dev/null
+++ b/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
diff --git a/scripts/check_plugin.py b/scripts/check_plugin.py
index dd381ba..d0da1ac 100755
--- a/scripts/check_plugin.py
+++ b/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
diff --git a/src/scribe/mcp/tools/milestones.py b/src/scribe/mcp/tools/milestones.py
index 39b144d..cde1687 100644
--- a/src/scribe/mcp/tools/milestones.py
+++ b/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"),
     }
 
 
diff --git a/src/scribe/mcp/tools/projects.py b/src/scribe/mcp/tools/projects.py
index b2620d6..8f90d57 100644
--- a/src/scribe/mcp/tools/projects.py
+++ b/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
 
 
diff --git a/src/scribe/mcp/tools/rulebooks.py b/src/scribe/mcp/tools/rulebooks.py
index 178cbbe..e21fdb2 100644
--- a/src/scribe/mcp/tools/rulebooks.py
+++ b/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),
diff --git a/src/scribe/mcp/tools/search.py b/src/scribe/mcp/tools/search.py
index e00730f..a686a1c 100644
--- a/src/scribe/mcp/tools/search.py
+++ b/src/scribe/mcp/tools/search.py
@@ -197,17 +197,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
diff --git a/src/scribe/mcp/tools/tasks.py b/src/scribe/mcp/tools/tasks.py
index e097292..32eef56 100644
--- a/src/scribe/mcp/tools/tasks.py
+++ b/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
diff --git a/src/scribe/routes/plugin.py b/src/scribe/routes/plugin.py
index b951413..e050018 100644
--- a/src/scribe/routes/plugin.py
+++ b/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():
diff --git a/src/scribe/services/planning.py b/src/scribe/services/planning.py
index 4f58c97..d7addeb 100644
--- a/src/scribe/services/planning.py
+++ b/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,
     }
diff --git a/src/scribe/services/plugin_context.py b/src/scribe/services/plugin_context.py
index 48ccf2f..18bee4e 100644
--- a/src/scribe/services/plugin_context.py
+++ b/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").
@@ -1250,6 +1259,101 @@ 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]
+        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)
+
+        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,
+        )
+        # 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 +1479,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 []
diff --git a/src/scribe/services/retrieval_telemetry.py b/src/scribe/services/retrieval_telemetry.py
index af27dda..e21629a 100644
--- a/src/scribe/services/retrieval_telemetry.py
+++ b/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__)
@@ -428,10 +429,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 +551,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 +578,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 +602,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
diff --git a/src/scribe/services/rule_usage.py b/src/scribe/services/rule_usage.py
index 061005b..b30402d 100644
--- a/src/scribe/services/rule_usage.py
+++ b/src/scribe/services/rule_usage.py
@@ -30,23 +30,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 +78,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 +129,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 +192,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 +222,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 +244,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 +261,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
diff --git a/src/scribe/services/rulebooks.py b/src/scribe/services/rulebooks.py
index ba11b69..c8e380e 100644
--- a/src/scribe/services/rulebooks.py
+++ b/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"],
diff --git a/tests/test_inception_rules.py b/tests/test_inception_rules.py
index 4f1172f..6a614e6 100644
--- a/tests/test_inception_rules.py
+++ b/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():
diff --git a/tests/test_rule_usage_wiring.py b/tests/test_rule_usage_wiring.py
index 7d4c667..a012b8c 100644
--- a/tests/test_rule_usage_wiring.py
+++ b/tests/test_rule_usage_wiring.py
@@ -274,3 +274,306 @@ 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):
+    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", 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", **kwargs):
+    from scribe.services import plugin_context as pc
+    with ExitStack() as stack:
+        for ctx in _tool_patches(pc, hits, recorder):
+            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"
+        )
diff --git a/tests/test_services_retrieval_telemetry.py b/tests/test_services_retrieval_telemetry.py
index e5b529b..1bea2ca 100644
--- a/tests/test_services_retrieval_telemetry.py
+++ b/tests/test_services_retrieval_telemetry.py
@@ -511,3 +511,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()
diff --git a/tests/test_services_rule_usage.py b/tests/test_services_rule_usage.py
index 6efb23c..95cbe8b 100644
--- a/tests/test_services_rule_usage.py
+++ b/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()