fix(6247): a getter is seen by its pulls, so the silent list stops naming every pull door

usage and rule_usage gain pulled_by_source, read from the per-source rows
the readout already had, and _silent_surfaces counts those doors as having
emitted. The rules-listing points were seen as of the previous commit,
through rule_usage.by_source.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

frontend/src/views/SettingsView.vue

@@ -159,16 +159,10 @@ const kbWritePathEnabled = ref(true);
 // code embeddings sit on a much higher similarity floor than prose, so 0.55
 // lets unrelated code through. Shares top-k, not the threshold.
 const kbWritePathThreshold = ref("0.68");
-const kbRuleHintThreshold = ref("0.72");
+// One bar for every rule hint — prompt, command and edit alike. They search
+// the same rules, and what limits each is its budget, not the bar.
+const kbRuleThreshold = ref("0.70");
 const kbCheckpointThreshold = ref("0.80");
-// The two ACT arms do not share a bar. A write-path query is a code payload; a
-// pre-tool query is a shell command, often under a dozen words — less text,
-// less signal, lower scores for the same relevance — so one shared bar makes
-// them behave like different subsystems.
-const kbToolRuleThreshold = ref("0.68");
-// And the prompt boundary is a third query shape again — the operator's own
-// prose rather than anything a tool produced.
-const kbPromptRuleThreshold = ref("0.72");
 // The reply arm: its bar is the bar at which a finished reply
 // is HELD for one read, so it sits with the checkpoint, not with the hints.
 const kbReplyRuleThreshold = ref("0.8");
@@ -367,14 +361,10 @@ async function saveKbInject() {
   // Same `|| default` reasoning: falling back to 0 would surface every
   // snippet in the corpus on every edit, which is the failure this knob fixes.
   const wpT = asBar(kbWritePathThreshold.value, 0.68);
-  // Same `|| default` reasoning again, and it bites harder here: a rule hint
-  // fires on every write, so a fallback of 0 would attach a standing rule to
-  // every edit in the session.
-  const rhT = asBar(kbRuleHintThreshold.value, 0.72);
-  // Same `|| default` guard, and the same reason: this arm fires before every
-  // Bash call, so a fallback of 0 would put a rule in front of every command.
-  const trT = asBar(kbToolRuleThreshold.value, 0.68);
-  const prT = asBar(kbPromptRuleThreshold.value, 0.72);
+  // Same `|| default` reasoning again, and it bites harder here: rule hints
+  // fire on every edit and before every command, so a fallback of 0 would put
+  // a standing rule beside each of them.
+  const ruleT = asBar(kbRuleThreshold.value, 0.7);
   // The checkpoint bar, and the `|| default` guard matters most here of all:
   // this is the only number that can STOP a call, so a fallback of 0 would
   // hold the first command of every session behind whatever ranked first.
@@ -409,10 +399,8 @@ async function saveKbInject() {
   kbGateNoteOverlapFloor.value = String(gateOverlap);
   kbPlanMatchThreshold.value = String(planT);
   kbWritePathThreshold.value = String(wpT);
-  kbRuleHintThreshold.value = String(rhT);
+  kbRuleThreshold.value = String(ruleT);
   kbCheckpointThreshold.value = String(cpT);
-  kbToolRuleThreshold.value = String(trT);
-  kbPromptRuleThreshold.value = String(prT);
   kbReplyRuleThreshold.value = String(ryT);
   savingKbInject.value = true;
   kbInjectSaved.value = false;
@@ -426,20 +414,15 @@ async function saveKbInject() {
       // measurements that split them.
       kb_writepath_enabled: kbWritePathEnabled.value ? 'true' : 'false',
       kb_writepath_threshold: String(wpT),
-      // A THIRD corpus with a third bar — see RULEHINT_DEFAULT_THRESHOLD
-      // in services/plugin_context.py for why rules cannot share the
-      // code threshold any more than code could share the prose one.
-      kb_rulehint_threshold: String(rhT),
+      // A third corpus with a third bar — see RULE_FLOOR_DEFAULT in
+      // services/retrieval_pipeline.py for why every rule hint shares it.
+      kb_rule_threshold: String(ruleT),
       // The one bar that stops a call rather than annotating it. Its own key,
-      // never derived from the two rule bars above: it answers a different
+      // never derived from the rule bar above: it answers a different
       // question of the same scores — not "is this worth showing" but "is the
       // corpus confident enough to be read first" — and a number that moves
       // when another moves is a number nobody can reason about.
       kb_checkpoint_threshold: String(cpT),
-      // A FOURTH and FIFTH bar, and they are separate keys on purpose: one
-      // number cannot serve arms whose queries are different shapes. Moving one must not move the others.
-      kb_toolrule_threshold: String(trT),
-      kb_promptrule_threshold: String(prT),
       // The reply backstop's own bar: it HOLDS a reply, so like the
       // checkpoint it is never derived from a hint bar.
       kb_replyrule_threshold: String(ryT),
@@ -864,14 +847,8 @@ onMounted(async () => {
   if (allSettings.kb_checkpoint_threshold !== undefined) {
     kbCheckpointThreshold.value = allSettings.kb_checkpoint_threshold;
   }
-  if (allSettings.kb_rulehint_threshold !== undefined) {
-    kbRuleHintThreshold.value = allSettings.kb_rulehint_threshold;
-  }
-  if (allSettings.kb_toolrule_threshold !== undefined) {
-    kbToolRuleThreshold.value = allSettings.kb_toolrule_threshold;
-  }
-  if (allSettings.kb_promptrule_threshold !== undefined) {
-    kbPromptRuleThreshold.value = allSettings.kb_promptrule_threshold;
+  if (allSettings.kb_rule_threshold !== undefined) {
+    kbRuleThreshold.value = allSettings.kb_rule_threshold;
   }
   if (allSettings.kb_replyrule_threshold !== undefined) {
     kbReplyRuleThreshold.value = allSettings.kb_replyrule_threshold;
@@ -1772,10 +1749,10 @@ async function deleteUser(userId: number) {
           <p class="field-hint">How many snippets and issues one edit may be offered (1–10). Reach for this rather than the threshold when the hint feels long: lowering it keeps the best matches and drops the tail, while raising the bar above throws away good matches along with weak ones.</p>
         </div>
         <div class="field">
-          <label for="kb-rulehint-threshold">Standing-rule confidence threshold (0–1)</label>
+          <label for="kb-rule-threshold">Standing-rule confidence threshold (0–1)</label>
           <input
-            id="kb-rulehint-threshold"
-            v-model="kbRuleHintThreshold"
+            id="kb-rule-threshold"
+            v-model="kbRuleThreshold"
             type="number"
             min="0"
             max="1"
@@ -1784,14 +1761,14 @@ async function deleteUser(userId: number) {
             style="max-width: 8rem"
           />
           <p class="field-hint">
-            The same hint can mention a standing rule whose trigger resembles the
-            code being written. <em>Every</em> rule is eligible — nothing is
-            preloaded any more, so this is the only way a rule reaches a write.
-            Stricter than the threshold above, because there are far fewer rules
-            than snippets: with a small set something always ranks first, so the
-            bar has to carry more of the judgement. If rules arrive unread, or one
-            you needed never showed up, that is Claude's to notice and correct
-            from the telemetry — and the budget below is usually the better lever.
+            The bar for a standing rule or preference to be offered at all,
+            whether against the code being written, a command about to run, or
+            what you just said. <em>Every</em> rule is eligible — nothing is
+            preloaded, so this is how a rule reaches the work. One bar serves all
+            three: they search the same rules, and how many each may show is set
+            by its own budget below, which is usually the better lever. If rules
+            arrive unread, or one you needed never showed up, that is Claude's to
+            notice and correct from the telemetry.
           </p>
         </div>
         <div class="field">
@@ -1834,26 +1811,6 @@ async function deleteUser(userId: number) {
             only — file edits are never held.
           </p>
         </div>
-        <div class="field">
-          <label for="kb-toolrule-threshold">Command confidence threshold (0–1)</label>
-          <input
-            id="kb-toolrule-threshold"
-            v-model="kbToolRuleThreshold"
-            type="number"
-            min="0"
-            max="1"
-            step="0.01"
-            class="fs-input input"
-            style="max-width: 8rem"
-          />
-          <p class="field-hint">
-            The bar for a rule surfacing before a <em>command</em> runs, where the
-            query is the command text rather than code. Lower than the one above
-            on purpose: a shell command is short, so it scores lower for the same
-            relevance — at a shared bar this arm spoke on 2% of calls against the
-            write path's 37%.
-          </p>
-        </div>
         <div class="field">
           <label for="kb-toolrule-topk">Max rules per command</label>
           <input
@@ -1868,26 +1825,6 @@ async function deleteUser(userId: number) {
           />
           <p class="field-hint">How many standing rules one command may be shown (1–10). This is the busiest arm there is — it fires before every command — so its budget is the one most worth keeping small.</p>
         </div>
-        <div class="field">
-          <label for="kb-promptrule-threshold">Prompt confidence threshold (0–1)</label>
-          <input
-            id="kb-promptrule-threshold"
-            v-model="kbPromptRuleThreshold"
-            type="number"
-            min="0"
-            max="1"
-            step="0.01"
-            class="fs-input input"
-            style="max-width: 8rem"
-          />
-          <p class="field-hint">
-            The bar for a rule or preference surfacing against <em>what you just
-            said</em>, before anything is done. It is the only moment that reaches
-            a rule about how to answer rather than how to act, so it runs once a
-            turn rather than once a tool call. A third query shape — your prose,
-            not a command or a file — which is why it carries its own number.
-          </p>
-        </div>
         <div class="field">
           <label for="kb-promptrule-topk">Max rules per prompt</label>
           <input

src/grimoire/mcp/tools/rulebooks.py

@@ -10,13 +10,16 @@ preview-style warning.
 """
 from __future__ import annotations
 
+from datetime import datetime, timedelta, timezone
+
 from grimoire.mcp._context import current_user_id
 from grimoire.services import dedup as dedup_svc
 from grimoire.services import moments as moments_svc
 from grimoire.services import rulebooks as rulebooks_svc
 from grimoire.services import trash as trash_svc
+from grimoire.services.lesson_rules import confirmed_lesson_counts
 from grimoire.services.rule_usage import (
-    record_rule_outcome, record_rule_pulled,
+    empty_rule_usage, record_rule_outcome, record_rule_pulled, usage_for_rules,
 )
 from grimoire.services import moment_delivery
 
@@ -204,6 +207,7 @@ def _rule_summary(r, moments: list[str] | None = None) -> dict:
 
 async def list_rules(
     rulebook_id: int = 0, topic_id: int = 0, project_id: int = 0,
+    with_usage: bool = False, days: int = 0,
 ) -> dict:
     """List rules — filter by rulebook, topic, and/or project.
 
@@ -213,9 +217,21 @@ async def list_rules(
         project_id: 0 = no filter; positive = that project's OWN rules.
             Global rules apply to every project, so they are listed by
             rulebook or topic (or unfiltered), not under each project.
+        with_usage: add each rule's `usage` and `confirmed_lessons`. Off by
+            default: a listing is for applying rules, and this is for judging
+            them.
+        days: with `with_usage`, count only the last N days; 0 = everything
+            recorded.
 
     rulebook_id and topic_id AND-combine; project_id lists a project's rules
     on its own. Ownership-scoped.
+
+    `usage` is the rule's own telemetry: `surfaced_count` (a ranked surface
+    chose it), `ambient_count` (a bulk delivery carried it), `pull_count`,
+    `applied_count`, `departed_count` and the latest of each.
+    `confirmed_lessons` counts the lessons judged instances of the rule. Read
+    together they answer whether rules that situations keep proving are the
+    ones that surface and get opened.
     """
     uid = current_user_id()
     rows = await rulebooks_svc.list_rules(
@@ -224,12 +240,20 @@ async def list_rules(
         topic_id=topic_id or None,
         project_id=project_id or None,
     )
+    ids = [r.id for r in rows]
     # One batched read for the page, not one per row.
-    mounted = await rulebooks_svc.list_rule_moments([r.id for r in rows])
-    return {
-        "rules": [_rule_summary(r, mounted.get(r.id)) for r in rows],
-        "total": len(rows),
-    }
+    mounted = await rulebooks_svc.list_rule_moments(ids)
+    rules = [_rule_summary(r, mounted.get(r.id)) for r in rows]
+    if with_usage:
+        since = (
+            datetime.now(timezone.utc) - timedelta(days=days) if days > 0 else None
+        )
+        usage = await usage_for_rules(ids, since=since)
+        lessons = await confirmed_lesson_counts(uid, ids)
+        for row in rules:
+            row["usage"] = usage.get(row["id"], empty_rule_usage())
+            row["confirmed_lessons"] = lessons.get(row["id"], 0)
+    return {"rules": rules, "total": len(rows)}
 
 
 async def get_rule(rule_id: int) -> dict:

src/grimoire/mcp/tools/search.py

@@ -509,6 +509,15 @@ async def retrieval_telemetry(
     Read it against `sources["write_path_rule"]`, which declines the large
     majority of its calls like any other surface.
 
+    `rule_usage.by_source` is the per-surface split, the twin of
+    `usage.by_source`: for each surface that showed rules, `rules_surfaced`
+    (distinct rules), `rules_pulled` (how many of those an agent opened in the
+    window) and `pull_through`, null on an ambient surface. It is the readout
+    that says which arm — the prompt arm, the tool arm, a lesson pointing at
+    its rule — earns its lines. The same upper bound as the note split: a rule
+    two surfaces showed and an agent opened once counts for both.
+    `by_source_failed: true` means that one query failed.
+
     `system_usage` — the RULINGS arm, from `system_usage_events`: how
     often an area's rulings were shown because a command (`rulings_pre_tool`)
     or an edit (`rulings_write_path`) touched its files, how often a System

src/grimoire/services/lesson_rules.py

@@ -324,6 +324,34 @@ async def lessons_for_rule(user_id: int, rule_id: int) -> list[dict]:
     ]
 
 
+async def confirmed_lesson_counts(user_id: int, rule_ids) -> dict[int, int]:
+    """{rule_id: confirmed lessons the caller can read}, for a list of rules.
+
+    `lessons_for_rule` in bulk, counted rather than listed: one GROUP BY, so a
+    rule list can say which rules are proven by situations without a query
+    per row. Rules with none read 0.
+    """
+    from grimoire.services.access import readable_notes_clause
+
+    ids = [int(i) for i in rule_ids or []]
+    out = {i: 0 for i in ids}
+    if not ids:
+        return out
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(LessonRuleLink.rule_id, func.count())
+            .join(Note, Note.id == LessonRuleLink.lesson_id)
+            .where(LessonRuleLink.rule_id.in_(ids))
+            .where(LessonRuleLink.state == CONFIRMED)
+            .where(Note.deleted_at.is_(None))
+            .where(readable_notes_clause(user_id))
+            .group_by(LessonRuleLink.rule_id)
+        )).all()
+    for rule_id, n in rows:
+        out[int(rule_id)] = int(n)
+    return out
+
+
 async def judgments_for_lessons(lesson_ids) -> dict[int, dict]:
     """{lesson_id: {"rule_judgment": linked|no_rule|unjudged, "no_rule"?}}.
 

src/grimoire/services/plugin_context.py

@@ -144,7 +144,7 @@ WRITEPATH_THRESHOLD_KEY = SURFACES["write_path"].floor_key
 WRITEPATH_DEFAULT_ENABLED = True
 WRITEPATH_DEFAULT_THRESHOLD = SURFACES["write_path"].floor_default
 
-# The standing-rule arm gets its own bar, for a third corpus: 0.68 above is
+# The standing-rule arms get their own bar, for a third corpus: 0.68 above is
 # calibrated on code against note PROSE, not code against RULE TEXT.
 #
 # THE STRUCTURAL ARGUMENT, which holds on any install, including one with six
@@ -155,64 +155,27 @@ WRITEPATH_DEFAULT_THRESHOLD = SURFACES["write_path"].floor_default
 #      pool always returns something, so "the best match cleared the bar"
 #      drifts from "a good match exists" toward "N things were ranked". A bar
 #      calibrated for best-of-thousands is cleared by best-of-forty as
-#      arithmetic rather than relevance. (The arms search every tier, which
-#      makes the pool larger; the bar still errs toward silence on purpose.)
+#      arithmetic rather than relevance.
 #   2. Rules are short imperative technical English — a far more HOMOGENEOUS
 #      corpus than note prose, and a more homogeneous corpus has a HIGHER
 #      noise floor than prose's 0.55-0.63.
 #
-# WHY 0.72 AND NOT A NUMBER OFF A HISTOGRAM. The exact offset between prose's
-# floor and rule-text's is not derivable in general — it depends on how an
-# install writes its rules — so the default errs deliberately toward SILENCE
-# rather than toward recall, on an asymmetry that is itself structural: this
-# hint fires on EVERY write. A missed rule is recoverable, because the rule is
-# still in Grimoire and the agent can search it. A hint that cries wolf is not:
-# it teaches the reader to skip the whole block, and the surface is lost along
-# with the true positives it would have carried.
+# ONE BAR FOR EVERY RULE ARM THAT ANNOTATES. The write path, the command arm
+# and the prompt arm read the same key (`RULE_FLOOR_KEY` in retrieval_pipeline):
+# they search the same rules, and what limits each one is its budget and band.
 #
 # TUNE IT FROM YOUR OWN INSTANCE: `retrieval_telemetry` reports
-# `rule_usage.pull_through`. Raise this if rules arrive unread; lower it if
-# rules you needed never arrived. What would RETIRE it: a cross-encoder rerank,
-# which would make a similarity bar the wrong control entirely.
-#
-# This bar is the WRITE-PATH arm's. A code payload is long and rich, which is
-# the case 0.72 suits; the command arm has its own bar below.
+# `rule_usage.pull_through` and `near_miss_samples`. Raise it if rules arrive
+# unread; lower it if rules you needed never arrived. What would RETIRE it: a
+# cross-encoder rerank, which would make a similarity bar the wrong control
+# entirely.
 RULEHINT_THRESHOLD_KEY = SURFACES["write_path_rule"].floor_key
 RULEHINT_DEFAULT_THRESHOLD = SURFACES["write_path_rule"].floor_default
 
-# THE COMMAND ARM'S OWN BAR, AND WHY IT IS NOT THE WRITE PATH'S.
-#
-# A write-path query is a code payload, long and rich, while a pre-tool query
-# is a shell command — often under a dozen words. Less text, less signal, lower
-# scores for the same relevance. At the write path's 0.72 the command arm
-# turns nearly everything away, with its refused mass piled just under the
-# line, which is the shape a bar set too high leaves.
-#
-# WHAT 0.68 IS CALIBRATED AGAINST. Replayed queries, consequential acts
-# against innocuous ones, scored against a rule corpus with symptom-bearing
-# triggers:
-#
-#   0.7571  git push origin dev                 consequential
-#   0.7245  cd ...; git fetch; git add -A       consequential
-#   0.7193  git pull --rebase origin dev        consequential
-#   0.6850  docker compose up -d                consequential
-#   ----------------------------------------  0.68
-#   0.6735  wc -l src/*.py && date              innocuous
-#   0.6544  grep -rn useState src/              innocuous
-#   0.6099  sed -n '120,160p' package.json      innocuous
-#   0.6056  ls -la && cat README.md             innocuous
-#
-# At 0.72 three of the four consequential acts would retrieve nothing, though
-# the right rules ranked first for each.
-#
-# THE SEPARATION IS NARROW (about 0.01), and that is a caveat, not a result.
-# Eight probes set a direction; they do not settle a number.
-# `near_miss_samples` on real traffic is what settles it, and this is the bar
-# to re-read first.
-#
-# WHY A LOWER BAR IS SAFE HERE. The band does noise control downstream: a marginal hit that
-# clears the bar still has to score within `_RULEHINT_BAND` of the top to be
-# rendered. The bar's job shrank, so the bar can.
+# THE COMMAND ARM READS THE SAME BAR. A shell command is short, so it scores
+# lower than a code payload for the same relevance and the shared bar admits
+# less here. The band does noise control downstream: a hit that clears the bar
+# still has to score within `_RULEHINT_BAND` of the top to be rendered.
 #
 # The innocuous scores near the line are largely CROSS-PROJECT BLEED rather
 # than bad ranking: another project's shell-command rule matching a shell
@@ -478,17 +441,11 @@ def _autoinject_query(prompt: str, context: str) -> str:
 # The operator's message is the only query that exists before one is composed,
 # and this arm is what runs against it.
 PROMPTRULE_THRESHOLD_KEY = SURFACES["prompt_rule"].floor_key
-# INHERITED FROM THE ACT ARMS. 0.72 is tuned against
-# code and shell commands. An operator's prose is a different query shape
-# against the same documents, and nothing yet says the two distributions line
-# up — triggers are written in the vocabulary of the MOMENT, which for most
-# rules is act vocabulary, so prose may well score lower across the board.
-#
-# Starting at the act arms' number anyway is deliberate: it is the only value
-# with evidence behind it, and guessing lower would put an unmeasured bar in
-# front of a corpus that binds. Every call is logged under `prompt_rule`, so
-# real traffic settles it — read `near_miss_samples` before moving this, not
-# the percentile alone.
+# THE SAME BAR AS THE ACT ARMS. An operator's prose is a different query shape
+# against the same documents, and triggers are written in the vocabulary of
+# the moment, which for most rules is act vocabulary. Every call is logged
+# under `prompt_rule`, so read `near_miss_samples` before moving the shared
+# bar, not the percentile alone.
 PROMPTRULE_DEFAULT_THRESHOLD = SURFACES["prompt_rule"].floor_default
 
 # ANCHORED ON THIS HOOK'S BUDGET rather than the act arms'. This arm fires
@@ -1171,9 +1128,8 @@ async def get_writepath_config(user_id: int) -> dict:
     independent pairs, resolved from the registry.
 
     Each floor's reason is recorded where its default is: WRITEPATH (code
-    embeddings sit on a much higher similarity floor than prose), RULEHINT (a
-    third corpus again), TOOLRULE (a shell command is a different query shape
-    from a code payload and scores lower for the same relevance).
+    embeddings sit on a much higher similarity floor than prose) and RULEHINT
+    (a third corpus again, whose one bar both rule arms read).
 
     THE BUDGET IS SPLIT TOO. "How many titles at once" does not mean the same
     thing on both surfaces: this arm fires before every Write and Edit while
@@ -1502,14 +1458,10 @@ async def build_write_path_hint(
             logger.warning("write-time derive check failed", exc_info=True)
     staleness: list[str] = []
 
-    # The standing-rule arm further down is deliberately left on the far side
-    # of the guard below: it runs a SEMANTIC search, and moving it above would
-    # run an embedding query on every write in the session.
     # The design arm is decided HERE, above the guard, so a UI write
     # that matched no prior art still carries it — and it returns on its own
-    # rather than joining the guard's condition, because joining it would let
-    # a design line switch the standing-rule arm below on for writes where it
-    # has never run, moving that arm's call distribution under its floor.
+    # rather than joining the guard's condition: a design line alone is not a
+    # reason to render the prior-art menu's frame.
     design_text, design_dedup = await _design_arm(
         user_id, project_id, path, set(exclude_derive or []),
     )
@@ -1520,7 +1472,62 @@ async def build_write_path_hint(
         user_id, project_id, [path], seen=seen_ruling_systems,
         source="rulings_write_path",
     )
-    if not staleness and not synced and not menu and not suggested and not divergence and not derive:
+    # ── Standing rules that may apply here ───────────────────────────────
+    #
+    # A SUGGESTION, not a binding surface, and the distinction is the design:
+    # a rule BINDS by being tagged to an area the project works in,
+    # resolved deterministically at enter_project. This arm reaches for
+    # something weaker and still useful — a conditional rule whose trigger
+    # resembles what is being written, noticed at the moment it is relevant
+    # rather than by being resident in every session.
+    #
+    # EVERY RULE IS ELIGIBLE, including one already named this session: a
+    # rule the session was told about an hour ago is not a rule in front of
+    # the reader now.
+    #
+    # The arm itself is `retrieval_pipeline.run_rule_arm`: the band, the
+    # fresh/repeat split, the unconditional call row and the fresh-only
+    # surfacing rows are written there once for all three
+    # rule arms. What is decided HERE is only what makes this moment this
+    # moment — the query is the code being written (or the path, for an edit
+    # with no body), and a line says the rule may apply "here".
+    #
+    # EVERY WRITE WITH A PATH GETS THE CHECK, whether or not prior art matched:
+    # a rule about the file being written does not depend on a snippet
+    # resembling it. It runs before the notes guard below for that reason,
+    # at the price of one more search per write; its query, `code or path`,
+    # differs from the notes arm's concept query, so it is its own embedding.
+    #
+    # Fails open like every other arm: a rule hint must never break a write.
+    # The direct band, then a rule reached through a linked lesson,
+    # composed by `_rule_moment` exactly as the other two rule builders are.
+    rule_lines: list[str] = []
+    rule_ids: list[int] = []
+    shown_rule_ids: list[int] = []
+    checkpoint: dict = {}
+    try:
+        result = await _rule_moment(
+            rp.WRITE_PATH_RULE,
+            rp.RuleMoment(
+                user_id=user_id, query=code or path, project_id=project_id,
+                where="here", checkpoint_where="here",
+                exclude=frozenset(exclude_rule_ids or []),
+                held=frozenset(held_rule_ids or []),
+            ),
+            floor=cfg["rule_threshold"], budget=cfg["rule_top_k"],
+            checkpoint_floor=cfg["checkpoint_threshold"],
+        )
+        rule_lines = result.lines
+        rule_ids.extend(result.rule_ids)
+        # Every rule line the band showed, repeats included — what the
+        # soft-link recorder pairs with this response's lessons.
+        shown_rule_ids = result.shown_rule_ids
+        checkpoint = result.checkpoint
+    except Exception:
+        logger.debug("write-path rule arm failed", exc_info=True)
+
+    if (not staleness and not synced and not menu and not suggested
+            and not divergence and not derive and not rule_lines):
         if design_text or rulings["lines"]:
             return {
                 **empty,
@@ -1656,52 +1663,7 @@ async def build_write_path_hint(
             user_id=user_id, note_ids=ids, source=arm, project_id=project_id,
         )
 
-    # ── Standing rules that may apply here ───────────────────────────────
-    #
-    # A SUGGESTION, not a binding surface, and the distinction is the design:
-    # a rule BINDS by being tagged to an area the project works in,
-    # resolved deterministically at enter_project. This arm reaches for
-    # something weaker and still useful — a conditional rule whose trigger
-    # resembles what is being written, noticed at the moment it is relevant
-    # rather than by being resident in every session.
-    #
-    # EVERY RULE IS ELIGIBLE, including one already named this session: a
-    # rule the session was told about an hour ago is not a rule in front of
-    # the reader now.
-    #
-    # The arm itself is `retrieval_pipeline.run_rule_arm`: the band, the
-    # fresh/repeat split, the unconditional call row and the fresh-only
-    # surfacing rows are written there once for all three
-    # rule arms. What is decided HERE is only what makes this moment this
-    # moment — the query is the code being written (or the path, for an edit
-    # with no body), and a line says the rule may apply "here".
-    #
-    # Fails open like every other arm: a rule hint must never break a write.
-    # The direct band, then a rule reached through a linked lesson,
-    # composed by `_rule_moment` exactly as the other two rule builders are.
-    rule_ids: list[int] = []
-    shown_rule_ids: list[int] = []
-    checkpoint: dict = {}
-    try:
-        result = await _rule_moment(
-            rp.WRITE_PATH_RULE,
-            rp.RuleMoment(
-                user_id=user_id, query=code or path, project_id=project_id,
-                where="here", checkpoint_where="here",
-                exclude=frozenset(exclude_rule_ids or []),
-                held=frozenset(held_rule_ids or []),
-            ),
-            floor=cfg["rule_threshold"], budget=cfg["rule_top_k"],
-            checkpoint_floor=cfg["checkpoint_threshold"],
-        )
-        lines.extend(result.lines)
-        rule_ids.extend(result.rule_ids)
-        # Every rule line the band showed, repeats included — what the
-        # soft-link recorder pairs with this response's lessons.
-        shown_rule_ids = result.shown_rule_ids
-        checkpoint = result.checkpoint
-    except Exception:
-        logger.debug("write-path rule arm failed", exc_info=True)
+    lines.extend(rule_lines)
 
     # A lesson and a rule on this one response: the soft-link
     # recorder counts the pair, keyed on the FILE — every edit to one file is

src/grimoire/services/retrieval_pipeline.py

@@ -460,16 +460,31 @@ class RuleArm:
     declared: Declared | None = None
 
 
-# The arms differ on purpose where these flags differ. Whether the prompt arm
-# should band, and whether the act arms should reserve a preference, are open
-# questions to settle on telemetry, not by symmetry.
+# ONE BAR FOR EVERY RULE ARM THAT ANNOTATES. The prompt, command and write
+# arms search the same corpus, and what limits what they show is the budget
+# and the band, not the floor. The floor only answers "is this worth ranking
+# at all", and that question has one answer per corpus, not one per query
+# shape — so the three share one settings key, and moving it moves all three.
+# `set_dial` records the move on every surface that shares the key.
+#
+# Set low on purpose: a hit under it is never shown, so a bar set too high
+# costs a rule that binds, while one set a little low costs a line the band
+# usually drops. The reply backstop keeps its own number, because there the
+# floor is a stop bar and not a hint bar.
+RULE_FLOOR_KEY = "kb_rule_threshold"
+RULE_FLOOR_DEFAULT = 0.70
+
+# The arms differ on purpose where these flags differ: the prompt arm takes no
+# band, so its budget alone decides how many hits it shows, and the act arms
+# band and may hold the act. Every arm that annotates reserves a line for a
+# preference.
 PROMPT_RULE = RuleArm(
     "prompt_rule", band=False, compact_tail=False, checkpoint=False,
     preference_slot=True,
     tuning=Surface(
         name="prompt_rule",
-        floor_key="kb_promptrule_threshold",
-        floor_default=0.72,
+        floor_key=RULE_FLOOR_KEY,
+        floor_default=RULE_FLOOR_DEFAULT,
         budget_key="kb_promptrule_top_k",
         budget_default=3,
         asks="the operator's message, against rule triggers",
@@ -480,11 +495,11 @@ PROMPT_RULE = RuleArm(
 )
 PRE_TOOL_RULE = RuleArm(
     "pre_tool_rule", band=True, compact_tail=True, checkpoint=True,
-    preference_slot=False,
+    preference_slot=True,
     tuning=Surface(
         name="pre_tool_rule",
-        floor_key="kb_toolrule_threshold",
-        floor_default=0.68,
+        floor_key=RULE_FLOOR_KEY,
+        floor_default=RULE_FLOOR_DEFAULT,
         budget_key="kb_toolrule_top_k",
         budget_default=5,
         asks="the command about to run, against rule triggers",
@@ -495,11 +510,11 @@ PRE_TOOL_RULE = RuleArm(
 )
 WRITE_PATH_RULE = RuleArm(
     "write_path_rule", band=True, compact_tail=True, checkpoint=True,
-    preference_slot=False,
+    preference_slot=True,
     tuning=Surface(
         name="write_path_rule",
-        floor_key="kb_rulehint_threshold",
-        floor_default=0.72,
+        floor_key=RULE_FLOOR_KEY,
+        floor_default=RULE_FLOOR_DEFAULT,
         budget_key="kb_rulehint_top_k",
         budget_default=5,
         asks="the code being written, against rule triggers",
@@ -689,6 +704,13 @@ async def _reserve_slot_for_preference(
     same shape one corpus over, where snippets lose to project records that
     merely resemble the query.
 
+    The act arms reserve the slot for the same reason one step later. A
+    preference about how to act ("let each action land before the next") is
+    ranked against every rule whose trigger names the same command, and the
+    act is its last chance: once the command has run there is nothing left
+    for it to shape. The slot's query reuses the arm's query, so it costs one
+    kind-filtered search and no extra embedding.
+
     THE SLOT BUYS POSITION, NOT A LOWER BAR — same as `reuse_slot`, which also
     reserves at `cfg["threshold"]`. A weak preference cannot buy the slot, so
     silence stays the default and the reserved line is never worse than the

src/grimoire/services/retrieval_surfaces.py

@@ -110,6 +110,9 @@ def dial_for_key(key: str) -> tuple[str, str] | None:
 
     Derived rather than listed so a seventh surface is covered the moment it is
     added here, which is the only way this stays true.
+
+    A floor key several surfaces share resolves to the first of them;
+    `set_dial` records the move on every surface in `sharing_floor`.
     """
     for surface in SURFACES.values():
         if key == surface.floor_key:
@@ -119,6 +122,16 @@ def dial_for_key(key: str) -> tuple[str, str] | None:
     return None
 
 
+def sharing_floor(name: str) -> list[str]:
+    """Every surface that reads the same floor key as `name`, itself included.
+
+    The rule arms share one bar, so moving one surface's floor moves its
+    siblings, and each of them needs the event that says so.
+    """
+    key = get_surface(name).floor_key
+    return [n for n, s in SURFACES.items() if s.floor_key == key]
+
+
 async def floor_for(user_id: int, name: str) -> float:
     """This install's current floor for a surface, clamped to [0, 1]."""
     s = get_surface(name)

src/grimoire/services/retrieval_telemetry.py

@@ -803,6 +803,9 @@ def _silent_surfaces(sources: dict, usage: dict, rule_usage: dict,
         return []
     seen = set(sources) | set((usage.get("by_source") or {}))
     seen |= set((rule_usage.get("by_source") or {}))
+    # A getter shows nothing, so it is seen only by its pulls.
+    seen |= set((usage.get("pulled_by_source") or {}))
+    seen |= set((rule_usage.get("pulled_by_source") or {}))
     seen |= set(((system_usage or {}).get("by_source") or {}))
     return [
         {"source": s, "kind": POINTS[s].kind, "what": POINTS[s].what}
@@ -1109,7 +1112,7 @@ async def retrieval_summary(
 
     # Assigned inside the try below; named here so the readout can tell
     # "this query failed" from "this window has no rows".
-    by_source_rows = None
+    by_source_rows = rule_by_source_rows = None
     rule_rows = None
     distinct_rules_surfaced = distinct_rules_pulled = 0
     # None means the coverage read did not happen — distinct from a table with
@@ -1426,6 +1429,53 @@ async def retrieval_summary(
                 rule_rows = None
                 distinct_rules_surfaced = distinct_rules_pulled = 0
                 distinct_rules_acted = 0
+
+            # The note block's per-source join, at the rule grain. The
+            # top-level rule ratio blends every arm — the prompt arm, the tool
+            # arm, a lesson pointing at its rule — and "which of them earns its
+            # lines" is the question each arm's bar is tuned against. Its own
+            # guard, for the reason the note join has one.
+            try:
+                rule_pulled_ids = (
+                    select(RuleUsageEvent.rule_id)
+                    .where(
+                        RuleUsageEvent.created_at >= since,
+                        RuleUsageEvent.user_id == user_id,
+                        RuleUsageEvent.event == RULE_PULLED,
+                        RuleUsageEvent.source.startswith("mcp_", autoescape=True),
+                    )
+                    .distinct()
+                    .subquery()
+                )
+                rule_surfaced_pairs = (
+                    select(RuleUsageEvent.source, RuleUsageEvent.rule_id)
+                    .where(
+                        RuleUsageEvent.created_at >= since,
+                        RuleUsageEvent.user_id == user_id,
+                        RuleUsageEvent.event == RULE_SURFACED,
+                    )
+                    .distinct()
+                    .subquery()
+                )
+                rule_by_source_rows = (
+                    await session.execute(
+                        select(
+                            rule_surfaced_pairs.c.source,
+                            func.count().label("rules_surfaced"),
+                            func.count(rule_pulled_ids.c.rule_id).label("rules_pulled"),
+                        )
+                        .select_from(
+                            rule_surfaced_pairs.outerjoin(
+                                rule_pulled_ids,
+                                rule_pulled_ids.c.rule_id == rule_surfaced_pairs.c.rule_id,
+                            )
+                        )
+                        .group_by(rule_surfaced_pairs.c.source)
+                    )
+                ).all()
+            except Exception:
+                logger.warning("per-source rule pull-through read failed", exc_info=True)
+                rule_by_source_rows = None
     except Exception:
         logger.warning("retrieval summary read failed", exc_info=True)
         out["read_failed"] = True
@@ -1439,6 +1489,9 @@ async def retrieval_summary(
         "distinct_notes_surfaced": int(distinct_surfaced or 0),
         "distinct_notes_pulled": int(distinct_pulled or 0),
     }
+    # Pulls per door. `by_source` is keyed by the surface that showed a record
+    # and leaves pulls out, so this is what says each getter is recording.
+    pulled_by_source: dict[str, int] = {}
     for event, source, n in urows:
         n = int(n)
         if event == SURFACED:
@@ -1448,6 +1501,7 @@ async def retrieval_summary(
                 usage["surfaced"] += n
         elif event == PULLED:
             usage["pulled"] += n
+            pulled_by_source[source] = pulled_by_source.get(source, 0) + n
             # The mcp_/rest_ split is load-bearing (see NoteUsageEvent's own
             # comment). "Is
             # this record dead weight?" is answered by ANY pull; "was that
@@ -1466,6 +1520,7 @@ async def retrieval_summary(
         round(usage["pulled_by_agent"] / usage["surfaced"], 4)
         if usage["surfaced"] else None
     )
+    usage["pulled_by_source"] = pulled_by_source
 
     # The same question, per surface — which is the one the top-level ratio
     # cannot answer. A corpus average of 0.05 is compatible with one surface
@@ -1538,6 +1593,8 @@ async def retrieval_summary(
         "applied": 0,
         "departed": 0,
         "distinct_rules_acted": int(distinct_rules_acted or 0),
+        # Pulls per door, as on the note block.
+        "pulled_by_source": {},
     }
     if rule_rows is None:
         # The FLAG is added, the shape is kept — matching `by_source_failed`
@@ -1560,6 +1617,8 @@ async def retrieval_summary(
                     rule_usage["surfaced"] += n
             elif event == RULE_PULLED:
                 rule_usage["pulled"] += n
+                pulled = rule_usage["pulled_by_source"]
+                pulled[source] = pulled.get(source, 0) + n
                 # Same split, and it carries MORE weight here than for notes.
                 # The arm's whole claim is "this rule may apply to what you are
                 # writing", and only an agent opening it says the claim landed.
@@ -1596,6 +1655,28 @@ async def retrieval_summary(
         round(rule_usage["pulled_by_agent"] / rule_usage["surfaced"], 4)
         if rule_usage["surfaced"] else None
     )
+    # Per surface, the note block's shape and its caveat: a pull records the
+    # door, not the surface that led to it, so a rule two arms surfaced and an
+    # agent opened once counts as pulled for both. An upper bound — a surface
+    # reading near zero here is not being flattered.
+    if rule_by_source_rows is None:
+        rule_usage["by_source"] = {}
+        rule_usage["by_source_failed"] = True
+    else:
+        rule_usage["by_source"] = {
+            source: {
+                "rules_surfaced": int(n_surfaced or 0),
+                "rules_pulled": int(n_pulled or 0),
+                # None on an ambient surface, as for notes: nothing chose
+                # those rules, so the ratio would be misread.
+                "pull_through": (
+                    None if is_ambient(source) or not n_surfaced
+                    else round(int(n_pulled or 0) / int(n_surfaced), 4)
+                ),
+                "ambient": is_ambient(source),
+            }
+            for source, n_surfaced, n_pulled in rule_by_source_rows
+        }
     rule_usage.update(_coverage((rule_complete or {}).get("*"), since))
     out["rule_usage"] = rule_usage
     out["system_usage"] = await _system_usage(user_id, since)

src/grimoire/services/retrieval_tuning.py

@@ -44,6 +44,7 @@ from grimoire.services.retrieval_surfaces import (
     budget_for,
     floor_for,
     get_surface,
+    sharing_floor,
     surface_names,
 )
 from grimoire.services.settings import set_setting
@@ -257,6 +258,13 @@ async def set_dial(
             "reason": text, "actor": actor, "unchanged": True,
         }
 
+    # A shared floor moves every surface that reads it, and each one's history
+    # has to show the move, or its telemetry changes with no event to explain it.
+    # The surface the caller named is recorded first, its siblings after it.
+    moved = [surface] + (
+        [n for n in sharing_floor(surface) if n != surface] if dial == "floor" else []
+    )
+
     await set_setting(user_id, key, stored)
     async with async_session() as session:
         # Stamped with the space this number was chosen in. Read at
@@ -264,12 +272,13 @@ async def set_dial(
         # the caller recording it are the same call, so there is no window in
         # which they could disagree.
         stamp = calibration_stamp()
-        session.add(RetrievalTuningEvent(
-            user_id=user_id, surface=surface, dial=dial,
-            old_value=old, new_value=applied, actor=actor, reason=text,
-            embedding_model=stamp["embedding_model"],
-            shape_version=stamp["shape_version"],
-        ))
+        for name in moved:
+            session.add(RetrievalTuningEvent(
+                user_id=user_id, surface=name, dial=dial,
+                old_value=old, new_value=applied, actor=actor, reason=text,
+                embedding_model=stamp["embedding_model"],
+                shape_version=stamp["shape_version"],
+            ))
         await session.commit()
 
     return {
@@ -286,6 +295,9 @@ async def set_dial(
         # Always present, both ways round: a caller that has to test for the
         # key's absence to learn the answer will eventually forget to.
         "unchanged": False,
+        # The other surfaces this move carried with it: a shared floor is one
+        # number, and a caller tuning one rule arm has tuned them all.
+        "moved_with": [n for n in moved if n != surface],
     }
 
 

src/grimoire/services/rule_usage.py

@@ -59,8 +59,9 @@ as short and as slow-moving as possible.
 from __future__ import annotations
 
 import logging
+from datetime import datetime
 
-from sqlalchemy import case, func, select
+from sqlalchemy import case, func, select, true
 
 from grimoire.models import async_session
 from grimoire.models.base import iso
@@ -333,13 +334,18 @@ def empty_rule_usage() -> dict:
     }
 
 
-async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
+async def usage_for_rules(
+    rule_ids: list[int], since: datetime | None = None,
+) -> dict[int, dict]:
     """Aggregate usage for a set of rules: {rule_id: {counts + timestamps}}.
 
     One GROUP BY for the whole page rather than a query per row — this feeds a
     list view, so the per-row shape would be N+1 by construction. Rules with no
     events come back with `empty_rule_usage()`, so the caller never has to tell
     "no events" from "not in the result".
+
+    `since` limits the counts to events at or after it, so two windows can be
+    compared; without it they cover everything recorded.
     """
     ids = [int(r) for r in rule_ids]
     out: dict[int, dict] = {rid: empty_rule_usage() for rid in ids}
@@ -370,6 +376,10 @@ async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
                         ambient,
                     )
                     .where(RuleUsageEvent.rule_id.in_(ids))
+                    .where(
+                        RuleUsageEvent.created_at >= since
+                        if since is not None else true()
+                    )
                     .group_by(
                         RuleUsageEvent.rule_id,
                         RuleUsageEvent.event,

tests/test_integration_lesson_rule_links.py

@@ -93,6 +93,22 @@ async def test_naming_rules_confirms_them_and_both_sides_read_the_link(world):
     assert [(l["id"], l["state"]) for l in lessons] == [(world["lesson"], "confirmed")]
 
 
+async def test_a_rule_list_counts_confirmed_lessons_only(world):
+    """The bulk count a rule list reads: a rejected link is not a lesson
+    proving the rule, a rule with no links reads 0, and a stranger counts
+    nothing through a lesson they cannot read."""
+    owner = world["owner"]
+    await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])
+    await links_svc.judge_link(owner, world["lesson"], world["r2"], "reject", "not it")
+    rules = [world["r1"], world["r2"], world["foreign"]]
+
+    assert await links_svc.confirmed_lesson_counts(owner, rules) == {
+        world["r1"]: 1, world["r2"]: 0, world["foreign"]: 0,
+    }
+    assert (await links_svc.confirmed_lesson_counts(world["stranger"], rules))[world["r1"]] == 0
+    assert await links_svc.confirmed_lesson_counts(owner, []) == {}
+
+
 async def test_a_rule_left_out_of_the_set_is_rejected_not_forgotten(world):
     owner = world["owner"]
     await links_svc.set_lesson_rules(owner, world["lesson"], [world["r1"], world["r2"]])

tests/test_mcp_tool_rulebooks.py

@@ -573,3 +573,36 @@ def test_rule_outcome_is_registered_and_is_not_read_only():
     assert "rule_outcome" in mcp.names
     assert "rule_outcome" in _WRITE_TOOLS
     assert "rule_outcome" not in _READ_ONLY_TOOLS
+
+
+@pytest.mark.asyncio
+async def test_list_rules_carries_usage_only_when_asked():
+    """A listing is for applying rules, so it stays lean; `with_usage` adds
+    each rule's telemetry and confirmed-lesson count for judging them, with
+    `days` passed down as a window."""
+    from grimoire.mcp.tools import rulebooks as tools
+    from grimoire.services.rule_usage import empty_rule_usage
+
+    rows = [fake_rule(id=1, title="dev is home"), fake_rule(id=2, title="ask first")]
+    seen = {**empty_rule_usage(), "surfaced_count": 4, "pull_count": 1}
+    usage = AsyncMock(return_value={1: seen})
+    lessons = AsyncMock(return_value={1: 3})
+    with patch.object(tools.rulebooks_svc, "list_rules", AsyncMock(return_value=rows)), \
+         patch.object(tools.rulebooks_svc, "list_rule_moments", AsyncMock(return_value={})), \
+         patch.object(tools, "usage_for_rules", usage), \
+         patch.object(tools, "confirmed_lesson_counts", lessons):
+        lean = await tools.list_rules()
+        assert all("usage" not in r and "confirmed_lessons" not in r for r in lean["rules"])
+        usage.assert_not_awaited()
+
+        out = await tools.list_rules(with_usage=True, days=14)
+
+    by_id = {r["id"]: r for r in out["rules"]}
+    assert by_id[1]["usage"]["surfaced_count"] == 4
+    assert by_id[1]["confirmed_lessons"] == 3
+    # A rule missing from either read is zero-filled, never absent.
+    assert by_id[2]["usage"] == empty_rule_usage()
+    assert by_id[2]["confirmed_lessons"] == 0
+    assert usage.await_args.args[0] == [1, 2]
+    assert usage.await_args.kwargs["since"] is not None
+    lessons.assert_awaited_once_with(7, [1, 2])

tests/test_retrieval_surfaces.py

@@ -71,13 +71,29 @@ def test_every_surface_name_is_a_real_telemetry_source():
     )
 
 
-def test_no_two_surfaces_share_a_settings_key():
-    """One dial, one label."""
-    floors = [s.floor_key for s in rs.SURFACES.values()]
+def test_a_shared_settings_key_is_one_bar_with_one_default():
+    """One dial, one label. Surfaces may share a floor key, as the rule arms
+    do, but then they must ship the same default: otherwise the one number
+    means something different on each surface until somebody sets it."""
+    by_key: dict[str, set[float]] = {}
+    for s in rs.SURFACES.values():
+        by_key.setdefault(s.floor_key, set()).add(s.floor_default)
+    split = {k: v for k, v in by_key.items() if len(v) > 1}
+    assert not split, f"a shared floor key ships two defaults: {split}"
+
     budgets = [s.budget_key for s in rs.SURFACES.values()]
-    assert len(set(floors)) == len(floors), f"duplicate floor key in {floors}"
     assert len(set(budgets)) == len(budgets), f"duplicate budget key in {budgets}"
-    assert not (set(floors) & set(budgets)), "a floor key doubles as a budget key"
+    assert not (set(by_key) & set(budgets)), "a floor key doubles as a budget key"
+
+
+def test_the_rule_arms_that_annotate_share_one_bar_and_the_reply_stop_does_not():
+    from grimoire.services.retrieval_pipeline import RULE_FLOOR_KEY
+
+    assert sorted(rs.sharing_floor("prompt_rule")) == [
+        "pre_tool_rule", "prompt_rule", "write_path_rule",
+    ]
+    assert rs.SURFACES["prompt_rule"].floor_key == RULE_FLOOR_KEY
+    assert rs.sharing_floor("reply_rule") == ["reply_rule"]
 
 
 def test_every_surface_says_what_it_asks_over_what_and_how_often():

tests/test_retrieval_tuning.py

@@ -86,9 +86,9 @@ async def test_a_good_change_writes_the_setting_and_the_event():
 
     setter.assert_awaited_once()
     _uid, key, stored = setter.await_args.args
-    assert key == "kb_promptrule_threshold" and stored == "0.66"
+    assert key == "kb_rule_threshold" and stored == "0.66"
 
-    event = session.add.call_args.args[0]
+    event = session.add.call_args_list[0].args[0]
     assert event.surface == "prompt_rule" and event.dial == "floor"
     # The before-value is what makes the history answerable.
     assert event.old_value == 0.72 and event.new_value == 0.66
@@ -97,6 +97,37 @@ async def test_a_good_change_writes_the_setting_and_the_event():
     assert out["clamped"] is False
 
 
+@pytest.mark.asyncio
+async def test_a_shared_floor_records_the_move_on_every_surface_that_reads_it():
+    """One key, one write, an event per surface. A rule arm whose bar moved
+    with no event of its own would show a telemetry shift nothing explains."""
+    session, ctx = _patches(floor=0.70)
+    with ctx[0], patch.object(rt, "set_setting", AsyncMock()) as setter, \
+         ctx[2], ctx[3]:
+        out = await rt.set_dial(
+            1, "pre_tool_rule", "floor", 0.74,
+            reason="read the 10 highest declines: every one was a real miss",
+        )
+
+    setter.assert_awaited_once()
+    recorded = {c.args[0].surface for c in session.add.call_args_list}
+    assert recorded == {"pre_tool_rule", "prompt_rule", "write_path_rule"}
+    assert sorted(out["moved_with"]) == ["prompt_rule", "write_path_rule"]
+
+
+@pytest.mark.asyncio
+async def test_a_budget_moves_only_its_own_surface():
+    session, ctx = _patches(budget=3)
+    with ctx[0], patch.object(rt, "set_setting", AsyncMock()), ctx[2], ctx[3]:
+        out = await rt.set_dial(
+            1, "prompt_rule", "budget", 4,
+            reason="on-point hits sit at ranks 4 and 5 as often as rank 1",
+        )
+
+    assert [c.args[0].surface for c in session.add.call_args_list] == ["prompt_rule"]
+    assert out["moved_with"] == []
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("dial, sent, applied", [
     ("floor", 1.4, 1.0),

tests/test_retrieval_warnings.py

@@ -279,6 +279,19 @@ def test_a_point_seen_only_in_usage_counts_as_having_emitted() -> None:
     assert "enter_project" not in quiet
 
 
+def test_a_getter_seen_only_by_its_pulls_counts_as_having_emitted() -> None:
+    """A getter surfaces nothing, so no `by_source` ever names it; its pulls
+    are the only sign it records. Without them every pull door reads silent
+    on every active window."""
+    quiet = {p["source"] for p in _silent_surfaces(
+        {}, {"pulled_by_source": {"mcp_get_note": 3}},
+        {"pulled_by_source": {"mcp_get_rule": 2}}, active=True)}
+    assert "mcp_get_note" not in quiet
+    assert "mcp_get_rule" not in quiet
+    # A door with no pulls is still reported.
+    assert "mcp_get_snippet" in quiet
+
+
 # ── settings parsing ──────────────────────────────────────────────────────
 
 @pytest.mark.parametrize("raw,fallback,want", [

tests/test_routes_retrieval_tuning.py

@@ -76,10 +76,15 @@ def test_the_route_does_not_take_the_actor_from_the_caller():
 def test_every_registry_key_is_recognised_as_a_dial():
     """Derived from the registry, both directions. A seventh surface added
     without a row here is a floor that can be written silently again."""
-    from grimoire.services.retrieval_surfaces import SURFACES, dial_for_key
+    from grimoire.services.retrieval_surfaces import (
+        SURFACES, dial_for_key, sharing_floor,
+    )
 
     for name, surface in SURFACES.items():
-        assert dial_for_key(surface.floor_key) == (name, "floor")
+        # A shared floor resolves to one of the surfaces that read it, and
+        # `set_dial` records the move on all of them.
+        owner, dial = dial_for_key(surface.floor_key)
+        assert dial == "floor" and name in sharing_floor(owner)
         assert dial_for_key(surface.budget_key) == (name, "budget")
 
 

tests/test_rule_usage_wiring.py

@@ -33,16 +33,8 @@ pytestmark = pytest.mark.usefixtures("_bind_user")
 # each of these does.
 
 
-# The write-path hint returns early when a write matched nothing at all — no
-# staleness, no synced record, no prior-art menu, no shape signal. The rule arm
-# sits deliberately on the FAR side of that guard, because it runs a semantic
-# search and moving it above would mean an embedding query on every write in
-# the session (which is why its gating is a separate question from
-# precision).
-#
-# So a fixture that stubs every other arm to empty never reaches the rule arm
-# at all — which is what the first run of this file did. The note hit below is
-# not decoration: it is the condition the arm requires in order to fire.
+# A prior-art hit for the fixtures that want the menu beside the rule lines.
+# The rule arm runs on every write with a path, whether or not this matched.
 _PRIOR_ART = [(0.72, fake_note(id=9, title="debounce helper", user_id=1,
                                note_type="snippet"))]
 
@@ -161,7 +153,8 @@ async def test_the_arm_searches_on_its_OWN_bar_not_the_code_one():
             1, "frontend/src/api/client.ts", code="x" * 400,
         )
 
-    kw = search.await_args.kwargs
+    # The first search is the arm's own; the preference slot asks after it.
+    kw = search.await_args_list[0].kwargs
     assert kw["threshold"] == 0.81, "the arm is still using the code threshold"
     assert kw["limit"] == pc.RULEHINT_LIMIT
     # NO CATEGORY FILTER, on the axis that exists: the act arms narrow by
@@ -179,21 +172,16 @@ async def test_the_arm_searches_on_its_OWN_bar_not_the_code_one():
 
 
 @pytest.mark.asyncio
-async def test_the_arm_does_not_fire_on_a_write_that_matched_nothing():
-    """The gate, pinned — because the fixture above now depends on it and a
-    silent change would make every other test here pass vacuously.
-
-    A write matching no prior art returns before the rule arm runs. That is
-    deliberate: the arm is a semantic search, and ungating it means an
-    embedding query on every write in the session. The gate stays until the
-    arm's precision is fixed, so this failing is a reason to look at precision
-    rather than to update the assertion.
-    """
+async def test_a_write_that_matched_no_prior_art_still_gets_its_rule_check():
+    """A rule about the file being written does not depend on a snippet
+    resembling it, so the arm runs before the prior-art guard, not behind it."""
     rec = MagicMock()
     hits = [(0.71, fake_rule(id=156, title="A wait with no deadline is a bug"))]
-    await _run_arm(hits, rec, prior_art=[])
+    out = await _run_arm(hits, rec, prior_art=[])
 
-    assert rec.call_count == 0
+    assert rec.call_args.kwargs["rule_ids"] == [156]
+    assert "get_rule(156)" in out["context"]
+    assert out["rule_ids"] == [156]
 
 
 @pytest.mark.asyncio
@@ -834,10 +822,11 @@ async def test_the_tool_arm_logs_the_call_that_found_nothing():
     out = await _run_tool_arm([], rec, retrieval_log=log)
 
     assert out == {"context": "", "rule_ids": [], "checkpoint": {}}
-    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 arm's own row; the preference slot logs beside it under its own name.
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == "pre_tool_rule"]
+    assert len(rows) == 1
+    assert rows[0].kwargs["results"] == []
+    assert rows[0].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()
@@ -850,8 +839,9 @@ async def test_the_tool_arm_logs_a_call_whose_only_hit_was_already_shown():
     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"] == []
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == "pre_tool_rule"]
+    assert len(rows) == 1
+    assert rows[0].kwargs["results"] == []
     rec.assert_not_called()
 
 
@@ -982,14 +972,21 @@ async def test_a_genuine_ranker_decline_reports_zero_suppression():
     assert row.kwargs["suppressed"] is not None
 
 
+def _tool_row(log):
+    """The tool arm's own call row; the preference slot logs beside it."""
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == "pre_tool_rule"]
+    assert len(rows) == 1, rows
+    return rows[0].kwargs
+
+
 @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
+    assert _tool_row(log)["results"] == []
+    assert _tool_row(log)["suppressed"] == 1
 
 
 @pytest.mark.asyncio
@@ -1002,8 +999,8 @@ async def test_a_shown_hit_is_not_counted_as_suppressed():
     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
+    assert _tool_row(log)["suppressed"] == 1
+    assert len(_tool_row(log)["results"]) == 1
 
 
 # ── The identity between the two tables ─────────────────────────────────

tests/test_services_retrieval_telemetry.py

@@ -618,6 +618,88 @@ async def test_the_preload_lands_in_ambient_and_never_in_the_ratio(_dispose_engi
         await cleanup()
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_rule_by_source_says_which_arm_earns_its_lines(_dispose_engine):
+    """The rule ratio blends every arm. Per surface it can say which one is
+    working: distinct rules per arm, the agent opens among them, and no ratio
+    on a bulk delivery."""
+    from grimoire.services.retrieval_telemetry import retrieval_summary
+
+    cleanup = await _rule_events(990016, [
+        # The write arm chose rule 5301 twice — one rule — and 5302 once.
+        (5301, "surfaced", "write_path_rule"),
+        (5301, "surfaced", "write_path_rule"),
+        (5302, "surfaced", "write_path_rule"),
+        # The tool arm chose 5303 and nothing came of it.
+        (5303, "surfaced", "pre_tool_rule"),
+        # The preload carried 5301 as well.
+        (5301, "surfaced", "session_start"),
+        # An agent opened 5301; a person opened 5303, which is not a hint
+        # landing.
+        (5301, "pulled", "mcp_get_rule"),
+        (5303, "pulled", "rest_rule"),
+    ])
+    try:
+        ru = (await retrieval_summary(990016, days=30))["rule_usage"]
+        assert "by_source_failed" not in ru, "the join did not execute"
+        by_source = ru["by_source"]
+
+        wp = by_source["write_path_rule"]
+        assert (wp["rules_surfaced"], wp["rules_pulled"]) == (2, 1)
+        assert wp["pull_through"] == 0.5
+        assert wp["ambient"] is False
+
+        tool = by_source["pre_tool_rule"]
+        assert (tool["rules_surfaced"], tool["rules_pulled"]) == (1, 0)
+        assert tool["pull_through"] == 0.0
+
+        preload = by_source["session_start"]
+        # The pull is counted against every surface that showed the rule,
+        # the upper bound the readout states.
+        assert (preload["rules_surfaced"], preload["rules_pulled"]) == (1, 1)
+        assert preload["pull_through"] is None
+        assert preload["ambient"] is True
+
+        assert "mcp_get_rule" not in by_source, "a pull is not a surface"
+    finally:
+        await cleanup()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_a_pull_door_and_a_rules_listing_are_not_reported_silent(
+    _dispose_engine,
+):
+    """One note pulled through its getter and one rules payload on get_task:
+    both points emitted, so neither may be named as silent."""
+    from sqlalchemy import delete
+
+    from grimoire.models import async_session
+    from grimoire.models.note_usage import NoteUsageEvent
+    from grimoire.services.retrieval_telemetry import retrieval_summary
+
+    UID = 990017
+    cleanup = await _rule_events(UID, [(5401, "surfaced", "get_task")])
+    async with async_session() as s:
+        s.add_all([
+            NoteUsageEvent(user_id=UID, note_id=11, event="surfaced", source="auto_inject"),
+            NoteUsageEvent(user_id=UID, note_id=11, event="pulled", source="mcp_get_note"),
+        ])
+        await s.commit()
+    try:
+        out = await retrieval_summary(UID, days=30)
+        assert out["usage"]["pulled_by_source"] == {"mcp_get_note": 1}
+        quiet = {p["source"] for p in out["silent_surfaces"]}
+        assert "mcp_get_note" not in quiet
+        assert "get_task" not in quiet
+    finally:
+        await cleanup()
+        async with async_session() as s:
+            await s.execute(delete(NoteUsageEvent).where(NoteUsageEvent.user_id == UID))
+            await s.commit()
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_ambient_alone_reports_no_ratio(_dispose_engine):

tests/test_services_rule_usage.py

@@ -198,6 +198,47 @@ async def test_usage_for_rules_aggregates_per_rule(_dispose_engine):
             await s.commit()
 
 
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_usage_for_rules_counts_only_the_window_it_is_given(_dispose_engine):
+    """Two windows have to be comparable, so an event before `since` is left
+    out of every count — and without `since` it is counted."""
+    from datetime import datetime, timedelta, timezone
+
+    from sqlalchemy import delete
+
+    from grimoire.models import async_session
+    from grimoire.models.rule_usage import RuleUsageEvent
+
+    old = datetime.now(timezone.utc) - timedelta(days=30)
+    async with async_session() as s:
+        s.add_all([
+            RuleUsageEvent(user_id=990022, rule_id=6011, event=SURFACED,
+                           source="write_path_rule", created_at=old),
+            RuleUsageEvent(user_id=990022, rule_id=6011, event=PULLED,
+                           source="mcp_get_rule", created_at=old),
+            RuleUsageEvent(user_id=990022, rule_id=6011, event=SURFACED,
+                           source="write_path_rule"),
+        ])
+        await s.commit()
+    try:
+        since = datetime.now(timezone.utc) - timedelta(days=14)
+        recent = (await rule_usage.usage_for_rules([6011], since=since))[6011]
+        assert recent["surfaced_count"] == 1
+        assert recent["pull_count"] == 0
+        assert recent["last_pulled_at"] is None
+
+        everything = (await rule_usage.usage_for_rules([6011]))[6011]
+        assert everything["surfaced_count"] == 2
+        assert everything["pull_count"] == 1
+    finally:
+        async with async_session() as s:
+            await s.execute(
+                delete(RuleUsageEvent).where(RuleUsageEvent.user_id == 990022)
+            )
+            await s.commit()
+
+
 @pytest.mark.integration
 @pytest.mark.asyncio
 async def test_usage_for_rules_on_an_empty_id_list_asks_the_database_nothing(

tests/test_settings_defaults_agree.py

@@ -41,7 +41,7 @@ ROOT = pathlib.Path(__file__).resolve().parents[1]
 _SERVICES = ROOT / "src" / "grimoire" / "services"
 _VUE = ROOT / "frontend" / "src" / "views" / "SettingsView.vue"
 
-# THE SIX RETRIEVAL FLOORS live in one registry, so their side of
+# THE RETRIEVAL FLOORS live in one registry, so their side of
 # the pairing is a SURFACE NAME rather than a module constant. That is strictly
 # better for this guard: the surface name is also the telemetry source, so a row
 # here names the same arm the readout does, and a floor that moved cannot be
@@ -49,9 +49,10 @@ _VUE = ROOT / "frontend" / "src" / "views" / "SettingsView.vue"
 _SURFACE_PAIRS = (
     ("auto_inject", "kbInjectThreshold"),
     ("write_path", "kbWritePathThreshold"),
-    ("write_path_rule", "kbRuleHintThreshold"),
-    ("pre_tool_rule", "kbToolRuleThreshold"),
-    ("prompt_rule", "kbPromptRuleThreshold"),
+    # The three rule hints read one bar, so one control serves them.
+    ("write_path_rule", "kbRuleThreshold"),
+    ("pre_tool_rule", "kbRuleThreshold"),
+    ("prompt_rule", "kbRuleThreshold"),
     ("reply_rule", "kbReplyRuleThreshold"),
 )
 

tests/test_write_path_design_arm.py

@@ -181,12 +181,13 @@ async def test_a_ui_write_with_no_prior_art_still_hears_the_design_system():
 
 
 @pytest.mark.asyncio
-async def test_a_design_only_write_does_not_switch_the_rule_arm_on():
-    """The rule arm runs only where there is prior art. A design line that
-    joined that gate would start a semantic rule search on every UI write —
-    a new population of calls under a floor tuned without them."""
-    _, rules = await _hint(_quiet())
-    rules.assert_not_called()
+async def test_a_write_with_only_a_design_line_still_gets_its_rule_check():
+    """Every write with a path gets the rule check, prior art or not. With
+    nothing ranked, the design line still arrives on its own."""
+    out, rules = await _hint(_quiet())
+    rules.assert_awaited()
+    assert rules.await_args_list[0].args[1] == "frontend/src/B.vue"
+    assert out["context"].startswith("> Design system binds `frontend/src/B.vue`")
 
 
 @pytest.mark.asyncio

tests/test_write_path_trigger.py

@@ -485,13 +485,13 @@ async def test_writepath_threshold_is_operator_tunable_and_clamped():
 
 
 @pytest.mark.asyncio
-async def test_the_rule_arm_has_its_own_tunable_bar():
+async def test_the_rule_arms_have_their_own_tunable_bar():
     """Settable and clamped again, for the THIRD corpus.
 
     Separate from the code threshold above and separately settable, because the
-    two are measured against different things: 0.68 was derived from code
-    against note PROSE, and rules are short imperative technical
-    English — a more homogeneous corpus whose noise floor sits higher.
+    two are measured against different things: the code bar against note
+    PROSE, the rule bar against short imperative technical English — a more
+    homogeneous corpus whose noise floor sits higher.
     """
     from grimoire.services import plugin_context as pc
 
@@ -524,33 +524,28 @@ async def test_the_two_write_path_bars_are_independent():
 
 
 @pytest.mark.asyncio
-async def test_the_two_act_arms_read_independent_rule_bars():
-    """The command arm's bar moves without the write path's.
+async def test_both_act_arms_read_the_one_rule_bar():
+    """One stored value moves the command arm and the write arm together.
 
-    The two act arms behave like different subsystems at the same number,
-    because a code payload is long and rich where a shell command is short. A config assembler that
-    reads one key into both fields would silently undo that, and the symptom
-    would be invisible: both arms would simply agree again.
+    They search the same rules, and what limits each is its budget and band.
+    A config assembler that read a second key into either field would split
+    the bar again with nothing in the settings to show it.
     """
     from grimoire.services import plugin_context as pc
 
-    stored = {pc.RULEHINT_THRESHOLD_KEY: "0.75", pc.TOOLRULE_THRESHOLD_KEY: "0.61"}
+    assert pc.TOOLRULE_THRESHOLD_KEY == pc.RULEHINT_THRESHOLD_KEY
+    stored = {pc.RULEHINT_THRESHOLD_KEY: "0.61"}
     with _stored_settings(stored):
         cfg = await pc.get_writepath_config(1)
 
-    assert cfg["rule_threshold"] == 0.75
+    assert cfg["rule_threshold"] == 0.61
     assert cfg["tool_rule_threshold"] == 0.61
 
 
 @pytest.mark.asyncio
-async def test_a_garbage_command_bar_falls_back_to_its_own_default():
-    """Not to 0.0, and not to the write path's default.
-
-    Falling back to 0.0 would attach a rule to every Bash call in the session;
-    falling back to the sibling's default would quietly re-merge the two bars
-    that are kept separate, which is the harder failure to see because the arm
-    keeps working.
-    """
+async def test_a_garbage_rule_bar_falls_back_to_its_default():
+    """Not to 0.0, which would attach a rule to every Bash call and every
+    write in the session."""
     from grimoire.services import plugin_context as pc
 
     stored = {pc.TOOLRULE_THRESHOLD_KEY: "banana"}
@@ -558,20 +553,7 @@ async def test_a_garbage_command_bar_falls_back_to_its_own_default():
         cfg = await pc.get_writepath_config(1)
 
     assert cfg["tool_rule_threshold"] == pc.TOOLRULE_DEFAULT_THRESHOLD
-
-
-def test_the_command_bar_defaults_below_the_write_path_bar():
-    """A DIRECTION check, like its sibling above, and for the same reason: the
-    value is measured against one corpus, the relationship is not.
-
-    A shell command carries less text than a code payload and therefore scores
-    lower for the same relevance; at the write path's bar, consequential
-    commands retrieve nothing. Tuning either value stays free; inverting the
-    relationship would mute the command arm.
-    """
-    from grimoire.services import plugin_context as pc
-
-    assert pc.TOOLRULE_DEFAULT_THRESHOLD < pc.RULEHINT_DEFAULT_THRESHOLD
+    assert cfg["rule_threshold"] == pc.RULEHINT_DEFAULT_THRESHOLD
 
 
 def test_the_rule_bar_defaults_above_the_code_bar():
@@ -632,7 +614,9 @@ async def test_trivial_payload_skips_the_semantic_arm_entirely():
         out = await pc.build_write_path_hint(1, "src/x.py", code="x = 1")
     assert out["context"] == "" and out["note_ids"] == []
     search.assert_not_called()
-    rec.assert_not_called()
+    # The rule arm still checks the write and logs its own row; the notes arm
+    # logs none.
+    assert [c for c in rec.call_args_list if c.kwargs["source"] == "write_path"] == []
 
 
 @pytest.mark.asyncio
@@ -816,7 +800,8 @@ async def test_telemetry_records_the_query_actually_sent():
          patch.object(pc, "semantic_search_notes", AsyncMock(return_value=[])), \
          patch.object(pc, "record_retrieval", rec):
         await pc.build_write_path_hint(1, "src/x.py", code=REAL_CODE)
-    logged = rec.call_args.kwargs["query"]
+    row = next(c for c in rec.call_args_list if c.kwargs["source"] == "write_path")
+    logged = row.kwargs["query"]
     assert logged.startswith("debounce(fn, wait=0.25)")
     assert "nonlocal timer" not in logged