feat(telemetry): the preload emits, and the always-on set stops being unfalsifiable (#3473)

The ranked rule arm became measurable in M333. The preload did not — and
that is the surface whose value is actually in question. `list_always_on_rules`,
the SessionStart block and every `rules_payload` caller handed rules over
wholesale and emitted nothing, so the resident set's token cost was certain
and its usefulness could not be tested even in principle.

Bulk deliveries now record as AMBIENT, beside the ranked count and never
inside pull-through. Folding them in would mean growing the always-on set
depressed the arm's measured precision and trimming it flattered the arm,
neither for any reason to do with the arm.

`RANKED_SOURCES` inverts the note twin's `AMBIENT_SOURCES` deliberately: there
is one ranked rule source and this change adds seven bulk ones, so naming the
rare half makes a forgotten surface default to ambient — under-counting it —
rather than padding the denominator with surfacings nobody chose.

Two lookalike call sites are deliberately left silent, with a test to keep
them that way: the write-path etag arm and `rules_etag_for` read the rules to
build or compare a MARKER and show nobody anything.

No migration — `event` and `source` are plain Text with no CHECK (rule 36
does not apply). Snippet #2858 updated to the new `rules_payload` contract.

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

src/scribe/mcp/tools/milestones.py

@@ -57,7 +57,7 @@ async def get_milestone(milestone_id: int) -> dict:
     return {
         "milestone": out,
         "steps": [t.to_dict() for t in steps],
-        **rulebooks_svc.rules_payload(applicable),
+        **rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_milestone"),
     }
 
 

src/scribe/mcp/tools/projects.py

@@ -207,7 +207,7 @@ async def enter_project(project_id: int) -> dict:
         ],
         "design_system": design_system,
         "milestone_summary": milestone_summary,
-        **rulebooks_svc.rules_payload(applicable),
+        **rulebooks_svc.rules_payload(applicable, user_id=uid, source="enter_project"),
         "open_tasks": [
             {
                 "id": t.id, "title": t.title, "status": t.status,
@@ -251,7 +251,7 @@ async def get_project(project_id: int) -> dict:
     applicable = await rulebooks_svc.get_applicable_rules(
         project_id=project_id, user_id=uid,
     )
-    data.update(rulebooks_svc.rules_payload(applicable))
+    data.update(rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_project"))
     return data
 
 

src/scribe/mcp/tools/rulebooks.py

@@ -18,7 +18,7 @@ from scribe.mcp._context import current_user_id
 from scribe.services import dedup as dedup_svc
 from scribe.services import rulebooks as rulebooks_svc
 from scribe.services import trash as trash_svc
-from scribe.services.rule_usage import record_rule_pulled
+from scribe.services.rule_usage import record_rule_pulled, record_rule_surfaced
 
 
 # ── Rulebook CRUD ───────────────────────────────────────────────────────
@@ -265,6 +265,15 @@ async def list_always_on_rules(project_id: int = 0) -> dict:
     """
     uid = current_user_id()
     rules = await rulebooks_svc.list_always_on_rules(uid, project_id=project_id)
+    # AMBIENT source: the resident set, handed over whole. No ranker chose
+    # these, so they must not land in the pull-through numerator's denominator
+    # — but they must land SOMEWHERE, or the largest rule surface in the
+    # product stays the one surface its own scoreboard cannot see (#3473).
+    record_rule_surfaced(
+        user_id=uid,
+        rule_ids=[r.id for r in rules],
+        source="list_always_on_rules",
+    )
     return {
         "rules": [_rule_summary(r) for r in rules],
         "total": len(rules),

src/scribe/mcp/tools/search.py

@@ -197,17 +197,31 @@ It is an UPPER BOUND per surface: a pull records the door it came
     query failed while the rest of the readout stood.
 
     `rule_usage` — the same question for RULES, from `rule_usage_events`:
-    `surfaced`, `pulled` split into `pulled_by_agent` / `pulled_by_human`, the
-    distinct-rule counts, and `pull_through` on the same definition (agent
-    pulls over surfacings).
+    `surfaced` and `ambient`, `pulled` split into `pulled_by_agent` /
+    `pulled_by_human`, the distinct-rule counts, and `pull_through` on the same
+    definition (agent pulls over RANKED surfacings).
 
     A SEPARATE BLOCK, not folded into `usage`, and reading it as one number
     with that is the mistake to avoid. The corpora differ by orders of
     magnitude — a few dozen eligible rules against thousands of notes — so a
     blended ratio would be the note ratio with noise on it and would hide the
-    rule arm entirely. It also has no `ambient` key, because nothing surfaces a
-    rule un-ranked: `list_always_on_rules` and `enter_project` hand over rules
-    wholesale but emit no event, so there is no ambient class to separate.
+    rule arm entirely.
+
+    `surfaced` VS `ambient` IS THE READING THAT MATTERS HERE. `surfaced` counts
+    rules a ranker chose — today only the write-path arm — and those are claims
+    a pull can settle. `ambient` counts BULK DELIVERIES: the SessionStart
+    preload, `list_always_on_rules`, and the `rules_payload` surfaces
+    (`enter_project`, `get_project`, `get_milestone`, `start_planning`,
+    `get_task`), which hand over the whole applicable set at once with nobody
+    choosing anything. A large `ambient` says the resident set is big and
+    arrives often — never that it is useful, and never that it is read.
+
+    `pull_through` therefore divides by `surfaced` alone. Fold the preload in
+    and growing the always-on set would depress the arm's measured precision
+    while trimming it would flatter it, for reasons having nothing to do with
+    the arm. To judge the PRELOAD instead, compare `ambient` against pulls of
+    those same rules over time: a resident set surfaced thousands of times and
+    opened never is the dead-weight signal, one tier up.
 
     Read it against `sources["write_path_rule"]`. That surface has never once
     declined to fire, and until this block existed there was no way to tell a

src/scribe/mcp/tools/tasks.py

@@ -103,7 +103,7 @@ async def get_task(task_id: int) -> dict:
         applicable = await rulebooks_svc.get_applicable_rules(
             project_id=note.project_id, user_id=uid,
         )
-        data.update(rulebooks_svc.rules_payload(applicable))
+        data.update(rulebooks_svc.rules_payload(applicable, user_id=uid, source="get_task"))
     data.update(await access_svc.describe_provenance(uid, note))
     # Same reasoning as get_note's record_pulled, and this is the tool where it
     # matters MOST: auto-inject ranks kind-blind over a corpus that is

src/scribe/services/planning.py

@@ -60,7 +60,7 @@ async def start_planning(user_id: int, project_id: int, title: str) -> dict:
 
     return {
         "milestone": milestone.to_dict(),
-        **rulebooks_svc.rules_payload(applicable),
+        **rulebooks_svc.rules_payload(applicable, user_id=user_id, source="start_planning"),
         "project_goal": getattr(project, "goal", "") or "",
         "open_task_count": open_count,
     }

src/scribe/services/plugin_context.py

@@ -1375,6 +1375,24 @@ async def build_session_context(
     # exclusion (milestone 297) takes a rulebook out of this block, and is
     # named below so the departure is visible rather than silent.
     rules = await rulebooks_svc.list_always_on_rules(user_id, project_id=project_id)
+    # AMBIENT source, and the one that matters most: this is the preload — the
+    # block every session opens with, chosen by nobody, paid for every turn.
+    #
+    # It emitted nothing until 2026-09-03, which made the resident set's cost
+    # certain and its usefulness unfalsifiable at the same time (#3473). Note
+    # #3089 is the argument this measurement finally lets someone test: that a
+    # rule arriving with thirty others, none of them relevant, is read as
+    # preamble rather than as a claim — so presence is not surfacing, and a
+    # tier-1 set can grow without anybody noticing it stopped working.
+    #
+    # Recorded even when the hook truncates the block below: the rules WERE
+    # delivered, and counting only the untruncated ones would quietly shrink
+    # the denominator exactly where the set is too big to read.
+    record_rule_surfaced(
+        user_id=user_id,
+        rule_ids=[r.id for r in rules],
+        source="session_start",
+    )
     excluded = (
         await rulebooks_svc.excluded_always_on_rulebooks(user_id, project_id)
         if project_id else []

src/scribe/services/retrieval_telemetry.py

@@ -30,6 +30,7 @@ from scribe.models.note_usage import PULLED, SURFACED, NoteUsageEvent
 from scribe.models.rule_usage import PULLED as RULE_PULLED
 from scribe.models.rule_usage import SURFACED as RULE_SURFACED
 from scribe.models.rule_usage import RuleUsageEvent
+from scribe.services.rule_usage import is_ambient
 from scribe.models.retrieval_log import RetrievalLog
 
 logger = logging.getLogger(__name__)
@@ -428,10 +429,15 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
                         .group_by(RuleUsageEvent.event, RuleUsageEvent.source)
                     )
                 ).all()
-                # No AMBIENT exclusion here, unlike the note twin: nothing
-                # surfaces a rule un-ranked yet. `list_always_on_rules` and
-                # `enter_project` deliver rules wholesale but emit no event, so
-                # there is no ambient class to subtract (milestone 333 step 1).
+                # The rows carry `source`, so the ranked/ambient split is done
+                # below rather than in SQL — the bulk surfaces started emitting
+                # on 2026-09-03 (#3473), so there IS an ambient class now.
+                #
+                # `distinct_rules_surfaced` deliberately counts BOTH classes. It
+                # answers "how many distinct rules did this install put in front
+                # of an agent at all", which is the denominator for dead weight
+                # — and a rule delivered by the preload a hundred times and
+                # never opened is the most important case that question has.
                 distinct_rules_surfaced = (
                     await session.execute(
                         select(func.count(func.distinct(RuleUsageEvent.rule_id)))
@@ -545,11 +551,18 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
     # been comparing across windows, without telling them it now measures
     # something else.
     #
-    # No `ambient` key, unlike its twin. Nothing surfaces a rule un-ranked yet;
-    # the absence is a fact about the data rather than an oversight, and it
-    # returns the moment a bulk loader starts emitting.
+    # `ambient` now carries the bulk deliveries — the SessionStart preload,
+    # `list_always_on_rules`, and every `rules_payload` surface (#3473). Before
+    # they emitted, this block had no ambient key and said the absence was a
+    # fact about the data. It was, and it was also the thing that made the
+    # always-on set impossible to judge: the largest rule surface in the
+    # product was the one surface its own scoreboard could not see.
+    #
+    # READ THE TWO SEPARATELY, ALWAYS. `surfaced` is a claim a ranker made and
+    # a pull can settle. `ambient` is a delivery nobody chose, so a high count
+    # says the set is large and resident, never that it is useful.
     rule_usage = {
-        "surfaced": 0,
+        "surfaced": 0, "ambient": 0,
         "pulled": 0, "pulled_by_agent": 0, "pulled_by_human": 0,
         "distinct_rules_surfaced": int(distinct_rules_surfaced or 0),
         "distinct_rules_pulled": int(distinct_rules_pulled or 0),
@@ -565,7 +578,14 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
         for event, source, n in rule_rows:
             n = int(n)
             if event == RULE_SURFACED:
-                rule_usage["surfaced"] += n
+                # One definition of ranked-vs-ambient, imported rather than
+                # restated — the per-rule badge readout reads the same
+                # predicate, and two spellings of "what counts as surfaced" is
+                # precisely the uneven wiring #3246 found across this system.
+                if is_ambient(source):
+                    rule_usage["ambient"] += n
+                else:
+                    rule_usage["surfaced"] += n
             elif event == RULE_PULLED:
                 rule_usage["pulled"] += n
                 # Same split, and it carries MORE weight here than for notes.
@@ -582,6 +602,15 @@ async def retrieval_summary(user_id: int | None, *, days: int = 30) -> dict:
     # empty numerator AND denominator that is a claim the data does not
     # support, and it is the reading that would make a brand-new install look
     # like a broken one.
+    #
+    # RANKED SURFACINGS ONLY in the denominator, and this is the load-bearing
+    # line of the whole change. Pull-through asks "was that hint any use", and
+    # only a surface that CHOSE what it showed can be judged by it. Folding the
+    # preload in would divide the same pulls by a number that grows with every
+    # session and every rule added to the resident set — so enlarging the
+    # always-on set would DEPRESS the arm's measured precision, and trimming it
+    # would flatter it, neither for any reason to do with the arm. The ambient
+    # count sits beside it, unaveraged, and is read as size rather than skill.
     rule_usage["pull_through"] = (
         round(rule_usage["pulled_by_agent"] / rule_usage["surfaced"], 4)
         if rule_usage["surfaced"] else None

src/scribe/services/rule_usage.py

@@ -30,23 +30,43 @@ Design notes, mirroring `note_usage`:
   - 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.
+AMBIENT VS RANKED. 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 exactly that shape: the SessionStart preload,
+`list_always_on_rules`, and every `rules_payload` surface hand over the whole
+applicable set at once, chosen by nobody.
+
+Until 2026-09-03 those bulk surfaces emitted nothing, and this module said so —
+"an empty `AMBIENT_SOURCES` would be machinery pretending to a distinction the
+data does not yet contain". True as far as it went, but it had a consequence
+worth naming, because it is the reason the bucket exists now: the always-on
+set's token cost was certain and its usefulness was UNFALSIFIABLE, permanently
+and by construction. The one surface whose value was actually in question was
+the one surface exempt from the scoreboard that judges every other.
+
+They emit now. The split is the readout-level change the old note promised — a
+`case()`, no migration, because `event` and `source` are plain Text with no
+CHECK constraint. `source` stays granular so a reader can still tell the
+preload from `enter_project` from the ranked arm.
+
+WHY THIS NAMES THE RANKED SOURCES AND THE TWIN NAMES THE AMBIENT ONES. A
+deliberate divergence, on the failure mode rather than on symmetry. Both shapes
+fail silently when someone adds a surface and forgets the list, so the question
+is which list changes more often — and here it is emphatically the ambient one:
+there is exactly ONE ranked rule source, and this change alone adds seven bulk
+ones. Naming the rare, stable half means a newly-added bulk surface defaults to
+`ambient`, which merely under-counts it, instead of defaulting to `ranked`,
+which would quietly pad the pull-through denominator with surfacings nobody
+chose and make the arm look imprecise. Same argument #3191 and #3430 make
+against hand-kept lists: keep the list that must be remembered as short and as
+slow-moving as possible.
 """
 from __future__ import annotations
 
 import logging
 
-from sqlalchemy import func, select
+from sqlalchemy import case, func, select
 
 from scribe.models import async_session
 from scribe.models.base import iso
@@ -55,6 +75,31 @@ from scribe.services.background import report_telemetry_failure, spawn
 
 logger = logging.getLogger(__name__)
 
+# The surfaces that CHOSE the rules they showed. Everything else is ambient —
+# see the module docstring for why the rare half is the half that gets named.
+#
+# Membership is the whole definition of the pull-through denominator: a ranked
+# surfacing is a claim ("this rule may apply to what you are doing") that a pull
+# can confirm or refute, while an ambient one is a delivery nobody decided on.
+# Add a source here only when a ranker picked it.
+RANKED_SOURCES = ("write_path_rule",)
+
+
+def is_ambient(source: str) -> bool:
+    """Was this surfacing a bulk delivery rather than a ranked choice?
+
+    One definition, read by both the per-rule badge readout and the aggregate
+    in `retrieval_telemetry` — the two used to be able to disagree about what
+    "surfaced" counted, which is the class of drift #3246 found across the
+    rules system.
+
+    Sync and pure, per the service canon (#2860), but deliberately PUBLIC where
+    that canon says such helpers stay `_private`. The departure is the point:
+    a module-private copy in each caller is exactly the second definition this
+    exists to prevent.
+    """
+    return source not in RANKED_SOURCES
+
 
 async def _report_failure(site: str) -> None:
     await report_telemetry_failure("rule_usage", site)
@@ -81,14 +126,21 @@ def record_rule_surfaced(
 ) -> 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.
+    Takes the whole delivery 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.
+    Record what was actually SHOWN, never what was considered. For the ranked
+    arm that means the post-filter hits: it drops what the session already
+    holds (`exclude_rule_ids`) before it speaks, and a rule 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.
+
+    Bulk surfaces pass their whole delivered set, which is the same rule read
+    from the other end — everything in a preload IS shown. `source` is what
+    separates the two afterwards (see `RANKED_SOURCES`); this function does not
+    care which kind it is recording.
     """
     try:
         rows = [
@@ -137,9 +189,17 @@ def empty_rule_usage() -> dict:
     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.
+
+    `surfaced_count` is RANKED surfacings only; `ambient_count` is the bulk
+    deliveries (see `RANKED_SOURCES`). The split is what keeps the badge's
+    "shown often, opened never → dead weight" reading honest: every rule in an
+    always-on set is delivered every session, so an unsplit counter would rank
+    the resident set as the most-surfaced rules in the install purely for being
+    resident.
     """
     return {
         "surfaced_count": 0,
+        "ambient_count": 0,
         "pull_count": 0,
         "last_surfaced_at": None,
         "last_pulled_at": None,
@@ -159,6 +219,19 @@ async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
     if not ids:
         return out
 
+    # Classified in SQL so the group stays small: per rule we get at most
+    # (surfaced-ranked, surfaced-ambient, pulled) rather than a row per distinct
+    # source. ONE labelled expression, bound to a variable and reused in the
+    # GROUP BY — a second `case()` instance there renders its own expanding-IN
+    # bind names under asyncpg, so the database sees two DIFFERENT expressions
+    # and rejects the query with a GroupingError. The note twin carries the
+    # same warning for the same reason, and #2663 is what it cost: the
+    # rejection was swallowed and every counter read zero in production while
+    # the writes were landing fine.
+    ambient = case(
+        (RuleUsageEvent.source.notin_(RANKED_SOURCES), True),
+        else_=False,
+    ).label("ambient")
     try:
         async with async_session() as session:
             rows = (
@@ -168,9 +241,14 @@ async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
                         RuleUsageEvent.event,
                         func.count().label("n"),
                         func.max(RuleUsageEvent.created_at).label("last_at"),
+                        ambient,
                     )
                     .where(RuleUsageEvent.rule_id.in_(ids))
-                    .group_by(RuleUsageEvent.rule_id, RuleUsageEvent.event)
+                    .group_by(
+                        RuleUsageEvent.rule_id,
+                        RuleUsageEvent.event,
+                        ambient,
+                    )
                 )
             ).all()
     except Exception:
@@ -180,14 +258,23 @@ async def usage_for_rules(rule_ids: list[int]) -> dict[int, dict]:
         await _report_failure("readout")
         return out
 
-    for rule_id, event, n, last_at in rows:
+    for rule_id, event, n, last_at, is_amb in rows:
         slot = out.get(int(rule_id))
         if slot is None:
             continue
-        if event == SURFACED:
+        if event == SURFACED and is_amb:
+            slot["ambient_count"] = int(n)
+        elif 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)
+            # Pulls are pulls regardless of what surfaced the rule — "did
+            # anyone ever open this?" does not depend on how it was found. Both
+            # halves accumulate, so this ADDS rather than assigns: a rule can
+            # now be pulled after a ranked hint and after a preload, and the
+            # split arrives as two rows.
+            slot["pull_count"] = slot["pull_count"] + int(n)
+            latest = iso(last_at)
+            if latest and (slot["last_pulled_at"] or "") < latest:
+                slot["last_pulled_at"] = latest
     return out

src/scribe/services/rulebooks.py

@@ -23,6 +23,7 @@ from scribe.services.verification import (
 )
 from scribe.services import rule_versions
 from scribe.models.rule_version import RuleVersion
+from scribe.services.rule_usage import record_rule_surfaced
 
 logger = logging.getLogger(__name__)
 
@@ -1395,7 +1396,7 @@ async def get_applicable_rules(
     }
 
 
-def rules_payload(applicable: dict) -> dict:
+def rules_payload(applicable: dict, *, user_id: int | None, source: str) -> dict:
     """The caller-facing shape of a get_applicable_rules() result.
 
     Every surface that hands rules to an agent (enter_project, get_project,
@@ -1406,7 +1407,34 @@ def rules_payload(applicable: dict) -> dict:
     `excluded_always_on` (milestone 297) names the always-on rulebooks this
     project decided NOT to inherit, so the departure is visible wherever the
     rules are.
+
+    IT ALSO RECORDS THE SURFACING, which is why it now takes a caller and a
+    source. Every one of those surfaces is a bulk delivery — the applicable set
+    handed over whole, chosen by nobody — so this is the one place that has to
+    emit for all of them. Doing it per-caller instead would be five sites to
+    remember, and #3430 gap 2 is what that costs: the process→skill sync went
+    un-emitted through an entire dedicated telemetry survey because nothing
+    forced its surface to be accounted for.
+
+    `source` stays the CALLER's name rather than a constant, so the readout can
+    still separate the session handshake from a mid-session milestone read;
+    `RANKED_SOURCES` in `rule_usage` is what folds them back together.
+
+    Emitting from here is safe in a way emitting from `get_applicable_rules`
+    would not be: this function is only ever called to BUILD A REPLY. The two
+    other callers of the rules machinery — the write-path etag arm
+    (`plugin_context`) and `rules_etag_for` — compute a marker and show nobody
+    anything, and counting those would put rules in the denominator that no
+    agent ever saw.
     """
+    record_rule_surfaced(
+        user_id=user_id,
+        rule_ids=(
+            [r["id"] for r in applicable.get("rules", [])]
+            + [r["id"] for r in applicable.get("project_rules", [])]
+        ),
+        source=source,
+    )
     return {
         "applicable_rules": applicable["rules"],
         "applicable_rules_truncated": applicable["truncated"],

tests/test_inception_rules.py

@@ -15,14 +15,17 @@ def test_rules_payload_carries_excluded_always_on_as_the_seventh_key():
     out = rules_payload({
         "rules": [], "truncated": False, "subscribed_rulebooks": [],
         "excluded_always_on": [{"id": 1, "title": "Family"}],
-    })
+    }, user_id=1, source="enter_project")
     assert set(out) == {
         "applicable_rules", "applicable_rules_truncated", "subscribed_rulebooks",
         "project_rules", "suppressed_rules", "suppressed_topics", "excluded_always_on",
     }
     assert out["excluded_always_on"] == [{"id": 1, "title": "Family"}]
     # An older applicable dict without the key still renders (empty list).
-    assert rules_payload({"rules": [], "truncated": False, "subscribed_rulebooks": []})["excluded_always_on"] == []
+    assert rules_payload(
+        {"rules": [], "truncated": False, "subscribed_rulebooks": []},
+        user_id=1, source="enter_project",
+    )["excluded_always_on"] == []
 
 
 def test_list_always_on_rules_service_and_tool_take_a_project_id():

tests/test_rule_usage_wiring.py

@@ -274,3 +274,130 @@ def test_the_bulk_loaders_are_not_counted_as_pulls():
                 "applicable rule at once — so counting it would drown the "
                 "surfaced:pulled ratio in ambient delivery."
             )
+
+
+# ── The AMBIENT end: bulk deliveries (#3473) ───────────────────────────
+#
+# The preload was the largest rule surface in the product and emitted nothing,
+# so its cost was certain and its usefulness unfalsifiable. These assert the
+# three delivery shapes now emit — and, just as importantly, that the two
+# lookalike call sites which show nobody anything do NOT.
+
+
+@pytest.mark.asyncio
+async def test_the_session_start_preload_records_what_it_delivered():
+    """The block every session opens with. Chosen by nobody, paid for every
+    turn — and until it emitted, invisible to the scoreboard that judges every
+    other surface."""
+    from scribe.services import plugin_context as pc
+
+    rec = MagicMock()
+    rules = [fake_rule(id=1, title="`dev` is home"),
+             fake_rule(id=2, title="`main` — never without explicit request")]
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch.object(pc.rulebooks_svc, "list_always_on_rules",
+                         AsyncMock(return_value=rules))
+        )
+        stack.enter_context(
+            patch.object(pc.rulebooks_svc, "excluded_always_on_rulebooks",
+                         AsyncMock(return_value=[]))
+        )
+        stack.enter_context(patch.object(pc, "record_rule_surfaced", rec))
+        stack.enter_context(
+            patch.object(pc, "_topic_titles", AsyncMock(return_value={}))
+        )
+        await pc.build_session_context(1, project_id=0)
+
+    assert rec.call_count == 1, "the preload recorded nothing"
+    kw = rec.call_args.kwargs
+    assert kw["rule_ids"] == [1, 2]
+    assert kw["source"] == "session_start"
+
+
+@pytest.mark.asyncio
+async def test_the_always_on_tool_records_what_it_handed_over():
+    from scribe.mcp.tools import rulebooks as tools
+
+    rec = MagicMock()
+    rules = [fake_rule(id=3, title="No GitHub — Fabled-Git only")]
+    with ExitStack() as stack:
+        stack.enter_context(
+            patch.object(tools.rulebooks_svc, "list_always_on_rules",
+                         AsyncMock(return_value=rules))
+        )
+        stack.enter_context(
+            patch.object(tools.rulebooks_svc, "rules_etag",
+                         MagicMock(return_value="etag"))
+        )
+        stack.enter_context(patch.object(tools, "record_rule_surfaced", rec))
+        await tools.list_always_on_rules()
+
+    assert rec.call_args.kwargs["rule_ids"] == [3]
+    assert rec.call_args.kwargs["source"] == "list_always_on_rules"
+
+
+def test_rules_payload_records_both_the_family_and_project_halves():
+    """One emit site for all five `rules_payload` surfaces.
+
+    Per-caller emission would be five sites to remember, and #3430 gap 2 is
+    what that costs: the process→skill sync went un-emitted through an entire
+    dedicated telemetry survey because nothing forced its surface to be
+    accounted for.
+    """
+    from scribe.services import rulebooks as svc
+
+    rec = MagicMock()
+    with patch.object(svc, "record_rule_surfaced", rec):
+        svc.rules_payload(
+            {
+                "rules": [{"id": 10}, {"id": 11}],
+                "project_rules": [{"id": 12}],
+                "truncated": False,
+                "subscribed_rulebooks": [],
+            },
+            user_id=1,
+            source="enter_project",
+        )
+
+    kw = rec.call_args.kwargs
+    assert kw["rule_ids"] == [10, 11, 12], "project-scoped rules were delivered too"
+    assert kw["source"] == "enter_project"
+
+
+def test_every_rules_payload_caller_names_itself():
+    """`source` is the CALLER's name, so the readout can still separate the
+    session handshake from a mid-session milestone read. A shared constant here
+    would collapse five distinguishable surfaces into one."""
+    import re
+
+    seen = set()
+    for path in Path("src/scribe").rglob("*.py"):
+        for m in re.finditer(r"rules_payload\([^)]*source=\"([a-z_]+)\"", path.read_text()):
+            seen.add(m.group(1))
+    assert seen == {
+        "enter_project", "get_project", "get_milestone",
+        "start_planning", "get_task",
+    }, f"a rules_payload caller is missing or misnamed: {sorted(seen)}"
+
+
+def test_the_marker_paths_stay_silent():
+    """The two call sites that read the rules and show NOBODY anything.
+
+    `rules_etag_for` and the write-path staleness arm both call
+    `list_always_on_rules` to build or compare a marker. Emitting there would
+    put rules in the denominator that no agent ever saw — the exact inflation
+    `record_rule_surfaced`'s docstring forbids, arriving from the one direction
+    nothing else guards.
+    """
+    svc_src = Path("src/scribe/services/rulebooks.py").read_text()
+    etag_fn = svc_src.split("async def rules_etag_for")[1].split("\ndef ")[0]
+    assert "record_rule_surfaced" not in etag_fn, (
+        "rules_etag_for emits a surfacing — it builds a marker, it shows nothing"
+    )
+
+    pc_src = Path("src/scribe/services/plugin_context.py").read_text()
+    staleness = pc_src.split("if rules_etag:")[1].split("# The guard sits BELOW")[0]
+    assert "record_rule_surfaced" not in staleness, (
+        "the staleness arm emits a surfacing — it compares a marker, it shows nothing"
+    )

tests/test_services_retrieval_telemetry.py

@@ -511,3 +511,68 @@ async def test_rule_usage_sees_only_its_own_users_events(_dispose_engine):
         assert (await retrieval_summary(990014, days=30))["rule_usage"]["surfaced"] == 1
     finally:
         await cleanup()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_the_preload_lands_in_ambient_and_never_in_the_ratio(_dispose_engine):
+    """The split that makes the always-on set judgeable (#3473).
+
+    Pull-through asks "was that hint any use", and only a surface that CHOSE
+    what it showed can be judged by it. If the preload counted toward the
+    denominator, growing the always-on set would DEPRESS the arm's measured
+    precision and trimming it would flatter it — neither for any reason to do
+    with the arm. So the resident deliveries are counted, reported, and kept
+    out of the ratio.
+    """
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    cleanup = await _rule_events(990012, [
+        # One rule the arm actually chose, and opened.
+        (5101, "surfaced", "write_path_rule"),
+        (5101, "pulled", "mcp_get_rule"),
+        # Four bulk deliveries across every shape of preload. Nobody chose any
+        # of them, and none may touch the denominator.
+        (5102, "surfaced", "session_start"),
+        (5103, "surfaced", "list_always_on_rules"),
+        (5104, "surfaced", "enter_project"),
+        (5105, "surfaced", "get_milestone"),
+    ])
+    try:
+        ru = (await retrieval_summary(990012, days=30))["rule_usage"]
+
+        assert ru["surfaced"] == 1, "only the arm chose a rule"
+        assert ru["ambient"] == 4, "the four bulk deliveries are reported, not dropped"
+
+        # 1 agent pull over 1 RANKED surfacing. Were the ambient four folded in
+        # the ratio would read 0.2 — the arm looking four times worse for
+        # having a large resident set beside it.
+        assert ru["pull_through"] == 1.0
+
+        # Dead-weight detection needs both classes: a rule delivered by the
+        # preload and never opened is the case that reading matters most for.
+        assert ru["distinct_rules_surfaced"] == 5
+        assert ru["distinct_rules_pulled"] == 1
+    finally:
+        await cleanup()
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_ambient_alone_reports_no_ratio(_dispose_engine):
+    """A brand-new install loads rules every session and may never trigger the
+    arm. That must read as "no ranked surfacings yet", not as a precision of
+    zero — the reading that would make a working install look broken."""
+    from scribe.services.retrieval_telemetry import retrieval_summary
+
+    cleanup = await _rule_events(990013, [
+        (5201, "surfaced", "session_start"),
+        (5202, "surfaced", "session_start"),
+    ])
+    try:
+        ru = (await retrieval_summary(990013, days=30))["rule_usage"]
+        assert ru["ambient"] == 2
+        assert ru["surfaced"] == 0
+        assert ru["pull_through"] is None
+    finally:
+        await cleanup()

tests/test_services_rule_usage.py

@@ -99,12 +99,32 @@ def test_the_zero_readout_names_every_key():
     — a missing key here would read as a broken readout on almost every row."""
     assert rule_usage.empty_rule_usage() == {
         "surfaced_count": 0,
+        "ambient_count": 0,
         "pull_count": 0,
         "last_surfaced_at": None,
         "last_pulled_at": None,
     }
 
 
+def test_only_a_ranker_counts_as_ranked():
+    """The bulk surfaces are ambient; the write-path arm is the only chooser.
+
+    Inverted against the note twin on purpose (see the module docstring): the
+    RARE half is the one that gets named, so a bulk surface added later and
+    forgotten defaults to ambient — under-counting it — instead of defaulting
+    to ranked and padding the pull-through denominator with surfacings nobody
+    chose.
+    """
+    assert not rule_usage.is_ambient("write_path_rule")
+    for bulk in (
+        "session_start", "list_always_on_rules", "enter_project",
+        "get_project", "get_milestone", "start_planning", "get_task",
+    ):
+        assert rule_usage.is_ambient(bulk), bulk
+    # The safe default is the whole point of the inversion.
+    assert rule_usage.is_ambient("some_surface_invented_next_year")
+
+
 def test_the_model_serialises_the_fields_the_ratio_needs():
     ev = RuleUsageEvent(
         user_id=7, rule_id=156, event=SURFACED, source="write_path_rule"
@@ -161,6 +181,10 @@ async def test_usage_for_rules_aggregates_per_rule(_dispose_engine):
         # never have to tell "no events" from "not in the result" — and on any
         # existing install that is nearly every rule.
         assert out[6003] == rule_usage.empty_rule_usage()
+
+        # Nothing ambient in this fixture, so the ambient bucket stays empty
+        # rather than absorbing the ranked hits.
+        assert out[6001]["ambient_count"] == 0
     finally:
         async with async_session() as s:
             await s.execute(
@@ -178,3 +202,56 @@ async def test_usage_for_rules_on_an_empty_id_list_asks_the_database_nothing(
     empty topic is nothing. An unguarded `IN ()` is both a pointless round trip
     and, on some drivers, a syntax error."""
     assert await rule_usage.usage_for_rules([]) == {}
+
+
+@pytest.mark.integration
+@pytest.mark.asyncio
+async def test_a_preloaded_rule_does_not_read_as_a_ranked_surfacing(_dispose_engine):
+    """The split that makes the always-on set judgeable (#3473).
+
+    A resident rule is delivered every session by a surface that chose
+    nothing. Counting those as `surfaced_count` would rank the always-on set
+    as the most-surfaced rules in the install purely for being resident — and
+    the badge's "shown often, opened never → dead weight" reading, which is
+    the whole reason the counter exists, would then be exactly backwards.
+    """
+    from sqlalchemy import delete
+
+    from scribe.models import async_session
+    from scribe.models.rule_usage import RuleUsageEvent
+
+    async with async_session() as s:
+        s.add_all([
+            # Delivered by the preload three times over: ambient, all of it.
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="session_start"),
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="list_always_on_rules"),
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="enter_project"),
+            # ...and once by the arm, which DID choose it.
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=SURFACED, source="write_path_rule"),
+            # Opened once after a hint and once from the list: pulls are pulls
+            # however the rule was found, so both land in the one counter.
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=PULLED, source="mcp_get_rule"),
+            RuleUsageEvent(user_id=990021, rule_id=6101,
+                           event=PULLED, source="rest_rule"),
+        ])
+        await s.commit()
+    try:
+        out = await rule_usage.usage_for_rules([6101])
+
+        assert out[6101]["surfaced_count"] == 1, "only the arm chose this rule"
+        assert out[6101]["ambient_count"] == 3, "three bulk deliveries"
+        # Both PULLED rows accumulate — the loop ADDS rather than assigns, so a
+        # rule opened after a hint and again from the list reports two, not one.
+        assert out[6101]["pull_count"] == 2
+        assert out[6101]["last_pulled_at"] is not None
+    finally:
+        async with async_session() as s:
+            await s.execute(
+                delete(RuleUsageEvent).where(RuleUsageEvent.user_id == 990021)
+            )
+            await s.commit()