feat(telemetry): rule_usage_events — the table, the service, and a restore that maps rule ids through the rule map (#3315)

Milestone 333 step 1. The write-path standing-rule arm is the only retrieval
surface in Scribe whose usefulness cannot be observed — and, not
coincidentally, the only one that has never declined to fire. 296 calls, zero
zero-result, 100% clearing its threshold, while every other surface declines
most of the time (#3311, and re-measured in note #3430). `retrieval_logs`
gives it scores; scores say what the ranker thought, never whether the hint
landed.

WHY A SIBLING TABLE AND NOT A COLUMN ON note_usage_events. The row carries no
note-specific field and the readout is the same shape, which is the strongest
case for sharing that note #3163 admits. What decides against it is identity at
RESTORE: the note importer maps note_id through note_id_map, so a rule id
parked in that column comes back attached to whatever note holds that number in
the target database. Not dropped — reattached. The restore reports success, the
counters are populated, and every one is about the wrong record, with no other
field to disagree with. rule_versions made the same call for the same reason;
this is the third rule-side sibling and it reads like the first two.

FK-free on rule_id and user_id, matching note_usage_events / retrieval_logs /
app_logs, and deliberately unlike rule_versions. A version belongs to a rule's
history and dies with it; telemetry outlives what it describes. Deleting a rule
must not erase the evidence that it was surfaced forty times and opened never,
because that evidence is the case for having deleted it.

The service uses `background.spawn` rather than a third copy of the
strong-reference dance — that module's own docstring says new callers should,
and a fourth copy is how one of them drifts. The AppLog canary #2663 demands is
kept, and since `rule_usage` needed exactly `note_usage`'s semantics, that
canary moved into `background.report_telemetry_failure` and note_usage now
calls it. `retrieval_telemetry` deliberately keeps its own: its canary is a
different shape (one process-wide flag, no AppLog row), so repointing it would
change behaviour rather than consolidate it.

No ambient bucket, and that is a decision. The note twin splits ranked from
ambient surfacings because enter_project and the skill sync deliver records
without choosing them (#2477). Rules have the same problem waiting —
list_always_on_rules loads them wholesale — but nothing emits here yet, so an
empty AMBIENT_SOURCES would be machinery pretending to a distinction the data
does not contain. `source` stays granular, so the split stays a readout-level
change needing no migration.

Backup carries it (v14). The round-trip test seeds a NOTE alongside the rule so
the target database has a note id to collide with — without that decoy, a
restore running rule ids through the wrong map would merely drop them and the
test would pass by absence, rather than failing on the populated-and-wrong
result that is the actual hazard.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01TcCs1CcQ1ormdnzSshKqvN

src/scribe/models/__init__.py

@@ -28,6 +28,7 @@ from scribe.models.invitation import InvitationToken  # noqa: E402, F401
 from scribe.models.embedding import NoteEmbedding, RuleEmbedding  # noqa: E402, F401
 from scribe.models.retrieval_log import RetrievalLog  # noqa: E402, F401
 from scribe.models.note_usage import NoteUsageEvent  # noqa: E402, F401
+from scribe.models.rule_usage import RuleUsageEvent  # noqa: E402, F401
 from scribe.models.project import Project  # noqa: E402, F401
 from scribe.models.milestone import Milestone  # noqa: E402, F401
 from scribe.models.task_log import TaskLog  # noqa: E402, F401

src/scribe/models/rule_usage.py

@@ -0,0 +1,95 @@
+from sqlalchemy import BigInteger, Index, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+from scribe.models.base import CreatedAtMixin, iso
+
+SURFACED = "surfaced"
+PULLED = "pulled"
+
+
+class RuleUsageEvent(Base, CreatedAtMixin):
+    """One row per time a rule was SURFACED to the agent, or PULLED in full.
+
+    The sibling `note_usage_events` has had since 2026-07, third in the line
+    after `rule_embeddings` and `rule_versions` — and, like those, it exists
+    because the rule side kept inheriting machinery built for notes and
+    quietly getting the weaker version of it.
+
+    WHY RULES NEED THEIR OWN AND CANNOT SHARE THE NOTE TABLE. Not squeamishness
+    about a polymorphic column — the row shares no note-specific fields and the
+    aggregate readout is the same shape, which is the strongest case for
+    sharing that note #3163 admits. What decides it is IDENTITY AT RESTORE. A
+    note id and a rule id are different namespaces resolved through different
+    maps, and `note_usage_events`'s importer maps `note_id` through
+    `note_id_map` and drops what does not resolve. A rule id parked in that
+    column would come back from a backup silently reattached to whatever note
+    happened to take that number — telemetry that is not merely lost but wrong,
+    and wrong in a way nothing downstream could detect.
+
+    WHAT THIS MEASURES, AND WHY IT DID NOT EXIST. The write-path standing-rule
+    arm is the only retrieval surface in Scribe whose usefulness cannot be
+    observed — and, not coincidentally, the only one that has never declined to
+    fire (#3311: 296 calls, zero zero-result, 100% clearing its threshold).
+    `retrieval_logs` gives it scores; scores say what the ranker thought, never
+    whether the hint landed. Without a pull counter no install can tune the arm
+    from evidence, only from the shape of a histogram.
+
+    Deliberately FK-FREE on `rule_id` and `user_id`, matching `note_usage_events`,
+    `retrieval_logs` and `app_logs` — and diverging from `rule_versions`, which
+    does carry FKs. The difference is what the row is FOR: a version is part of
+    a rule's history and dies with it, while telemetry outlives the row it
+    describes. Deleting a rule must not erase the evidence that it was surfaced
+    forty times and opened never, because that evidence is precisely the case
+    for having deleted it.
+
+    Cells left deliberately empty (note #3163's step 3): no share ACL — rules
+    have none of their own; no soft delete — nothing restores a telemetry row,
+    and the table is append-only; no embedding — an event is not a document.
+    """
+
+    __tablename__ = "rule_usage_events"
+
+    # BigInteger throughout, where the note twin uses Integer. `rule_id` has to
+    # be, since `rules.id` is BigInteger — and once one column is, matching the
+    # rest costs nothing and keeps the row uniform. A high-churn append-only
+    # telemetry table is a poor place to discover an id ceiling.
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    user_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True)
+    rule_id: Mapped[int] = mapped_column(BigInteger, nullable=False)
+
+    # 'surfaced' | 'pulled'
+    event: Mapped[str] = mapped_column(Text, nullable=False)
+
+    # Which surface produced it. A CONVENTION, not a fixed vocabulary, and the
+    # note twin's comment explains why this one deliberately does not enumerate
+    # its members: the previous such list went stale, naming a source nothing
+    # wrote while omitting ones that existed, and a half-true enumeration reads
+    # as authoritative in exactly the way that misleads (#2476).
+    # `grep -rn record_rule_pulled\|record_rule_surfaced src/` is the
+    # authoritative list, and unlike a comment it cannot drift.
+    #
+    # The mcp_/rest_ prefix split is load-bearing here for the same reason it is
+    # for notes, and more so: "is this rule dead weight?" is served by any pull,
+    # but "did that injected hint land?" — the question this arm exists to
+    # answer — is served by AGENT pulls only. Never aggregate across the prefix
+    # without saying why.
+    source: Mapped[str] = mapped_column(Text, nullable=False)
+
+    __table_args__ = (
+        # Every readout is "these rule ids, split by event" — a covering
+        # composite beats separate single-column indexes for it.
+        Index("ix_rule_usage_rule_event", "rule_id", "event"),
+        Index("ix_rule_usage_created_at", "created_at"),
+        Index("ix_rule_usage_user_id", "user_id"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "created_at": iso(self.created_at),
+            "user_id": self.user_id,
+            "rule_id": self.rule_id,
+            "event": self.event,
+            "source": self.source,
+        }

src/scribe/services/background.py

@@ -6,20 +6,63 @@ write that never errors and never lands (the #2663 GC footgun). This module is
 the one place that gets the pattern right: strong references in ``_pending``,
 discarded on completion, with failures logged at WARNING instead of vanishing.
 
-``note_usage`` and ``retrieval_telemetry`` predate this module and carry their
-own copies with bespoke canary semantics; new fire-and-forget callers use this
-instead of writing a fourth copy.
+``retrieval_telemetry`` predates this module and keeps its own copy, because
+its canary is a genuinely different shape — one process-wide flag and no
+AppLog row. ``note_usage`` and ``rule_usage`` share ``report_telemetry_failure``
+below. New fire-and-forget callers use ``spawn`` rather than writing another
+copy of the strong-reference dance.
 """
 from __future__ import annotations
 
 import asyncio
 import logging
+import traceback
 from collections.abc import Coroutine
 
 logger = logging.getLogger(__name__)
 
 _pending: set[asyncio.Task] = set()
 
+# Sites that have already dropped their once-per-process AppLog row, keyed
+# "<subsystem>:<site>". A readout can run on every list render — without this,
+# a broken table turns the error log into a firehose that buries the finding it
+# exists to surface.
+_reported: set[str] = set()
+
+
+async def report_telemetry_failure(subsystem: str, site: str) -> None:
+    """Make a swallowed telemetry failure visible. Call from an except block.
+
+    WARNING to the process log every time; one AppLog error row per process per
+    (subsystem, site) so the admin UI shows the outage without host access.
+
+    THIS IS NOT DECORATION. #2663 is the record of a telemetry subsystem running
+    at zero for weeks — every counter reading empty, indistinguishable from
+    "nobody uses this" — because every failure went to ``logger.debug``. A
+    subsystem whose failures are all invisible cannot report its own death.
+
+    The AppLog write is itself guarded: when the database is down it fails too,
+    and that is fine. The WARNING already said so, and a canary must never take
+    down the surface it watches.
+    """
+    logger.warning("%s telemetry %s failed", subsystem, site, exc_info=True)
+    key = f"{subsystem}:{site}"
+    if key in _reported:
+        return
+    _reported.add(key)
+    try:
+        from scribe.services.logging import log_error
+
+        await log_error(
+            endpoint=subsystem,
+            error_type=f"{subsystem}_{site}_failed",
+            error_message=f"{subsystem} telemetry {site} is failing; "
+            "usage counters will read zero until this is fixed",
+            traceback=traceback.format_exc(),
+        )
+    except Exception:
+        logger.debug("%s canary write failed", subsystem, exc_info=True)
+
 
 def spawn(coro: Coroutine, *, site: str) -> None:
     """Schedule ``coro`` fire-and-forget; ``site`` names it in failure logs.

src/scribe/services/backup.py

@@ -12,6 +12,7 @@ from scribe.models.note_version import NoteVersion
 from scribe.models.rule_version import RuleVersion
 from scribe.models.design_system import DesignSystem, DesignToken
 from scribe.models.note_usage import NoteUsageEvent
+from scribe.models.rule_usage import RuleUsageEvent
 from scribe.models.canonical_system import CanonicalSystem
 from scribe.models.rulebook import RuleRelation, rule_systems as rule_systems_t
 from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
@@ -62,8 +63,12 @@ logger = logging.getLogger(__name__)
 # _COLUMN_EXCLUSIONS and its guard landed with it, so the next such column
 # fails the build instead.
 # v13 (2026-08) added rule_versions — a rule's edit history (milestone 323).
+# v14 (2026-09) added rule_usage_events — the rule twin of note_usage_events
+# (milestone 333). Carrying it is the WHOLE REASON the table is separate: the
+# note importer maps note_id through note_id_map, so a rule id parked there
+# would restore attached to whatever note took that number.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 13
+BACKUP_VERSION = 14
 
 # Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
 # below, these two lists must together account for the entire schema — which is
@@ -92,6 +97,11 @@ _BACKED_UP = [
     # v13 (2026-08): a rule's edit history (milestone 323). note_versions has
     # always travelled; its sibling has no excuse not to.
     "rule_versions",
+    # v14 (2026-09): rule usage telemetry (milestone 333). Same argument
+    # note_usage_events makes for itself — pull-through is only ever
+    # accumulated, so a restore that dropped it would silently reset the
+    # measurement to zero while everything still looked fine.
+    "rule_usage_events",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -178,6 +188,8 @@ _COLUMN_EXCLUSIONS: dict[str, set[str]] = {
     "note_supersessions": {"id", "created_at"},
     "rule_relations": {"id", "created_at"},
     "note_usage_events": {"id"},
+    # Same as the note twin: the surrogate key is re-issued on insert.
+    "rule_usage_events": {"id"},
     "design_systems": {"deleted_at", "deleted_batch_id", "created_at", "updated_at"},
     "design_tokens": {"deleted_at", "deleted_batch_id", "created_at", "updated_at"},
     "repo_bindings": {"id", "created_at", "updated_at"},
@@ -321,6 +333,17 @@ def _usage_event_rows(rows) -> list[dict]:
     ]
 
 
+def _rule_usage_event_rows(rows) -> list[dict]:
+    return [
+        {
+            "user_id": r.user_id, "rule_id": r.rule_id, "event": r.event,
+            "source": r.source,
+            "created_at": r.created_at.isoformat() if r.created_at else None,
+        }
+        for r in rows
+    ]
+
+
 def _code_shape_rows(rows) -> list[dict]:
     return [r.to_dict() for r in rows]
 
@@ -606,6 +629,9 @@ async def export_full_backup() -> dict:
         )).scalars().all()
         design_tokens = (await session.execute(select(DesignToken))).scalars().all()
         usage_events = (await session.execute(select(NoteUsageEvent))).scalars().all()
+        rule_usage_events = (
+            await session.execute(select(RuleUsageEvent))
+        ).scalars().all()
         repo_bindings = (await session.execute(select(RepoBinding))).scalars().all()
         code_shapes = (await session.execute(select(CodeShape))).scalars().all()
         code_shape_events = (await session.execute(
@@ -665,6 +691,7 @@ async def export_full_backup() -> dict:
         "design_systems": _design_system_rows(design_systems),
         "design_tokens": _design_token_rows(design_tokens),
         "note_usage_events": _usage_event_rows(usage_events),
+        "rule_usage_events": _rule_usage_event_rows(rule_usage_events),
         "repo_bindings": _repo_binding_rows(repo_bindings),
         "note_supersessions": _note_supersession_rows(supersessions),
         "code_shapes": _code_shape_rows(code_shapes),
@@ -740,6 +767,14 @@ async def export_user_backup(user_id: int) -> dict:
         usage_events = (await session.execute(
             select(NoteUsageEvent).where(NoteUsageEvent.note_id.in_(note_ids))
         )).scalars().all() if note_ids else []
+        # Scoped through the RULE, not the event's user_id — the same call
+        # rule_versions makes one block up. user_id here is whoever the arm
+        # fired for, so filtering on it would carry this user's surfacings of
+        # someone ELSE's rule and drop the ones fired for someone else on
+        # theirs: the opposite of a per-user export.
+        rule_usage_events = (await session.execute(
+            select(RuleUsageEvent).where(RuleUsageEvent.rule_id.in_(_rule_ids))
+        )).scalars().all() if _rule_ids else []
         repo_bindings = (await session.execute(
             select(RepoBinding).where(RepoBinding.user_id == user_id)
         )).scalars().all()
@@ -858,6 +893,7 @@ async def export_user_backup(user_id: int) -> dict:
         "design_systems": _design_system_rows(design_systems),
         "design_tokens": _design_token_rows(design_tokens),
         "note_usage_events": _usage_event_rows(usage_events),
+        "rule_usage_events": _rule_usage_event_rows(rule_usage_events),
         "repo_bindings": _repo_binding_rows(repo_bindings),
         "note_supersessions": _note_supersession_rows(supersessions),
         "code_shapes": _code_shape_rows(code_shapes),
@@ -994,7 +1030,8 @@ async def _restore_v2(data: dict) -> dict:
         "rulebook_subscriptions": 0, "rule_suppressions": 0,
         "topic_suppressions": 0, "rulebook_exclusions": 0,
         "systems": 0, "record_systems": 0, "design_systems": 0,
-        "design_tokens": 0, "note_usage_events": 0, "repo_bindings": 0,
+        "design_tokens": 0, "note_usage_events": 0, "rule_usage_events": 0,
+        "repo_bindings": 0,
         "note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
         "code_shape_uses": 0, "canonical_systems": 0,
         "rule_systems": 0, "rule_relations": 0, "rule_versions": 0,
@@ -1496,6 +1533,25 @@ async def _restore_v2(data: dict) -> dict:
             ))
             stats["note_usage_events"] += 1
 
+        # The rule twin — and the reason it is a separate table at all.
+        # Resolved through rule_id_map, NOT note_id_map. A rule id run through
+        # the note map would either drop (best case) or land on whatever note
+        # took that number, producing telemetry that is wrong rather than
+        # missing and that nothing downstream could detect (milestone 333).
+        # Must come after the rules themselves; rule_id_map is populated there.
+        for ev in data.get("rule_usage_events", []):
+            mapped_rid = rule_id_map.get(ev.get("rule_id", 0))
+            if mapped_rid is None:
+                continue
+            session.add(RuleUsageEvent(
+                user_id=user_id_map.get(ev.get("user_id") or 0),
+                rule_id=mapped_rid,
+                event=ev.get("event", ""),
+                source=ev.get("source", ""),
+                created_at=_dt(ev.get("created_at")),
+            ))
+            stats["rule_usage_events"] += 1
+
         # 20. Repo bindings — small, but losing them means every bound repo
         # quietly stops loading its project at session start.
         for rb_data in data.get("repo_bindings", []):

src/scribe/services/note_usage.py

@@ -30,13 +30,13 @@ from __future__ import annotations
 
 import asyncio
 import logging
-import traceback
 
 from sqlalchemy import case, func, select
 
 from scribe.models import async_session
 from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
 from scribe.models.base import iso
+from scribe.services.background import report_telemetry_failure
 
 logger = logging.getLogger(__name__)
 
@@ -46,37 +46,19 @@ logger = logging.getLogger(__name__)
 # never lands. The done-callback discard keeps the set from growing.
 _pending: set[asyncio.Task] = set()
 
-# Sites that already dropped their once-per-process AppLog row. The readout
-# runs on every snippet list render — without this, a broken table would turn
-# the error log into a firehose that buries the finding it exists to surface.
-_reported: set[str] = set()
-
 
 async def _report_failure(site: str) -> None:
-    """Make a swallowed telemetry failure visible. Called from an except block.
+    """This subsystem's canary, now the shared one.
 
-    WARNING to the process log every time; one AppLog error row per process per
-    site so the admin UI shows the outage without host access. The AppLog write
-    is itself guarded — when the whole database is down it fails too, and that
-    is fine: the WARNING already said so, and a canary must never take down the
-    surface it watches.
+    The per-site dedup, the WARNING and the single AppLog row all moved to
+    `background.report_telemetry_failure` unchanged when `rule_usage` needed
+    the identical behaviour — two hand-kept copies of a thing whose whole job
+    is to be reliable is the wrong number. `retrieval_telemetry` deliberately
+    still has its own: its canary is a different shape (one process-wide flag,
+    no AppLog row), so repointing it would change behaviour rather than
+    consolidate it.
     """
-    logger.warning("note usage telemetry %s failed", site, exc_info=True)
-    if site in _reported:
-        return
-    _reported.add(site)
-    try:
-        from scribe.services.logging import log_error
-
-        await log_error(
-            endpoint="note_usage",
-            error_type=f"note_usage_{site}_failed",
-            error_message=f"note usage telemetry {site} is failing; "
-            "usage counters will read zero until this is fixed",
-            traceback=traceback.format_exc(),
-        )
-    except Exception:
-        logger.debug("note usage canary write failed", exc_info=True)
+    await report_telemetry_failure("note_usage", site)
 
 
 async def _insert_events(rows: list[dict]) -> None:

src/scribe/services/rule_usage.py

@@ -0,0 +1,193 @@
+"""Rule usage telemetry — did a surfaced rule ever get read?
+
+The sibling of `note_usage`, for the one retrieval surface in Scribe that
+could not be measured at all.
+
+Two event streams, deliberately independent:
+
+  - SURFACED: the write-path standing-rule arm put this rule in front of the
+    agent, unbidden, during a write.
+  - PULLED: someone then opened it in full (`get_rule`, or the REST detail
+    route).
+
+WHY THIS ARM AND NOT ANOTHER. Every other surface declines most of the time —
+`write_path` returns nothing on 78% of calls, `reuse_slot` on 79%, auto-inject
+on 39%. The rule arm has never once returned nothing (#3311). That is either a
+perfectly tuned surface or a bar it cannot fail to clear, and `retrieval_logs`
+cannot tell the two apart: it records what the ranker scored, never whether the
+hint was any use. The ratio these two streams produce is the missing half, and
+without it any threshold change is a number picked off a histogram.
+
+Design notes, mirroring `note_usage`:
+  - Writes are fire-and-forget through `background.spawn`, so telemetry never
+    adds latency to — or can break — the surface it observes. This module does
+    NOT carry its own copy of the strong-reference dance; `background` is the
+    one place that gets it right, and a fourth copy is how one of them drifts.
+  - Failures degrade, but never SILENTLY. `report_telemetry_failure` logs at
+    WARNING and drops one AppLog row per process per site. #2663 is the record
+    of this exact subsystem class running at zero for weeks — indistinguishable
+    from "nobody uses this" — because every failure went to `logger.debug`.
+  - Reads (`usage_for_rules`) are awaited and aggregated in one round-trip for
+    a whole page, never per row.
+
+NO AMBIENT BUCKET, YET — and that is a decision, not an omission. The note twin
+splits ranked surfacings from ambient ones because `enter_project` and the
+skill sync put records in front of the agent without choosing them, and
+counting those as surfacings makes recency read as popularity (#2477). Rules
+have the same shape of problem waiting: `list_always_on_rules` and
+`enter_project` load rules wholesale on every session. They do not emit here
+today, so there is nothing to bucket, and an empty `AMBIENT_SOURCES` would be
+machinery pretending to a distinction the data does not yet contain. When a
+bulk surface starts emitting, the split is a readout-level change — a tuple and
+a `case()`, exactly as in the twin — and needs no migration. Keep it that way:
+`source` stays granular so the choice remains available.
+"""
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import func, select
+
+from scribe.models import async_session
+from scribe.models.base import iso
+from scribe.models.rule_usage import PULLED, SURFACED, RuleUsageEvent
+from scribe.services.background import report_telemetry_failure, spawn
+
+logger = logging.getLogger(__name__)
+
+
+async def _report_failure(site: str) -> None:
+    await report_telemetry_failure("rule_usage", site)
+
+
+async def _insert_events(rows: list[dict]) -> None:
+    """Persist usage rows. Best-effort: failures degrade, visibly."""
+    try:
+        async with async_session() as session:
+            session.add_all([RuleUsageEvent(**row) for row in rows])
+            await session.commit()
+    except Exception:
+        await _report_failure("write")
+
+
+def _schedule(rows: list[dict]) -> None:
+    if not rows:
+        return
+    spawn(_insert_events(rows), site="rule_usage_write")
+
+
+def record_rule_surfaced(
+    *, user_id: int | None, rule_ids: list[int] | set[int], source: str
+) -> None:
+    """Fire-and-forget: record that these rules were shown to the agent.
+
+    Takes the whole hint at once — one insert per surfacing event, not per rule
+    — because a hint is a single decision and its rows should land together.
+
+    Record the RANKED hits only. The arm filters candidates before it speaks
+    (`exclude_rule_ids` drops what the session already holds), and a rule that
+    was considered and not shown was not surfaced. Counting those would inflate
+    the denominator with claims the agent never saw, which reads as a precision
+    problem the arm does not have.
+    """
+    try:
+        rows = [
+            {
+                "user_id": user_id,
+                "rule_id": int(rid),
+                "event": SURFACED,
+                "source": source,
+            }
+            for rid in rule_ids
+        ]
+    except Exception:
+        logger.debug("rule usage payload build failed", exc_info=True)
+        return
+    _schedule(rows)
+
+
+def record_rule_pulled(*, user_id: int | None, rule_id: int, source: str) -> None:
+    """Fire-and-forget: record that a rule was opened in full.
+
+    A PULL is somebody choosing to open one record. `list_always_on_rules` and
+    `enter_project` are NOT pulls — they are bulk resident loads that hand over
+    every applicable rule at once, and counting them would swamp the signal
+    with the very ambient delivery the ratio exists to distinguish from.
+    """
+    try:
+        rows = [
+            {
+                "user_id": user_id,
+                "rule_id": int(rule_id),
+                "event": PULLED,
+                "source": source,
+            }
+        ]
+    except Exception:
+        logger.debug("rule usage payload build failed", exc_info=True)
+        return
+    _schedule(rows)
+
+
+def empty_rule_usage() -> dict:
+    """The zero readout — what a rule with no recorded events looks like.
+
+    Callers render this shape unconditionally, so a rule predating the table
+    reads as "never surfaced, never pulled" rather than as a missing key. That
+    distinction matters more here than for notes: every rule in an install
+    predates this table, so for a while "no events" is the normal state and it
+    must not look like a broken readout.
+    """
+    return {
+        "surfaced_count": 0,
+        "pull_count": 0,
+        "last_surfaced_at": None,
+        "last_pulled_at": None,
+    }
+
+
+async def usage_for_rules(rule_ids: list[int]) -> 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".
+    """
+    ids = [int(r) for r in rule_ids]
+    out: dict[int, dict] = {rid: empty_rule_usage() for rid in ids}
+    if not ids:
+        return out
+
+    try:
+        async with async_session() as session:
+            rows = (
+                await session.execute(
+                    select(
+                        RuleUsageEvent.rule_id,
+                        RuleUsageEvent.event,
+                        func.count().label("n"),
+                        func.max(RuleUsageEvent.created_at).label("last_at"),
+                    )
+                    .where(RuleUsageEvent.rule_id.in_(ids))
+                    .group_by(RuleUsageEvent.rule_id, RuleUsageEvent.event)
+                )
+            ).all()
+    except Exception:
+        # A telemetry readout must not be able to break the list it decorates —
+        # but it must say it failed, or a broken readout is indistinguishable
+        # from a corpus nobody uses (#2663).
+        await _report_failure("readout")
+        return out
+
+    for rule_id, event, n, last_at in rows:
+        slot = out.get(int(rule_id))
+        if slot is None:
+            continue
+        if event == SURFACED:
+            slot["surfaced_count"] = int(n)
+            slot["last_surfaced_at"] = iso(last_at)
+        elif event == PULLED:
+            slot["pull_count"] = int(n)
+            slot["last_pulled_at"] = iso(last_at)
+    return out