2026-10-22 20:26:40 -04:00
38 changed files with 1243 additions and 387 deletions
@@ -44,6 +44,14 @@ export interface RemovedDefault {
  created_at: string | null;
 }
 export interface MomentRuleUsage {
  rule_id: number;
  /** Null for a rule deleted since it was delivered. */
  title: string | null;
  delivered: number;
  opened: boolean;
 }
 export interface MomentUsage {
  /** How many times a mounted rule arrived at this moment. */
  delivered: number;
@@ -53,6 +61,8 @@ export interface MomentUsage {
   *  the door, not the line that prompted it. */
  opened: number;
  last_delivered_at: string | null;
  /** Each rule delivered here, most delivered first. */
  by_rule: MomentRuleUsage[];
 }
 export interface MomentsPayload {
@@ -0,0 +1,37 @@
 <script setup lang="ts">
 import type { MomentRuleUsage } from "@/api/moments";
 /**
 * Which rules a moment delivered, and whether an agent went on to read each.
 * The moment's own counts cannot say this once two rules are mounted there.
 * Closed by default: the counts in the row above are what most visits want.
 */
 defineProps<{ rules: MomentRuleUsage[] }>();
 </script>
 <template>
  <details v-if="rules.length" class="moment-rules">
    <summary>Rules delivered here</summary>
    <ul class="delivered-list">
      <li v-for="r in rules" :key="r.rule_id" class="delivered-row">
        <span class="delivered-title">{{ r.title ?? `Rule ${r.rule_id} (deleted)` }}</span>
        <span class="delivered-count">delivered {{ r.delivered }}</span>
        <span :class="r.opened ? 'delivered-opened' : 'delivered-unopened'">
          {{ r.opened ? "opened" : "not opened" }}
        </span>
      </li>
    </ul>
  </details>
 </template>
 <style scoped>
 .moment-rules { margin-top: var(--fs-space-2); font-size: var(--fs-size-tiny); }
 .moment-rules summary { cursor: pointer; color: var(--fs-text-tertiary); }
 .delivered-list { list-style: none; margin: var(--fs-space-1) 0 0; padding: 0; display: flex; flex-direction: column; gap: var(--fs-space-1); }
 .delivered-row { display: flex; flex-wrap: wrap; align-items: baseline; gap: var(--fs-space-2); font-variant-numeric: tabular-nums; }
 .delivered-title { color: var(--fs-text-primary); font-size: var(--fs-size-body-sm); }
 .delivered-count { color: var(--fs-text-secondary); }
 .delivered-opened { color: var(--fs-text-secondary); }
 /* Not read is a fact about this window, not a fault: quiet, like "nothing mounted". */
 .delivered-unopened { color: var(--fs-text-tertiary); font-style: italic; }
 </style>
@@ -2,6 +2,7 @@
 import { computed, onMounted, ref } from "vue";
 import { useMomentsStore } from "@/stores/moments";
 import MomentProposals from "@/components/MomentProposals.vue";
 import MomentRuleUsage from "@/components/MomentRuleUsage.vue";
 import type { MomentAction, MomentUsage } from "@/api/moments";
 import { fmtDate } from "@/utils/dateFormat";
@@ -167,6 +168,7 @@ onMounted(() => store.load(true));
              >×</button>
            </span>
          </div>
          <MomentRuleUsage v-if="usage(m.name)" :rules="usage(m.name)!.by_rule" />
        </li>
      </ul>
@@ -186,6 +188,7 @@ onMounted(() => store.load(true));
                </template>
              </span>
            </div>
            <MomentRuleUsage v-if="usage(name)" :rules="usage(name)!.by_rule" />
          </li>
        </ul>
      </template>
@@ -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
+  // Same `|| default` reasoning again, and it bites harder here: rule hints
-  // fires on every write, so a fallback of 0 would attach a standing rule to
+  // fire on every edit and before every command, so a fallback of 0 would put
-  // every edit in the session.
+  // a standing rule beside each of them.
-  const rhT = asBar(kbRuleHintThreshold.value, 0.72);
+  const ruleT = asBar(kbRuleThreshold.value, 0.7);
  // 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);
  // 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
+      // A third corpus with a third bar — see RULE_FLOOR_DEFAULT in
-      // in services/plugin_context.py for why rules cannot share the
+      // services/retrieval_pipeline.py for why every rule hint shares it.
-      // code threshold any more than code could share the prose one.
+      kb_rule_threshold: String(ruleT),
      kb_rulehint_threshold: String(rhT),
      // 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) {
+  if (allSettings.kb_rule_threshold !== undefined) {
-    kbRuleHintThreshold.value = allSettings.kb_rulehint_threshold;
+    kbRuleThreshold.value = allSettings.kb_rule_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_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"
+            id="kb-rule-threshold"
-            v-model="kbRuleHintThreshold"
+            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
+            The bar for a standing rule or preference to be offered at all,
-            code being written. <em>Every</em> rule is eligible — nothing is
+            whether against the code being written, a command about to run, or
-            preloaded any more, so this is the only way a rule reaches a write.
+            what you just said. <em>Every</em> rule is eligible — nothing is
-            Stricter than the threshold above, because there are far fewer rules
+            preloaded, so this is how a rule reaches the work. One bar serves all
-            than snippets: with a small set something always ranks first, so the
+            three: they search the same rules, and how many each may show is set
-            bar has to carry more of the judgement. If rules arrive unread, or one
+            by its own budget below, which is usually the better lever. If rules
-            you needed never showed up, that is Claude's to notice and correct
+            arrive unread, or one you needed never showed up, that is Claude's to
-            from the telemetry — and the budget below is usually the better lever.
+            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
@@ -42,7 +42,11 @@ async def menus_to_review(
    Args:
        source: the arm to review. Only arms whose search can be re-run
-            exactly are offered; today that is `auto_inject`.
+            exactly are offered: `auto_inject`, and the rule arms
            `prompt_rule`, `pre_tool_rule` and `write_path_rule`. A rule
            line's passage is the trigger the arm matched, and for the
            command and write arms `within_budget` also needs the line to
            sit inside the band of the top hit, as it did when shown.
        n: how many calls to sample, 1–20. Random among the calls you have not
            judged, so a review does not describe one afternoon.
        days: how far back to sample from.
@@ -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])
+    mounted = await rulebooks_svc.list_rule_moments(ids)
-    return {
+    rules = [_rule_summary(r, mounted.get(r.id)) for r in rows]
-        "rules": [_rule_summary(r, mounted.get(r.id)) for r in rows],
+    if with_usage:
-        "total": len(rows),
+        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:
@@ -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
@@ -590,8 +599,11 @@ async def retrieval_telemetry(
      corpus. Not a verdict on them: a line carries its matched passage, so
      an unopened record may have been unrelated, enough as shown, or already
      in context. Judge a sample (`menus_to_review`) before acting on it.
-      For RULES there is no judged sample: a rule set aside again and again
+      For RULES, judge a sample of the rule arm that showed them
-      is a trigger to fix (`update_rule(when_to_apply=...)`), not a floor.
+      (`menus_to_review(source="prompt_rule")`, or `pre_tool_rule`,
      `write_path_rule`); a rule set aside again and again where it does
      not apply is a trigger to fix (`update_rule(when_to_apply=...)`), not
      a floor.
    - `read_and_unacted` — distinct rules OPENED in the window that recorded
      no outcome, against the ones that did. The worse sibling of `surfaced_never_pulled` above: a
      rule nobody opens is cheap, while a rule read and silently unchanged is
@@ -120,6 +120,12 @@ def _says(response: httpx.Response) -> str:
        "AccessDenied": "the key isn't allowed to do that",
    }
    said = hints.get(code) or message or response.reason_phrase
    if not code and 400 <= response.status_code < 500:
        # S3 refuses with an <Error><Code> body. A 4xx without one came from
        # something else at that address — most often a web server on the
        # default port, when the S3 service listens on a port of its own.
        said = ("the address answered, but not as S3 — check the scheme and "
                "the port (an S3 service usually listens on its own port)")
    return f"HTTP {response.status_code}{f' {code}' if code else ''}: {said}."
@@ -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"?}}.
@@ -143,8 +143,13 @@ WRITEPATH_ENABLED_KEY = "kb_writepath_enabled"
 WRITEPATH_THRESHOLD_KEY = SURFACES["write_path"].floor_key
 WRITEPATH_DEFAULT_ENABLED = True
 WRITEPATH_DEFAULT_THRESHOLD = SURFACES["write_path"].floor_default
 # How far past the write-path floor a snippet must resemble a NEW definition
 # before the hint asks whether the code is a copy of it. A margin, not a second
 # floor, so it moves when the operator tunes the write-path floor: at the
 # default 0.68 the copy bar is 0.80.
 WRITEPATH_COPY_MARGIN = 0.12
-# 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 +160,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
+#      arithmetic rather than relevance.
 #      makes the pool larger; the bar still errs toward silence on purpose.)
 #   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
+# ONE BAR FOR EVERY RULE ARM THAT ANNOTATES. The write path, the command arm
-# floor and rule-text's is not derivable in general — it depends on how an
+# and the prompt arm read the same key (`RULE_FLOOR_KEY` in retrieval_pipeline):
-# install writes its rules — so the default errs deliberately toward SILENCE
+# they search the same rules, and what limits each one is its budget and band.
 # 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.
 #
 # TUNE IT FROM YOUR OWN INSTANCE: `retrieval_telemetry` reports
-# `rule_usage.pull_through`. Raise this if rules arrive unread; lower it if
+# `rule_usage.pull_through` and `near_miss_samples`. Raise it if rules arrive
-# rules you needed never arrived. What would RETIRE it: a cross-encoder rerank,
+# unread; lower it if rules you needed never arrived. What would RETIRE it: a
-# which would make a similarity bar the wrong control entirely.
+# 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.
 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.
+# 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
-# A write-path query is a code payload, long and rich, while a pre-tool query
+# less here. The band does noise control downstream: a hit that clears the bar
-# is a shell command — often under a dozen words. Less text, less signal, lower
+# still has to score within `_RULEHINT_BAND` of the top to be rendered.
 # 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 innocuous scores near the line are largely CROSS-PROJECT BLEED rather
 # than bad ranking: another project's shell-command rule matching a shell
@@ -439,11 +407,13 @@ _CONCEPT_MIN_CHARS = 16
 _AUTOINJECT_MAX_TOP_K = MAX_BUDGET
 # THE CONVERSATION BESIDE THE PROMPT. The operator's message alone is a weak
-# query mid-session: it is mostly a follow-up — "yes do that", "now fix the
+# query mid-session: it is mostly a follow-up — "now fix the filter" — that
-# filter" — that names nothing a record could match. The hook sends the tail of the last assistant reply as `context`, and it is
+# names too little for a record to match. The hook sends the tail of the last
-# appended to a SHORT prompt only: a prompt that already says what it is about
+# assistant reply as `context`, and it is appended to a SHORT prompt only: a
-# is the better query on its own, and diluting it is the one way this can make
+# prompt that already says what it is about is the better query on its own,
-# retrieval worse.
+# and diluting it is the one way this can make retrieval worse. A prompt that
 # names nothing at all ("yes do that") gets no notes menu (`_has_subject`);
 # the prompt rule arm still searches it with the tail.
 #
 # Neither number costs a model token. The query is embedding input; what the
 # session pays for is the menu, and the budget bounds that unchanged.
@@ -455,6 +425,42 @@ _AUTOINJECT_CONTEXT_PROMPT_MAX = 280
 _AUTOINJECT_CONTEXT_MAX = 600
 # WORDS THAT NAME NO SUBJECT. A prompt made only of these ("keep going", "yes
 # please do", "1. yes 2. yes", "is it deployed?", "what is next?") is a turn of
 # the conversation, not a question about anything. Its query would be the
 # previous reply's tail alone, and the menu would hand back the records that
 # reply was built from, which the session already has in context. One word
 # outside this list ("now fix the filter") is a subject, and the prompt is
 # searched as usual.
 _NO_SUBJECT_WORDS = frozenset("""
    a about above after again agreed ahead all already also am an and any
    anything approved are as asked at awesome back be been before both build
    but by can carry ci commit continue continuing could cool correct
    deploy deployed described did do does done each else every everything
    fine finish first fix for from get go going good got great green had has
    have he here hold how i if im in instead into is it its just keep know
    last left let lets like looks loose lgtm make may maybe me mentioned merge
    milestone more move my next nice no not now of off ok okay on one open or
    order other our out over pass passes passing perfect please previous
    proceed project proposed push ready really recommended redeploy rest right
    said same see she ship should so some sounds stack stacking start step
    steps still stop suggested sure task tasks thank thanks that the them then
    there these they thing things this those through to try up us waiting
    want was we well were what whats when where which while who why will with
    work would yeah yep yes yet you your
 """.split())
 def _has_subject(prompt: str) -> bool:
    """Does the operator's message name anything a record could be about?
    Numbers are left out: a record named by number is looked up by id, not
    matched, so it never needs the ranked menu to be found.
    """
    words = re.findall(r"[a-z]+", prompt.lower().replace("'", ""))
    return any(w not in _NO_SUBJECT_WORDS for w in words)
 def _autoinject_query(prompt: str, context: str) -> str:
    """The prompt, with recent conversation appended when the prompt is thin.
@@ -478,17 +484,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
+# THE SAME BAR AS THE ACT ARMS. An operator's prose is a different query shape
-# code and shell commands. An operator's prose is a different query shape
+# against the same documents, and triggers are written in the vocabulary of
-# against the same documents, and nothing yet says the two distributions line
+# the moment, which for most rules is act vocabulary. Every call is logged
-# up — triggers are written in the vocabulary of the MOMENT, which for most
+# under `prompt_rule`, so read `near_miss_samples` before moving the shared
-# rules is act vocabulary, so prose may well score lower across the board.
+# bar, not the percentile alone.
 #
 # 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.
 PROMPTRULE_DEFAULT_THRESHOLD = SURFACES["prompt_rule"].floor_default
 # ANCHORED ON THIS HOOK'S BUDGET rather than the act arms'. This arm fires
@@ -655,10 +655,6 @@ async def build_autoinject_hint(
    # left to a ranking that can only match the words around them.
    named = await _named_records(user_id, record_refs_svc.named_record_ids(q))
    named_ids = [int(n.id) for n in named]
    # Everything below searches, logs and fills its slots on the ENRICHED
    # query, so the telemetry row records what was actually asked.
    q = _autoinject_query(q, context)
    # THE LEDGER IS A RENDERING FACT, NOT A RETRIEVAL ONE. Excluding seen ids
    # from the search would make the second time a note is the best answer
    # SILENCE, indistinguishable from "nothing matched", and would measure
@@ -672,15 +668,23 @@ async def build_autoinject_hint(
    # lesson slots in that order, and the surfacing rows are written there.
    # What is decided HERE is what makes this moment this moment — the query
    # the operator's words became, and the records they named by number.
-    result = await rp.run_note_arm(
+    if not _has_subject(q):
-        rp.AUTO_INJECT,
+        # Nothing to search for: the menu would answer the previous reply,
-        rp.NoteMoment(
+        # not the operator. Records named by number are still shown.
-            user_id=user_id, query=q, project_id=project_id,
+        kept, chunks = [], {}
-            seen=frozenset(already), named=frozenset(named_ids),
+    else:
-        ),
+        # Everything below searches, logs and fills its slots on the ENRICHED
-        floor=cfg["threshold"], budget=cfg["top_k"], io=_note_io(),
+        # query, so the telemetry row records what was actually asked.
-    )
+        q = _autoinject_query(q, context)
-    kept = result.menu
+        result = await rp.run_note_arm(
            rp.AUTO_INJECT,
            rp.NoteMoment(
                user_id=user_id, query=q, project_id=project_id,
                seen=frozenset(already), named=frozenset(named_ids),
            ),
            floor=cfg["threshold"], budget=cfg["top_k"], io=_note_io(),
        )
        kept, chunks = result.menu, result.chunks
    if not kept and not named:
        return empty
@@ -782,7 +786,7 @@ async def build_autoinject_hint(
        # a reason, and a reader cannot tell the two apart once indented alike.
        passage = "" if nid in already else _menu_passage(
            document_title(note.title, note.note_type, note.data, note.body),
-            (result.chunks.get(nid) or {}).get("text"), name,
+            (chunks.get(nid) or {}).get("text"), name,
        )
        who = _shared_by(note)
        lines.extend(menu_entry(
@@ -1171,9 +1175,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
+    embeddings sit on a much higher similarity floor than prose) and RULEHINT
-    third corpus again), TOOLRULE (a shell command is a different query shape
+    (a third corpus again, whose one bar both rule arms read).
    from a code payload and scores lower for the same relevance).
    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
@@ -1446,6 +1449,9 @@ async def build_write_path_hint(
                        note.title, note.note_type, note.data, note.body,
                    ),
                    "seen": int(note.id) in excluded,
                    # Read by the copy check, which asks for more
                    # resemblance than the menu does.
                    "score": float(score),
                    # Carried, not re-read off the rendered marker. The
                    # marker is prose assembled for a human and it already
                    # varies by kind, language and the `seen` flag — a
@@ -1500,16 +1506,35 @@ async def build_write_path_hint(
            derive = [d for d in found if d.get("key") not in skip]
        except Exception:
            logger.warning("write-time derive check failed", exc_info=True)
    # The COPY check: the derive check matches by name, so a renamed copy
    # of a recorded snippet passes it and shows only as "similar" prior art.
    # When this write introduces a definition and an unseen snippet on the
    # menu resembles it well past the menu's own floor, say that a copy is
    # the moment to extract. Only a recorded snippet can be seen this way —
    # the ledger keeps no bodies to compare an unrecorded sibling against.
    copies: list[dict] = []
    copy_bar = cfg["threshold"] + WRITEPATH_COPY_MARGIN
    strong = [
        it for _m, it in menu
        if it.get("kind") == "snippet" and not it.get("seen")
        and it.get("score", 0.0) >= copy_bar
    ]
    if strong and stamp_shapes and project_id:
        named = {d["symbol"] for d in derive} | {s["symbol"] for s in suggested}
        try:
            fresh = await shape_ledger_svc.new_shapes_at(project_id, path, stamp_shapes)
            copies = [
                {"symbol": n, "kind": k, "snippet": strong[0]}
                for k, n in fresh if n not in named
            ]
        except Exception:
            logger.warning("write-time copy 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
+    # rather than joining the guard's condition: a design line alone is not a
-    # a design line switch the standing-rule arm below on for writes where it
+    # reason to render the prior-art menu's frame.
    # has never run, moving that arm's call distribution under its floor.
    design_text, design_dedup = await _design_arm(
        user_id, project_id, path, set(exclude_derive or []),
    )
@@ -1520,7 +1545,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,
@@ -1632,6 +1712,8 @@ async def build_write_path_hint(
        lines.append(_divergence_line(path, divergence))
    if derive:
        lines.append(_derive_line(path, derive))
    if copies:
        lines.append(_copy_line(copies))
    # Split by arm, which is the whole reason this table exists. The place arm
    # carries no score and so has no home in retrieval_logs, yet it fires on
@@ -1656,52 +1738,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 ───────────────────────────────
+    lines.extend(rule_lines)
    #
    # 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)
    # 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
@@ -2004,6 +2041,22 @@ def _divergence_line(path: str, divergence: list[dict]) -> str:
    )
 def _copy_line(copies: list[dict]) -> str:
    """A new definition that closely resembles a recorded snippet under
    another name: the derive check cannot see a renamed copy, so this says
    it — as a question for the writer, since resemblance is not identity."""
    snip = copies[0]["snippet"]
    names = ", ".join(
        f"`{'.' if c['kind'] == 'css' else ''}{c['symbol']}`" for c in copies
    )
    return (
        f"> {names} {'is' if len(copies) == 1 else 'are'} new in this file and "
        f"this code resembles snippet #{snip['id']} `{snip.get('name') or snip['title']}` "
        f"(similar {snip['score']:.2f}). If it copies that shape, reuse #{snip['id']} "
        "or extract the shared shape now, before another copy is written."
    )
 def _suggest_line(path: str, suggested: list[dict]) -> str:
    """One line naming the recorded shape this code looks like, as evidence
    for the judgment the agent gives at the end of the turn — not a record of
@@ -460,16 +460,31 @@ class RuleArm:
    declared: Declared | None = None
-# The arms differ on purpose where these flags differ. Whether the prompt arm
+# ONE BAR FOR EVERY RULE ARM THAT ANNOTATES. The prompt, command and write
-# should band, and whether the act arms should reserve a preference, are open
+# arms search the same corpus, and what limits what they show is the budget
-# questions to settle on telemetry, not by symmetry.
+# 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_key=RULE_FLOOR_KEY,
-        floor_default=0.72,
+        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_key=RULE_FLOOR_KEY,
-        floor_default=0.68,
+        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_key=RULE_FLOOR_KEY,
-        floor_default=0.72,
+        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
@@ -46,13 +46,18 @@ from grimoire.models.base import iso
 from grimoire.models.note_usage import PULLED, NoteUsageEvent
 from grimoire.models.retrieval_judgment import VERDICTS, RetrievalJudgment
 from grimoire.models.retrieval_log import RetrievalLog
 from grimoire.models.rule_usage import PULLED as RULE_PULLED
 from grimoire.models.rule_usage import RuleUsageEvent
 logger = logging.getLogger(__name__)
 # The arms a review can re-run FAITHFULLY. A re-run with different parameters
 # from the arm's would judge a menu nobody was shown, so a surface joins this
-# only with its own search spelled out in `_rerun`.
+# only with its own search spelled out in `_rerun`. The rule arms named here
-REVIEWABLE = ("auto_inject",)
+# are the ones that annotate: each logs its query, floor, budget and project,
 # which is everything its search takes.
 RULE_REVIEWABLE = ("prompt_rule", "pre_tool_rule", "write_path_rule")
 REVIEWABLE = ("auto_inject", *RULE_REVIEWABLE)
 DEPTH_DEFAULT = 6
 DEPTH_MAX = 10
@@ -94,7 +99,58 @@ def _after(stamp, call) -> bool:
 async def _rerun(
    user_id: int, row: RetrievalLog, depth: int, report: dict | None = None,
 ) -> list[dict]:
-    """The logged call's query, searched again the way its arm searches.
+    """The logged call's query, searched again the way its arm searches."""
    if row.source in RULE_REVIEWABLE:
        return await _rerun_rules(user_id, row, depth, report)
    return await _rerun_notes(user_id, row, depth, report)
 async def _rerun_rules(
    user_id: int, row: RetrievalLog, depth: int, report: dict | None = None,
 ) -> list[dict]:
    """A rule arm's call, searched again: its query against rule triggers,
    with the floor and project it logged.
    `within_budget` is the arm's own cut: a line inside the budget, and, for
    an arm that bands, inside the band of the top hit. The preference slot is
    left out — it logs its own source. Rules created after the call are
    dropped and named in `report["postdated"]`, as for notes.
    """
    from grimoire.services import retrieval_pipeline as rp
    from grimoire.services.embeddings import semantic_search_rules
    arm = next(a for a in rp.RULE_ARMS if a.source == row.source)
    hits = await semantic_search_rules(
        user_id, row.query or "",
        limit=depth + POSTDATED_SLACK,
        threshold=row.threshold if row.threshold is not None else 0.0,
        project_id=row.project_id or None,
    )
    postdated = [int(r.id) for _s, r in hits if _after(r.created_at, row.created_at)]
    if report is not None:
        report["postdated"] = postdated
    hits = [h for h in hits if int(h[1].id) not in postdated][:depth]
    budget = int(row.limit_n or 0)
    cut = hits[:budget]
    kept = {int(r.id) for _s, r in (rp._rule_band(cut) if arm.band else cut)}
    return [{
        "rank": rank,
        "record_id": int(rule.id),
        "score": round(float(score), 4),
        "within_budget": int(rule.id) in kept,
        "changed_since_call": _after(rule.updated_at, row.created_at),
        "kind": rule.kind or "rule",
        "name": rule.title,
        # The trigger is what the arm matched against and what its line
        # quotes, so it is the passage the reader saw.
        "passage": rule.when_to_apply or rule.statement or None,
    } for rank, (score, rule) in enumerate(hits, start=1)]
 async def _rerun_notes(
    user_id: int, row: RetrievalLog, depth: int, report: dict | None = None,
 ) -> list[dict]:
    """A notes arm's call, searched again.
    `auto_inject`'s parameters, read from its spec (`retrieval_pipeline.
    AUTO_INJECT`) rather than restated: which kinds, whose records, and the
@@ -151,19 +207,25 @@ async def _opened_after(user_id: int, row: RetrievalLog, ids: list[int]) -> set[
    Agent pulls only — the `mcp_` prefix, per NoteUsageEvent's convention —
    because the question is whether the line moved the session, not whether
-    the operator later browsed to the record.
+    the operator later browsed to the record. A rule arm's lines are read
    from the rule table: a note and a rule can share an id.
    """
    if not ids or row.created_at is None:
        return set()
    table, id_col, pulled = (
        (RuleUsageEvent, RuleUsageEvent.rule_id, RULE_PULLED)
        if row.source in RULE_REVIEWABLE
        else (NoteUsageEvent, NoteUsageEvent.note_id, PULLED)
    )
    async with async_session() as session:
        found = await session.execute(
-            select(NoteUsageEvent.note_id).distinct().where(
+            select(id_col).distinct().where(
-                NoteUsageEvent.user_id == user_id,
+                table.user_id == user_id,
-                NoteUsageEvent.event == PULLED,
+                table.event == pulled,
-                NoteUsageEvent.source.startswith("mcp_", autoescape=True),
+                table.source.startswith("mcp_", autoescape=True),
-                NoteUsageEvent.note_id.in_(ids),
+                id_col.in_(ids),
-                NoteUsageEvent.created_at >= row.created_at,
+                table.created_at >= row.created_at,
-                NoteUsageEvent.created_at <= row.created_at + OPENED_WINDOW,
+                table.created_at <= row.created_at + OPENED_WINDOW,
            )
        )
        return {int(i) for (i,) in found.all()}
@@ -175,7 +237,7 @@ async def _logged_after(row: RetrievalLog, ids: list[int]) -> set[int]:
    (a closed task takes no claim stamp), so it is read here."""
    from grimoire.models.task_log import TaskLog
-    if not ids or row.created_at is None:
+    if not ids or row.created_at is None or row.source in RULE_REVIEWABLE:
        return set()
    async with async_session() as session:
        found = await session.execute(
@@ -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)
@@ -41,6 +41,7 @@ from grimoire.models.system_usage import SURFACED as SYSTEM_SURFACED
 from grimoire.models.system_usage import SystemUsageEvent
 from grimoire.services.rule_usage import is_ambient
 from grimoire.models.retrieval_log import RetrievalLog
 from grimoire.models.rulebook import Rule
 from grimoire.services.retrieval_review import judged_block
 from grimoire.services.retrieval_registry import (
    POINTS, UNBIDDEN, get_point, is_registered, sources_expected_to_emit,
@@ -489,9 +490,10 @@ def _num(raw: str, fallback):
 # to look next. ONE SENTENCE PER CORPUS because the two lines are built
 # differently: a note line carries its matched passage, so an unopened
 # note may have done its job, and the judged sample is the instrument that can
-# tell. A rule line carries only its trigger and asks to be opened, and no
+# tell. A rule line carries only its trigger and asks to be opened, so a rule
-# judged sample covers the rule arms (`retrieval_review.REVIEWABLE`), so
+# set aside again and again is usually a trigger to fix; the judged sample
-# sending that reader to `menus_to_review` would send them to a refusal.
+# (`retrieval_review.REVIEWABLE` names the rule arms it covers) says whether
 # it was set aside rightly.
 _NEVER_PULLED_READING = {
    "notes": (
        "That is not a verdict on them: a line carries its matched passage, "
@@ -504,11 +506,13 @@ _NEVER_PULLED_READING = {
        "Not a verdict either, and read differently from notes: the count "
        "includes rules that only arrived in a listing (the project "
        "handshake, a planning read), and a rule line shows its trigger, so a "
-        "session can rightly set one aside on the trigger alone. There is no "
+        "session can rightly set one aside on the trigger alone. To tell a "
-        "judged sample for the rule arms. A rule set aside again and again "
+        "rule set aside rightly from one missed, judge a sample of a rule "
-        "where it does not apply is a trigger to fix, not a floor to move — "
+        "arm with `menus_to_review(source=\"prompt_rule\")` (or "
-        "`update_rule(when_to_apply=...)`, then `what_might_apply` with the "
+        "`pre_tool_rule`, `write_path_rule`). A rule set aside again and "
-        "moment's own words to see where it ranks."
+        "again where it does not apply is a trigger to fix, not a floor to "
        "move — `update_rule(when_to_apply=...)`, then `what_might_apply` "
        "with the moment's own words to see where it ranks."
    ),
 }
@@ -803,6 +807,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}
@@ -933,7 +940,9 @@ async def moment_usage(user_id: int | None, since) -> dict:
    - `rules`: how many distinct rules did;
    - `opened`: how many of those an AGENT then opened (`mcp_*`), at or after
      their first delivery there in the window;
-    - `last_delivered_at`.
+    - `last_delivered_at`;
    - `by_rule`: each rule delivered there, most delivered first, with its
      `title`, `delivered` and whether it was `opened`.
    `opened` is an UPPER BOUND, for by_source's reason: a pull records the
    door, not the line that prompted it, so a rule delivered at two moments
@@ -980,15 +989,21 @@ async def moment_usage(user_id: int | None, since) -> dict:
                pull.source.startswith("mcp_", autoescape=True),
                pull.created_at >= per_rule.c.first_at,
            )
            # One row per (moment, rule); the moment's totals are summed from
            # these below, so `by_rule` and the totals cannot disagree.
            rows = (
                await session.execute(
                    select(
                        per_rule.c.moment,
-                        func.sum(per_rule.c.n).label("delivered"),
+                        per_rule.c.rule_id,
-                        func.count().label("rules"),
+                        Rule.title,
-                        func.count(case((opened, 1))).label("opened"),
+                        per_rule.c.n,
-                        func.max(per_rule.c.last_at).label("last_at"),
+                        opened.label("opened"),
-                    ).group_by(per_rule.c.moment)
+                        per_rule.c.last_at,
                    )
                    .select_from(per_rule)
                    .outerjoin(Rule, Rule.id == per_rule.c.rule_id)
                    .order_by(per_rule.c.moment, per_rule.c.n.desc(), per_rule.c.rule_id)
                )
            ).all()
            complete = await _complete_from(session, RuleUsageEvent, user_id)
@@ -997,15 +1012,26 @@ async def moment_usage(user_id: int | None, since) -> dict:
        block["moment_usage_failed"] = True
        return block
-    block["by_moment"] = {
+    by_moment: dict[str, dict] = {}
-        moment: {
+    for moment, rule_id, title, n, was_opened, last_at in rows:
-            "delivered": int(delivered or 0),
+        slot = by_moment.setdefault(moment, {
-            "rules": int(rules or 0),
+            "delivered": 0, "rules": 0, "opened": 0,
-            "opened": int(n_opened or 0),
+            "last_delivered_at": None, "by_rule": [],
-            "last_delivered_at": iso(last_at),
+        })
-        }
+        slot["delivered"] += int(n or 0)
-        for moment, delivered, rules, n_opened, last_at in rows
+        slot["rules"] += 1
-    }
+        slot["opened"] += 1 if was_opened else 0
        stamp = iso(last_at)
        if stamp and (slot["last_delivered_at"] or "") < stamp:
            slot["last_delivered_at"] = stamp
        # Which rule a shared moment delivered, and whether that rule was
        # opened: the per-moment counts cannot say once two rules are
        # mounted there. `title` is null for a rule deleted since.
        slot["by_rule"].append({
            "rule_id": int(rule_id), "title": title,
            "delivered": int(n or 0), "opened": bool(was_opened),
        })
    block["by_moment"] = by_moment
    block.update(_coverage(complete.get("*"), since))
    return block
@@ -1109,7 +1135,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 +1452,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 +1512,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 +1524,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 +1543,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 +1616,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 +1640,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 +1678,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)
@@ -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(
+        for name in moved:
-            user_id=user_id, surface=surface, dial=dial,
+            session.add(RetrievalTuningEvent(
-            old_value=old, new_value=applied, actor=actor, reason=text,
+                user_id=user_id, surface=name, dial=dial,
-            embedding_model=stamp["embedding_model"],
+                old_value=old, new_value=applied, actor=actor, reason=text,
-            shape_version=stamp["shape_version"],
+                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],
    }
@@ -43,7 +43,7 @@ from __future__ import annotations
 import logging
 from datetime import datetime, timezone
-from sqlalchemy import exists, func, not_, select
+from sqlalchemy import exists, func, not_, or_, select
 from sqlalchemy.exc import IntegrityError
 from grimoire.models import async_session
@@ -274,6 +274,11 @@ async def pending(user_id: int, rule_id: int | None = None, limit: int = 200) ->
    Each rule carries what it is mounted on now, so a proposal is read beside
    the mounts it would add to.
    A signal pair is a proposal only once it has crossed the bar and asked
    (`proposed_at` in its evidence). Below the bar it is a row gathering
    evidence, and listing it would bury the few pairs the signal stands
    behind under every rule that was once opened after a moment.
    """
    from grimoire.services.rulebooks import _owned_rules_clause
@@ -282,7 +287,13 @@ async def pending(user_id: int, rule_id: int | None = None, limit: int = 200) ->
        stmt = (
            select(RuleMomentJudgment, Rule)
            .join(Rule, Rule.id == RuleMomentJudgment.rule_id)
-            .where(RuleMomentJudgment.state == SUGGESTED, _owned_rules_clause(user_id))
+            .where(
                RuleMomentJudgment.state == SUGGESTED, _owned_rules_clause(user_id),
                or_(
                    RuleMomentJudgment.source != "signal",
                    RuleMomentJudgment.evidence["proposed_at"].astext.isnot(None),
                ),
            )
        )
        if rule_id:
            stmt = stmt.where(Rule.id == int(rule_id))
@@ -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,
@@ -2706,6 +2706,29 @@ async def write_time_derive(
    return out
 async def new_shapes_at(
    project_id: int, path: str, shapes: list[tuple[str, str]]
 ) -> list[tuple[str, str]]:
    """The (kind, name) shapes the hook named at ``path`` that the ledger has
    no live row for there — definitions this write introduces, as opposed to
    ones it edits inside. Order follows ``shapes``."""
    wanted = [(k, n) for k, n in shapes if n]
    if not wanted:
        return []
    async with async_session() as session:
        rows = (
            await session.execute(
                select(CodeShape.kind, CodeShape.symbol).where(
                    CodeShape.project_id == project_id,
                    CodeShape.path == path,
                    CodeShape.vanished_at.is_(None),
                )
            )
        ).all()
    here = {(k, _norm_symbol(s)) for k, s in rows}
    return [(k, n) for k, n in wanted if (k, _norm_symbol(n)) not in here]
 async def flag_divergence(project_id: int, *, since: datetime | None) -> int:
    """Flag shapes created after ``since`` (the previous refresh) that sit
    where a canon dominates and were not proposed as that canon. With no
@@ -20,6 +20,7 @@ from grimoire.services.plugin_context import (
    _AUTOINJECT_CONTEXT_MAX,
    _AUTOINJECT_CONTEXT_PROMPT_MAX,
    _autoinject_query,
    _has_subject,
    _menu_name,
    _menu_passage,
 )
@@ -56,6 +57,50 @@ def test_context_is_cut_from_the_end_of_the_reply():
    assert tail.endswith("TAIL") and "HEAD" not in tail
 # --- a prompt that names nothing ----------------------------------------------
@pytest.mark.parametrize("prompt", [
    "keep going", "yes please do", "1. yes 2. yes 3. yes", "start step 6",
    "is it deployed?", "awesome, what's next?", "go ahead",
    "is there anything else open in this project?",
    "please commit and let me know when it passes CI",
 ])
 def test_a_turn_of_the_conversation_names_no_subject(prompt):
    assert not _has_subject(prompt)
@pytest.mark.parametrize("prompt", [
    "now fix the filter", "keep steward, it's our own app",
    "continue into DRY 14", "I've added a maintenance plan",
    "fix the Windows gate first, then step 2",
 ])
 def test_one_word_of_its_own_is_a_subject(prompt):
    assert _has_subject(prompt)
@pytest.mark.asyncio
 async def test_a_prompt_with_no_subject_gets_no_menu_and_no_search():
    """Its query would be the previous reply's tail alone, and the menu would
    hand back what that reply was built from."""
    from grimoire.services import plugin_context as pc
    search = AsyncMock(return_value=[(0.9, fake_note(id=11, title="the reply's own record", user_id=1))])
    retrieval = MagicMock()
    with patch.object(pc, "get_autoinject_config",
                      AsyncMock(return_value={"enabled": True, "threshold": 0.55, "top_k": 3})), \
         patch.object(pc, "semantic_search_notes", search), \
         patch.object(pc, "record_retrieval", retrieval), \
         patch.object(pc, "record_surfaced", MagicMock()):
        out = await pc.build_autoinject_hint(
            1, "yes please do", project_id=2,
            context="I'll move the library filters into the side column.",
        )
    assert out["context"] == "" and out["note_ids"] == []
    search.assert_not_called()
    retrieval.assert_not_called()
 # --- what the hook extracts --------------------------------------------------
 def _rec(role: str, blocks: list[dict], sidechain: bool = False) -> str:
@@ -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"]])
@@ -143,6 +143,8 @@ async def test_the_signal_counts_sessions_and_asks_once_the_bar_is_crossed(world
    for situation in ("s1", "s1", "s2"):
        assert await judgments_svc.co_occurred(uid, done.id, ["work.verify"],
                                               situation=situation) == ""
    # Below the bar the pair is gathering evidence, not waiting on anyone.
    assert (await judgments_svc.pending(uid, rule_id=done.id))["rules"] == []
    line = await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation="s3")
    assert f"#{done.id}" in line and "`work.verify`" in line and "3 distinct sessions" in line
    assert "judge_rule_moments" in line
@@ -238,9 +238,10 @@ async def test_the_delivery_lookup_answers_by_home(world):
 async def test_moment_usage_counts_deliveries_and_the_agent_opens_that_followed(world):
-    """The per-moment readout, on real SQL: an EXISTS inside a
+    """The per-moment readout, on real SQL: a correlated EXISTS selected as a
-    COUNT(CASE …) is a shape no mock can vouch for, and the block's own guard
+    column beside an outer join is a shape no mock can vouch for, and the
-    would turn a database refusal into a quiet `moment_usage_failed`."""
+    block's own guard would turn a database refusal into a quiet
    `moment_usage_failed`."""
    from datetime import datetime, timedelta, timezone
    from sqlalchemy import delete
@@ -289,3 +290,13 @@ async def test_moment_usage_counts_deliveries_and_the_agent_opens_that_followed(
    report = block["by_moment"]["reply.report"]
    assert (report["delivered"], report["rules"], report["opened"]) == (1, 1, 0)
    assert report["last_delivered_at"]
    # Two rules share work.finish: which one arrived, and which was opened.
    assert finish["by_rule"] == [
        {"rule_id": rule.id, "title": rule.title, "delivered": 2, "opened": True},
        {"rule_id": other.id, "title": "Close the task with the push",
         "delivered": 1, "opened": False},
    ]
    assert report["by_rule"] == [
        {"rule_id": rule.id, "title": rule.title, "delivered": 1, "opened": False},
    ]
@@ -665,6 +665,28 @@ async def test_write_time_derive_names_the_family_or_the_canon_for_a_name(seeded
    assert await write_time_derive(pid, "v/B.vue", [("css", "log-empty")]) == []
@pytest.mark.integration
 async def test_new_shapes_at_names_only_the_definitions_the_file_does_not_hold(seeded):
    """A shape the ledger holds at the path is an edit inside it; one it
    does not hold there — even under a name living elsewhere — is new."""
    from grimoire.services.shape_ledger import new_shapes_at
    pid = seeded["pid"]
    await sync_repo_shapes(pid, REPO, _defs(
        ("src/a.py", "sym", "load", "def load():", "def load():\n    return 1"),
        ("src/b.py", "sym", "save", "def save():", "def save():\n    return 2"),
        ("v/A.vue", "css", "log-empty", ".log-empty {", ".log-empty { color: red }"),
    ), seen_marker="m1")
    assert await new_shapes_at(pid, "src/a.py", [("sym", "load"), ("sym", "save")]) == [
        ("sym", "save"),
    ]
    # CSS compares without the leading dot, as the rest of the ledger does.
    assert await new_shapes_at(pid, "v/A.vue", [("css", ".log-empty")]) == []
    assert await new_shapes_at(pid, "src/new.py", [("sym", "load")]) == [("sym", "load")]
    assert await new_shapes_at(pid, "src/a.py", [("sym", "")]) == []
@pytest.mark.integration
 async def test_derive_new_names_the_copy_that_joined_a_family_since_the_stamp(seeded):
    """The first sync seeds one `slug`; a later sync adds an identical
@@ -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])
@@ -139,7 +139,7 @@ async def test_a_named_record_comes_first_and_is_not_repeated_as_a_match():
    other = fake_note(id=11, title="something that matched the words", user_id=1)
    hits = [(0.80, named), (0.78, other)]
    out, surfaced, retrieval = await _hint(
-        "go ahead with 4448", hits=hits, records={4448: named},
+        "go ahead with 4448 and the filing fix", hits=hits, records={4448: named},
    )
    ctx = out["context"]
@@ -156,6 +156,23 @@ async def test_a_named_record_comes_first_and_is_not_repeated_as_a_match():
    assert row["suppressed"] == 1
@pytest.mark.asyncio
 async def test_a_prompt_with_no_subject_still_shows_what_it_names():
    """"go ahead with 4448" names a record and nothing else: the record is
    looked up and shown, and no ranked menu is searched for."""
    rec = _task(4448, "the named one")
    other = fake_note(id=11, title="what the reply was about", user_id=1)
    out, surfaced, retrieval = await _hint(
        "go ahead with 4448", hits=[(0.8, other)], records={4448: rec},
        context="I'll start on the importer next.",
    )
    assert out["note_ids"] == [4448]
    assert "Possibly relevant" not in out["context"]
    assert "auto_inject" not in surfaced
    retrieval.assert_not_called()
@pytest.mark.asyncio
 async def test_a_named_record_already_shown_is_a_pointer():
    rec = _task(4448, "the named one", "a long body that is already in context")
@@ -81,7 +81,7 @@ async def test_the_re_run_searches_the_way_the_arm_does():
        rerun_kw.update(kw)
        return []
-    row = SimpleNamespace(query="q", threshold=0.5, limit_n=3, project_id=None,
+    row = SimpleNamespace(source="auto_inject", query="q", threshold=0.5, limit_n=3, project_id=None,
                          created_at=CALL)
    with patch("grimoire.services.embeddings.semantic_search_notes", side_effect=rerun_search):
        await review._rerun(1, row, 3)
@@ -99,7 +99,7 @@ BEFORE = CALL - timedelta(days=1)
@pytest.mark.asyncio
 async def test_the_re_run_ranks_from_one_and_marks_the_budget_cut():
-    row = SimpleNamespace(query="q", threshold=0.55, limit_n=2, project_id=None,
+    row = SimpleNamespace(source="auto_inject", query="q", threshold=0.55, limit_n=2, project_id=None,
                          created_at=CALL)
    hits = [
        (0.81, fake_note(id=11, title="first", created_at=BEFORE, updated_at=BEFORE)),
@@ -127,7 +127,7 @@ async def test_a_record_written_after_the_call_is_dropped_before_ranking():
    """The session that made a call goes on to write about it, and that record
    outranks everything in a re-run. The call could never have been offered
    it, so it must not take a rank — or a verdict — from what was."""
-    row = SimpleNamespace(query="q", threshold=0.55, limit_n=1, project_id=None,
+    row = SimpleNamespace(source="auto_inject", query="q", threshold=0.55, limit_n=1, project_id=None,
                          created_at=CALL)
    after = CALL + timedelta(hours=1)
    hits = [
@@ -147,6 +147,103 @@ async def test_a_record_written_after_the_call_is_dropped_before_ranking():
            for ln in lines] == [(1, 22, True, False), (2, 23, False, True)]
 # ── the rule arms ────────────────────────────────────────────────────────
 def _rule(rid, title, created=BEFORE, updated=BEFORE):
    return SimpleNamespace(
        id=rid, title=title, kind="rule", when_to_apply=f"when {title}",
        statement="s", created_at=created, updated_at=updated,
    )
 def test_the_reviewable_rule_arms_are_the_ones_that_annotate():
    """Every rule arm that shows lines can be reviewed; the reply backstop
    shows none, so there is no menu to judge."""
    from grimoire.services import retrieval_pipeline as rp
    assert set(review.RULE_REVIEWABLE) == {
        a.source for a in rp.RULE_ARMS if not a.stop_only
    }
@pytest.mark.asyncio
 async def test_the_rule_re_run_searches_the_way_the_arm_does():
    """The floor and the project the arm searched with, run and compared."""
    from grimoire.services import retrieval_pipeline as rp
    arm_kw: dict = {}
    async def arm_search(_uid, _q, **kw):
        if "kind" not in kw:
            arm_kw.update(kw)
        return []
    await rp.run_rule_arm(
        rp.PROMPT_RULE, rp.RuleMoment(user_id=1, query="q", project_id=7),
        floor=0.7, budget=3,
        io=rp.RuleIO(search=arm_search, record_retrieval=MagicMock(),
                     record_rule_surfaced=MagicMock()),
    )
    rerun_kw: dict = {}
    async def rerun_search(_uid, _q, **kw):
        rerun_kw.update(kw)
        return []
    row = SimpleNamespace(source="prompt_rule", query="q", threshold=0.7, limit_n=3,
                          project_id=7, created_at=CALL)
    with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=rerun_search):
        await review._rerun(1, row, 3)
    for key in ("threshold", "project_id"):
        assert rerun_kw[key] == arm_kw[key], key
    assert "kind" not in rerun_kw
@pytest.mark.asyncio
@pytest.mark.parametrize(("source", "within"), [
    # No band: the budget alone is the cut.
    ("prompt_rule", [True, True, True, False]),
    # Banded: the third sits outside the band of the top hit, so it was
    # not shown though the budget had room for it.
    ("pre_tool_rule", [True, True, False, False]),
 ])
 async def test_a_rule_line_is_inside_the_cut_only_where_the_arm_showed_it(source, within):
    row = SimpleNamespace(source=source, query="git push", threshold=0.7, limit_n=3,
                          project_id=None, created_at=CALL)
    hits = [(0.86, _rule(1, "one")), (0.84, _rule(2, "two")),
            (0.76, _rule(3, "three")), (0.74, _rule(4, "four"))]
    async def search(_uid, _q, **kw):
        assert kw["threshold"] == 0.7 and kw["project_id"] is None
        return hits
    with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=search):
        lines = await review._rerun(1, row, 4)
    assert [ln["within_budget"] for ln in lines] == within
    assert lines[0]["passage"] == "when one"
    assert lines[0]["kind"] == "rule"
@pytest.mark.asyncio
 async def test_a_rule_written_after_the_call_is_dropped_before_ranking():
    row = SimpleNamespace(source="write_path_rule", query="q", threshold=0.7, limit_n=2,
                          project_id=None, created_at=CALL)
    after = CALL + timedelta(hours=1)
    hits = [(0.9, _rule(31, "written after", created=after, updated=after)),
            (0.8, _rule(32, "held then"))]
    async def search(_uid, _q, **kw):
        return hits
    rep: dict = {}
    with patch("grimoire.services.embeddings.semantic_search_rules", side_effect=search):
        lines = await review._rerun(1, row, 2, report=rep)
    assert rep["postdated"] == [31]
    assert [(ln["rank"], ln["record_id"]) for ln in lines] == [(1, 32)]
 # ── against Postgres ─────────────────────────────────────────────────────
@@ -127,7 +127,7 @@ async def test_injected_menu_attributes_another_users_note():
         patch.object(plugin_context, "record_retrieval", MagicMock()), \
         patch.object(plugin_context, "owner_names_for",
                      AsyncMock(return_value={9: "alex"})):
-        out = await plugin_context.build_autoinject_hint(7, "anything")
+        out = await plugin_context.build_autoinject_hint(7, "debounce helper")
    lines = out["context"].splitlines()
    theirs_line = next(ln for ln in lines if "#2" in ln)
@@ -160,7 +160,7 @@ async def test_injected_menu_labels_the_record_kind():
                      AsyncMock(return_value={"enabled": True, "threshold": 0.5,
                                              "top_k": 5})), \
         patch.object(plugin_context, "record_retrieval", MagicMock()):
-        out = await plugin_context.build_autoinject_hint(7, "anything")
+        out = await plugin_context.build_autoinject_hint(7, "debounce helper")
    lines = out["context"].splitlines()
    by_id = {n: next(ln for ln in lines if f"#{n} " in ln) for n in (1, 2, 3, 4, 5)}
@@ -71,13 +71,29 @@ def test_every_surface_name_is_a_real_telemetry_source():
    )
-def test_no_two_surfaces_share_a_settings_key():
+def test_a_shared_settings_key_is_one_bar_with_one_default():
-    """One dial, one label."""
+    """One dial, one label. Surfaces may share a floor key, as the rule arms
-    floors = [s.floor_key for s in rs.SURFACES.values()]
+    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():
@@ -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),
@@ -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", [
@@ -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")
@@ -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
+# A prior-art hit for the fixtures that want the menu beside the rule lines.
-# staleness, no synced record, no prior-art menu, no shape signal. The rule arm
+# The rule arm runs on every write with a path, whether or not this matched.
 # 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.
 _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():
+async def test_a_write_that_matched_no_prior_art_still_gets_its_rule_check():
-    """The gate, pinned — because the fixture above now depends on it and a
+    """A rule about the file being written does not depend on a snippet
-    silent change would make every other test here pass vacuously.
+    resembling it, so the arm runs before the prior-art guard, not behind it."""
    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.
    """
    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
+    # The arm's own row; the preference slot logs beside it under its own name.
-    assert log.call_args.kwargs["source"] == "pre_tool_rule"
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == "pre_tool_rule"]
-    assert log.call_args.kwargs["results"] == []
+    assert len(rows) == 1
-    assert log.call_args.kwargs["query"] == "curl -s https://git.example/api/v1/runs", (
+    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
+    rows = [c for c in log.call_args_list if c.kwargs.get("source") == "pre_tool_rule"]
-    assert log.call_args.kwargs["results"] == []
+    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 _tool_row(log)["results"] == []
-    assert log.call_args.kwargs["suppressed"] == 1
+    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 _tool_row(log)["suppressed"] == 1
-    assert len(log.call_args.kwargs["results"]) == 1
+    assert len(_tool_row(log)["results"]) == 1
 # ── The identity between the two tables ─────────────────────────────────
@@ -78,6 +78,22 @@ def test_a_refusal_says_why(tmp_path):
        asyncio.run(s3.put_file(TARGET, "b.tar", f, transport=transport))
 def test_a_web_server_at_the_address_is_named_as_not_s3():
    """No port in the address: the request lands on whatever serves port 80,
    and its plain 404 says nothing about S3."""
    page = b"<html><body><h1>404 Not Found</h1></body></html>"
    transport = httpx.MockTransport(lambda r: httpx.Response(404, content=page))
    with pytest.raises(s3.S3Error, match=r"HTTP 404: the address answered, but not as S3 — check the scheme and the port"):
        asyncio.run(s3.list_objects(TARGET, transport=transport))
 def test_a_server_error_without_a_body_is_not_blamed_on_the_port():
    transport = httpx.MockTransport(lambda r: httpx.Response(503))
    with pytest.raises(s3.S3Error, match="HTTP 503: Service Unavailable") as err:
        asyncio.run(s3.list_objects(TARGET, transport=transport))
    assert "port" not in str(err.value)
 def test_unreachable_says_so(tmp_path):
    f = tmp_path / "b.tar"
    f.write_bytes(b"x")
@@ -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):
@@ -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(
@@ -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"),
+    # The three rule hints read one bar, so one control serves them.
-    ("pre_tool_rule", "kbToolRuleThreshold"),
+    ("write_path_rule", "kbRuleThreshold"),
-    ("prompt_rule", "kbPromptRuleThreshold"),
+    ("pre_tool_rule", "kbRuleThreshold"),
    ("prompt_rule", "kbRuleThreshold"),
    ("reply_rule", "kbReplyRuleThreshold"),
 )
@@ -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():
+async def test_a_write_with_only_a_design_line_still_gets_its_rule_check():
-    """The rule arm runs only where there is prior art. A design line that
+    """Every write with a path gets the rule check, prior art or not. With
-    joined that gate would start a semantic rule search on every UI write —
+    nothing ranked, the design line still arrives on its own."""
-    a new population of calls under a floor tuned without them."""
+    out, rules = await _hint(_quiet())
-    _, rules = await _hint(_quiet())
+    rules.assert_awaited()
-    rules.assert_not_called()
+    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
@@ -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
+    two are measured against different things: the code bar against note
-    against note PROSE, and rules are short imperative technical
+    PROSE, the rule bar against short imperative technical English — a more
-    English — a more homogeneous corpus whose noise floor sits higher.
+    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():
+async def test_both_act_arms_read_the_one_rule_bar():
-    """The command arm's bar moves without the write path's.
+    """One stored value moves the command arm and the write arm together.
-    The two act arms behave like different subsystems at the same number,
+    They search the same rules, and what limits each is its budget and band.
-    because a code payload is long and rich where a shell command is short. A config assembler that
+    A config assembler that read a second key into either field would split
-    reads one key into both fields would silently undo that, and the symptom
+    the bar again with nothing in the settings to show it.
    would be invisible: both arms would simply agree again.
    """
    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():
+async def test_a_garbage_rule_bar_falls_back_to_its_default():
-    """Not to 0.0, and not to the write path's default.
+    """Not to 0.0, which would attach a rule to every Bash call and every
-
+    write in the session."""
    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.
    """
    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
-
+    assert cfg["rule_threshold"] == pc.RULEHINT_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
 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
@@ -1595,6 +1580,60 @@ async def test_the_write_time_derive_check_names_a_family_or_a_canon_in_band():
    assert out["context"] == "" and out["derive"] == []
@pytest.mark.asyncio
 async def test_a_new_definition_close_to_a_snippet_is_asked_whether_it_is_a_copy():
    """The derive check matches by name, so a renamed copy of a recorded
    snippet passes it. A new definition whose code resembles an unseen
    snippet past the copy bar gets the extraction prompt; a shape the file
    already holds, a weaker hit, a seen hit, or a name the derive check
    already named does not."""
    from grimoire.services import plugin_context as pc
    note = fake_note(id=31, title="debounce — rate-limit a callback", user_id=1,
                     note_type="snippet")
    async def run(score, fresh, *, derive=(), exclude=None):
        new_at = AsyncMock(return_value=fresh)
        with patch.multiple(
            pc, get_writepath_config=AsyncMock(return_value=_cfg()),
            semantic_search_notes=AsyncMock(return_value=[(score, note)]),
            record_retrieval=MagicMock(), owner_names_for=AsyncMock(return_value={}),
        ), patch.object(pc.snippets_svc, "list_snippets", AsyncMock(return_value=([], 0))), \
             patch.object(pc.shape_ledger_svc, "recent_pulls", AsyncMock(return_value={})), \
             patch.object(pc.shape_ledger_svc, "write_time_divergence", AsyncMock(return_value=[])), \
             patch.object(pc.shape_ledger_svc, "write_time_derive",
                          AsyncMock(return_value=list(derive))), \
             patch.object(pc.shape_ledger_svc, "new_shapes_at", new_at):
            out = await pc.build_write_path_hint(
                1, "src/x.py", code=REAL_CODE, project_id=24,
                stamp_shapes=[("sym", "throttle_calls")], exclude_ids=exclude,
            )
        return out["context"], new_at
    bar = pc.WRITEPATH_DEFAULT_THRESHOLD + pc.WRITEPATH_COPY_MARGIN
    ctx, new_at = await run(bar + 0.04, [("sym", "throttle_calls")])
    new_at.assert_awaited_once_with(24, "src/x.py", [("sym", "throttle_calls")])
    assert "> `throttle_calls` is new in this file and this code resembles snippet #31 " \
           "`debounce`" in ctx
    assert f"(similar {bar + 0.04:.2f}). If it copies that shape, reuse #31" in ctx
    # The file already holds the shape: an edit inside it, not a copy.
    ctx, _ = await run(bar + 0.04, [])
    assert "is new in this file" not in ctx
    # On the menu, but under the copy bar: no ledger read at all.
    ctx, new_at = await run(bar - 0.04, [("sym", "throttle_calls")])
    assert "is new in this file" not in ctx and "similar" in ctx
    new_at.assert_not_awaited()
    # Shown already this session: the line was said with its first showing.
    ctx, _ = await run(bar + 0.04, [("sym", "throttle_calls")], exclude=[31])
    assert "is new in this file" not in ctx
    # The derive check named the shape: its line is the stronger claim.
    family = {"symbol": "throttle_calls", "kind": "sym", "key": "dup:1",
              "family": {"group": "dup:1", "label": "throttle_calls", "identical": True,
                         "files": ["a/y.py"], "file_count": 1, "size": 2}}
    ctx, _ = await run(bar + 0.04, [("sym", "throttle_calls")], derive=[family])
    assert "is new in this file" not in ctx
 def test_the_hook_keeps_a_derive_channel_and_sends_it_back(tmp_path):
    """Derive keys the server returns land in the session's own
    `.derive.ids` file and go back as `exclude_derive` on the next write —