feat(4636): rule pull-through per surface, and per-rule usage with confirmed lessons on list_rules

retrieval_telemetry gains rule_usage.by_source, the twin of usage.by_source:
distinct rules each surface showed, how many an agent opened, and a ratio
that is null on a bulk delivery. list_rules(with_usage=True, days=N) adds
each rule's usage, windowed, and how many lessons are confirmed instances
of it, so linked and unlinked rules can be compared through MCP.

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

src/grimoire/services/lesson_rules.py

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

src/grimoire/services/retrieval_telemetry.py

@@ -1109,7 +1109,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 +1426,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
@@ -1596,6 +1643,28 @@ async def retrieval_summary(
         round(rule_usage["pulled_by_agent"] / rule_usage["surfaced"], 4)
         if rule_usage["surfaced"] else None
     )
+    # Per surface, the note block's shape and its caveat: a pull records the
+    # door, not the surface that led to it, so a rule two arms surfaced and an
+    # agent opened once counts as pulled for both. An upper bound — a surface
+    # reading near zero here is not being flattered.
+    if rule_by_source_rows is None:
+        rule_usage["by_source"] = {}
+        rule_usage["by_source_failed"] = True
+    else:
+        rule_usage["by_source"] = {
+            source: {
+                "rules_surfaced": int(n_surfaced or 0),
+                "rules_pulled": int(n_pulled or 0),
+                # None on an ambient surface, as for notes: nothing chose
+                # those rules, so the ratio would be misread.
+                "pull_through": (
+                    None if is_ambient(source) or not n_surfaced
+                    else round(int(n_pulled or 0) / int(n_surfaced), 4)
+                ),
+                "ambient": is_ambient(source),
+            }
+            for source, n_surfaced, n_pulled in rule_by_source_rows
+        }
     rule_usage.update(_coverage((rule_complete or {}).get("*"), since))
     out["rule_usage"] = rule_usage
     out["system_usage"] = await _system_usage(user_id, since)

src/grimoire/services/rule_usage.py

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