fix(6251): the rules warning sends its reader to the rule-arm review

surfaced_never_pulled told a rules reader there was no judged sample for the
rule arms. There is one now; the guard tying the text to REVIEWABLE caught it.

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

src/grimoire/mcp/tools/search.py

@@ -599,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
-      is a trigger to fix (`update_rule(when_to_apply=...)`), not a floor.
+      For RULES, judge a sample of the rule arm that showed them
+      (`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

src/grimoire/services/retrieval_telemetry.py

@@ -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
-# judged sample covers the rule arms (`retrieval_review.REVIEWABLE`), so
-# sending that reader to `menus_to_review` would send them to a refusal.
+# tell. A rule line carries only its trigger and asks to be opened, so a rule
+# set aside again and again is usually a trigger to fix; the judged sample
+# (`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 "
-        "judged sample for the rule arms. A rule set aside again and 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."
+        "session can rightly set one aside on the trigger alone. To tell a "
+        "rule set aside rightly from one missed, judge a sample of a rule "
+        "arm with `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, 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."
     ),
 }
 
@@ -936,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
@@ -983,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"),
-                        func.count().label("rules"),
-                        func.count(case((opened, 1))).label("opened"),
-                        func.max(per_rule.c.last_at).label("last_at"),
-                    ).group_by(per_rule.c.moment)
+                        per_rule.c.rule_id,
+                        Rule.title,
+                        per_rule.c.n,
+                        opened.label("opened"),
+                        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)
@@ -1000,15 +1012,26 @@ async def moment_usage(user_id: int | None, since) -> dict:
         block["moment_usage_failed"] = True
         return block
 
-    block["by_moment"] = {
-        moment: {
-            "delivered": int(delivered or 0),
-            "rules": int(rules or 0),
-            "opened": int(n_opened or 0),
-            "last_delivered_at": iso(last_at),
-        }
-        for moment, delivered, rules, n_opened, last_at in rows
-    }
+    by_moment: dict[str, dict] = {}
+    for moment, rule_id, title, n, was_opened, last_at in rows:
+        slot = by_moment.setdefault(moment, {
+            "delivered": 0, "rules": 0, "opened": 0,
+            "last_delivered_at": None, "by_rule": [],
+        })
+        slot["delivered"] += int(n or 0)
+        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