feat(moments): mounted rules arrive when their moment happens, through every door (milestone 458 step 4a, #4922)

A rule mounted on a moment now reaches the session when an act reaches
that moment, with no semantic match involved:

- run_moment_arm on the pipeline: a lookup, not a ranked search. Each
  line names the moment and the act that reached it ("at work.deliver,
  reached by `git push`"), so a misfire is visible where it lands and
  can be unmapped in-session. A repeat is cited, not quoted; fresh
  rules are recorded surfaced under source moment_rule with the moment
  in detail. No retrieval_logs row, as for the other lookups, so no
  latency is persisted for this arm.
- rule_scope: a rule's home clause, moved out of semantic_search_rules
  so the moment lookup scopes by the same one.
- rulebooks.rules_on_moments / mounted_moments.
- The plugin door: a catch-all PreToolUse hook (scribe_moment.sh). It
  keeps /moment-tools' answer on disk for five minutes, so a call to a
  tool that cannot reach a mounted rule sends nothing, and an install
  that has mounted nothing sends one request per window. It shares the
  rules ledger with the other arms and fails open silently.
- The MCP door: Scribe's own tools named by the shipped mappings carry
  moment_rules in their response, so a client without the plugin gets
  them too. The hook skips those tools. A guard pins the attach on
  every one.

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

src/scribe/mcp/tools/lessons.py

@@ -19,6 +19,7 @@ from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
 from scribe.mcp.tools import systems as systems_tools
 from scribe.services.note_usage import attach_usage, record_pulled
+from scribe.services.moment_delivery import attach_moment_rules
 
 
 # The payload shape lives in the service (`lesson_to_dict`), shared with the
@@ -245,7 +246,7 @@ async def create_lesson(
         await _offer_candidates(uid, data, what, when_to_apply, project_id)
     elif no_rule.strip():
         await _name_convergence(uid, data, note.id)
-    return data
+    return await attach_moment_rules(uid, "create_lesson", {"project_id": project_id}, data)
 
 
 async def _name_convergence(uid: int, data: dict, lesson_id: int) -> None:

src/scribe/mcp/tools/milestones.py

@@ -19,6 +19,7 @@ from scribe.services import task_logs as task_logs_svc
 from scribe.services import rulebooks as rulebooks_svc
 from scribe.services import trash as trash_svc
 from scribe.services.record_refs import refuse_guessed_ids
+from scribe.services.moment_delivery import attach_moment_rules
 
 
 async def list_milestones(project_id: int) -> dict:
@@ -132,7 +133,9 @@ async def create_milestone(
         body=body or None,
         status=status,
     )
-    return milestone.to_dict()
+    return await attach_moment_rules(
+        uid, "create_milestone", {"project_id": project_id}, milestone.to_dict(),
+    )
 
 
 async def update_milestone(
@@ -172,7 +175,10 @@ async def update_milestone(
     milestone = await milestones_svc.update_milestone(uid, milestone_id, **fields)
     if milestone is None:
         raise ValueError(f"milestone {milestone_id} not found")
-    return milestone.to_dict()
+    return await attach_moment_rules(
+        uid, "update_milestone", {"status": status, "project_id": project_id},
+        milestone.to_dict(),
+    )
 
 
 async def delete_milestone(milestone_id: int) -> dict:

src/scribe/mcp/tools/notes.py

@@ -23,6 +23,7 @@ from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
 from scribe.services.note_usage import record_pulled
 from scribe.services.record_refs import refuse_guessed_ids
+from scribe.services.moment_delivery import attach_moment_rules
 
 
 async def list_notes(
@@ -236,7 +237,7 @@ async def create_note(
     await systems_tools.attach_systems(uid, uid, data, note.id, project_id or None)
     await supersession_svc.attach_relations(uid, note.id, data, hint=True)
     data.update(dedup_svc.note_overlap_response(overlaps, "note"))
-    return data
+    return await attach_moment_rules(uid, "create_note", {"project_id": project_id}, data)
 
 
 async def update_note(

src/scribe/mcp/tools/processes.py

@@ -14,6 +14,7 @@ from scribe.services import notes as notes_svc
 from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
 from scribe.services.note_usage import record_pulled
+from scribe.services.moment_delivery import attach_moment_rules
 
 
 async def list_processes(
@@ -106,7 +107,7 @@ async def create_process(
     )
     if system_ids:
         await systems_svc.set_record_systems(uid, note.id, system_ids)
-    return note.to_dict()
+    return await attach_moment_rules(uid, "create_process", {}, note.to_dict())
 
 
 async def get_process(name_or_id: str, project_id: int = 0) -> dict:

src/scribe/mcp/tools/rulebooks.py

@@ -22,6 +22,7 @@ from scribe.services import trash as trash_svc
 from scribe.services.rule_usage import (
     record_rule_outcome, record_rule_pulled,
 )
+from scribe.services.moment_delivery import attach_moment_rules
 
 
 # ── Rulebook CRUD ───────────────────────────────────────────────────────
@@ -536,7 +537,7 @@ async def create_rule(
     )
     data = await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
     data.update(dedup_svc.overlap_response(overlaps, "rule"))
-    return data
+    return await attach_moment_rules(uid, "create_rule", {}, data)
 
 
 async def create_project_rule(
@@ -645,7 +646,7 @@ async def create_project_rule(
     )
     data = await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
     data.update(dedup_svc.overlap_response(overlaps, "rule"))
-    return data
+    return await attach_moment_rules(uid, "create_project_rule", {"project_id": project_id}, data)
 
 
 async def update_rule(
@@ -878,7 +879,7 @@ async def create_preference(
     )
     data = await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
     data.update(dedup_svc.overlap_response(overlaps, "preference"))
-    return data
+    return await attach_moment_rules(uid, "create_preference", {}, data)
 
 
 async def update_preference(

src/scribe/mcp/tools/snippets.py

@@ -17,6 +17,7 @@ from scribe.services import dedup as dedup_svc
 from scribe.services import snippets as snippets_svc
 from scribe.services.note_usage import attach_usage, record_pulled
 from scribe.services import systems as systems_svc
+from scribe.services.moment_delivery import attach_moment_rules
 
 
 async def list_snippets(
@@ -223,7 +224,7 @@ async def create_snippet(
     advice = snippets_svc.trigger_advice(when_to_use)
     if advice:
         data["trigger_advice"] = advice
-    return data
+    return await attach_moment_rules(uid, "create_snippet", {"project_id": project_id}, data)
 
 
 async def get_snippet(snippet_id: int, project_id: int = 0) -> dict:

src/scribe/mcp/tools/tasks.py

@@ -46,6 +46,7 @@ from scribe.services import trash as trash_svc
 from scribe.services.note_usage import record_pulled
 from scribe.services.record_refs import refuse_guessed_ids
 from scribe.services.text import elide
+from scribe.services.moment_delivery import attach_moment_rules
 
 
 # A work log entry is prose, often long — the discipline asks for what was
@@ -466,7 +467,9 @@ async def update_task(
         if prefs:
             data["reply_preferences"] = prefs
             data["report_back"] = REPORT_BACK_CUE + " " + REPLY_PREFERENCES_CUE
-    return data
+    return await attach_moment_rules(
+        uid, "update_task", {"status": status, "project_id": project_id}, data,
+    )
 
 
 async def add_task_log(task_id: int, content: str) -> dict:
@@ -734,6 +737,7 @@ async def start_planning(
     )
     if isinstance(result, dict):
         result.update(dedup_svc.batch_overlap_response(overlaps))
+        await attach_moment_rules(uid, "start_planning", {"project_id": project_id}, result)
     return result
 
 

src/scribe/routes/plugin.py

@@ -13,6 +13,7 @@ from quart import Blueprint, g, jsonify, request
 from scribe.auth import admin_required, get_current_user_id, login_required
 from scribe.config import Config
 from scribe.services import lesson_rules as lesson_rules_svc
+from scribe.services import moment_delivery as moment_delivery_svc
 from scribe.services import plugin_context as plugin_ctx_svc
 from scribe.services import repo_bindings as repo_bindings_svc
 from scribe.services import report_check as report_check_svc
@@ -241,6 +242,58 @@ async def pre_tool_rules():
     return jsonify(result)
 
 
+@plugin_bp.get("/moment-tools")
+@login_required
+async def moment_tools():
+    """The tools whose calls can deliver a mounted rule (milestone 458 step 4).
+
+    Read by the plugin's catch-all PreToolUse hook once per session window and
+    kept on disk, so a call that cannot reach a mounted rule costs no request.
+    `tools` are tool KEYS — bare, lowercased, no MCP server prefix — the same
+    keys the moment mappings are written in. Empty on an install that has
+    mounted nothing, which keeps that hook entirely off the wire.
+    """
+    return jsonify({"tools": await moment_delivery_svc.reachable_tools(g.user.id)})
+
+
+@plugin_bp.post("/moment")
+@login_required
+async def moment():
+    """The rules mounted on the moments this tool call reaches (milestone 458).
+
+    Body: the PreToolUse event as Claude Code delivers it — `tool_name` and
+    `tool_input` are read, nothing else. Sent whole because which input field
+    a mapping matches on is the mapping's business, not the hook's.
+
+    Query: `repo` / `project_id` for the scope, and the shared session ledger —
+    `exclude_rule_ids` (named this session: a repeat is cited, not quoted) and
+    `held_rule_ids` (opened) — exactly as /tool-rules takes them.
+
+    Returns `context` (the lines, each naming the moment and the action that
+    reached it, so a misfire is visible and can be unmapped in-session),
+    `rule_ids` (the FRESH ones, for the hook to append to the ledger) and
+    `moments` (the names reached). A lookup, not a ranked search: no
+    retrieval_logs row; each fresh rule is recorded surfaced with its moment.
+    """
+    event = await request.get_json(silent=True) or {}
+    tool = str(event.get("tool_name") or "").strip()
+    tool_input = event.get("tool_input")
+    if not tool:
+        return jsonify({"context": "", "rule_ids": [], "moments": []})
+    project_id, _repo, _unbound = await _project_scope()
+    reached, result = await moment_delivery_svc.deliver_for_act(
+        g.user.id, tool, tool_input if isinstance(tool_input, dict) else {},
+        project_id=project_id,
+        exclude=frozenset(_int_list(request.args.get("exclude_rule_ids"))),
+        held=frozenset(_int_list(request.args.get("held_rule_ids"))),
+    )
+    return jsonify({
+        "context": "\n".join(result.lines),
+        "rule_ids": result.rule_ids,
+        "moments": [hit["moment"] for hit in reached],
+    })
+
+
 @plugin_bp.get("/prior-art")
 @login_required
 @memoized_query_embeddings

src/scribe/services/embeddings.py

@@ -1528,8 +1528,8 @@ async def semantic_search_rules(
 
     Returns an empty list if the embedder is unavailable or on any error.
     """
-    from scribe.models.project import Project
-    from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
+    from scribe.models.rulebook import Rule
+    from scribe.services.rule_scope import joined_to_homes, rule_home
 
     # See the sibling search: stamped before anything can return (#3765).
     if report is not None:
@@ -1545,30 +1545,23 @@ async def semantic_search_rules(
     distance = RuleEmbedding.embedding.cosine_distance(query_vec)
 
     try:
-        # topic_id XOR project_id (migration 0059), so a rule matches exactly
-        # one arm of whichever clause applies. Inside the try: the access
-        # check reads the database too, and this function fails open.
-        global_rule = Rulebook.owner_user_id == user_id
-        if everywhere:
-            home = or_(global_rule, Project.user_id == user_id)
-        elif project_id and await can_read_project(user_id, project_id):
-            home = or_(global_rule, Rule.project_id == project_id)
-        else:
-            home = global_rule
+        # The home clause is shared with the moment lookup (rule_scope).
+        # Inside the try: the access check reads the database too, and this
+        # function fails open.
+        home = await rule_home(user_id, project_id, everywhere=everywhere)
 
         async with async_session() as session:
             rows = (await session.execute(
-                select(
-                    Rule,
-                    distance.label("distance"),
-                    RuleEmbedding.chunk_index,
-                    RuleEmbedding.chunk_text,
+                joined_to_homes(
+                    select(
+                        Rule,
+                        distance.label("distance"),
+                        RuleEmbedding.chunk_index,
+                        RuleEmbedding.chunk_text,
+                    )
+                    .select_from(RuleEmbedding)
+                    .join(Rule, RuleEmbedding.rule_id == Rule.id)
                 )
-                .select_from(RuleEmbedding)
-                .join(Rule, RuleEmbedding.rule_id == Rule.id)
-                .outerjoin(RulebookTopic, Rule.topic_id == RulebookTopic.id)
-                .outerjoin(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
-                .outerjoin(Project, Rule.project_id == Project.id)
                 .where(
                     Rule.deleted_at.is_(None),
                     # No threshold predicate — see the note above

src/scribe/services/moment_delivery.py

@@ -0,0 +1,119 @@
+"""Delivering the rules mounted on a moment, through every door (milestone 458 step 4).
+
+One function decides what an act reaches and what arrives with it; the doors
+differ only in how they carry the answer:
+
+- the plugin's catch-all PreToolUse hook → `/api/plugin/moment`, with the
+  session ledger, so a rule already named this session is cited not repeated;
+- Scribe's own MCP tools → `attach_moment_rules`, in the tool's response, so
+  a client without the plugin still gets `work.finish` when a task closes;
+- the Stop hook → `deliver_moments`, for the reply moments no tool call marks.
+
+The pipeline stage is `retrieval_pipeline.run_moment_arm`; this module only
+resolves the act to its moments and hands both doors the same result.
+"""
+from __future__ import annotations
+
+import logging
+
+from scribe.services import moment_actions
+from scribe.services import moments as catalog
+from scribe.services import retrieval_pipeline as rp
+
+logger = logging.getLogger(__name__)
+
+
+def _io() -> rp.MomentIO:
+    """Resolved at CALL time, so a test patching either module's name is seen."""
+    from scribe.services import rule_usage, rulebooks
+
+    return rp.MomentIO(
+        lookup=rulebooks.rules_on_moments,
+        record_rule_surfaced=rule_usage.record_rule_surfaced,
+    )
+
+
+async def reachable_tools(user_id: int) -> list[str]:
+    """The tools a call to which could deliver a mounted rule, as tool keys.
+
+    What the plugin's catch-all hook reads once per session window, so the
+    calls that cannot reach a mounted rule — most of them, and every one on
+    an install that has mounted nothing — never leave the machine. An action
+    counts only when its moment carries a mount; the skill loader counts when
+    any `skill.<name>` moment does.
+    """
+    from scribe.services import rulebooks
+
+    mounted = await rulebooks.mounted_moments(user_id)
+    if not mounted:
+        return []
+    mappings = await moment_actions.list_mappings(user_id)
+    keys = {
+        moment_actions.tool_key(action.tool)
+        for action, _via in moment_actions.effective_actions(mappings)
+        if action.moment in mounted
+    }
+    if any(name.startswith(catalog.SKILL_PREFIX) for name in mounted):
+        keys.add(moment_actions.SKILL_TOOL)
+    return sorted(keys)
+
+
+async def deliver_moments(
+    user_id: int, reached: list[dict], *, project_id: int | None = None,
+    exclude: frozenset[int] = frozenset(), held: frozenset[int] = frozenset(),
+) -> rp.RuleResult:
+    """The mounted rules for moments already known to have happened."""
+    return await rp.run_moment_arm(
+        reached,
+        rp.RuleMoment(user_id=user_id, query="", project_id=project_id,
+                      exclude=exclude, held=held),
+        io=_io(),
+    )
+
+
+async def deliver_for_act(
+    user_id: int, tool: str, tool_input: dict | None, *,
+    project_id: int | None = None,
+    exclude: frozenset[int] = frozenset(), held: frozenset[int] = frozenset(),
+) -> tuple[list[dict], rp.RuleResult]:
+    """Which moments this act reached, and the mounted rules they deliver."""
+    reached = await moment_actions.moments_for(user_id, tool, tool_input or {})
+    if not reached:
+        return [], rp.RuleResult()
+    result = await deliver_moments(
+        user_id, reached, project_id=project_id, exclude=exclude, held=held,
+    )
+    return reached, result
+
+
+async def attach_moment_rules(
+    user_id: int, tool: str, arguments: dict | None, data: dict,
+) -> dict:
+    """Carry the rules mounted on this tool call's moments in its own response.
+
+    The MCP door (#4286's attach shape): an in-band decoration on a payload
+    already being returned, fail-open, and absent when there is nothing to
+    say. The project is the record's own when it has one, else the caller's
+    argument — a project's mounted rules belong to work in that project.
+
+    No session ledger reaches this door, so a mounted rule arrives every time
+    its moment happens here. That is the right failure for the moments these
+    tools mark: closing a task is exactly when a rule about finishing applies,
+    however many tasks were closed before it.
+    """
+    if not isinstance(data, dict):
+        return data
+    try:
+        project_id = data.get("project_id") or (arguments or {}).get("project_id") or None
+        _reached, result = await deliver_for_act(
+            user_id, tool, arguments or {}, project_id=project_id,
+        )
+        if result.lines:
+            data["moment_rules"] = {
+                "lines": result.lines,
+                "rule_ids": result.shown_rule_ids,
+                "open_with": "get_rule(id)",
+            }
+    except Exception:  # noqa: BLE001 - a decoration never breaks the payload
+        logger.debug("moment rules for %s could not be attached", tool, exc_info=True)
+    return data

src/scribe/services/retrieval_pipeline.py

@@ -662,3 +662,100 @@ async def run_rule_arm(
     except Exception:  # noqa: BLE001 - a recall aid never breaks its act
         logger.debug("%s arm failed", arm.source, exc_info=True)
         return RuleResult()
+
+
+# ── The moment arm (milestone 458) ───────────────────────────────────────
+#
+# A LOOKUP beside the ranked arms, not one of them. A rule mounted on a moment
+# arrives because somebody said it belongs there, not because the work's words
+# resemble it — which is the whole point: a rule about WHEN work is finished
+# never scores against what is said while finishing it. So there is no score,
+# no floor and no band, and it writes no retrieval_logs row (the rulings arm's
+# precedent, #4769): every score-shaped warning would misread a lookup. What it
+# records is the surfacing, with the moment beside each row, so a mount's
+# pull-through can be read per moment.
+#
+# It shares everything else with the ranked arms: the line renderer, the
+# session ledger (a repeat renders as a citation, never hidden — #3750) and
+# the fresh-only surfacing rows (#3752).
+
+MOMENT_RULE_SOURCE = "moment_rule"
+
+# A cap against a misconfiguration, not a ranking budget. Mounts are
+# deliberate, so the ordinary case is a handful; a rulebook that mounted
+# dozens of rules on one moment would otherwise bury the act it annotates.
+# The remedy for a moment that hits this is fewer mounts (unmount through
+# update_rule), which is why it is not a tuning dial.
+MOMENT_RULE_LIMIT = 8
+
+
+@dataclass(frozen=True)
+class MomentIO:
+    """The mounted-rule lookup and the recorder the moment arm reports to."""
+
+    lookup: Callable[..., Any]
+    """`rulebooks.rules_on_moments`, or a stand-in with its signature."""
+
+    record_rule_surfaced: Callable[..., Any]
+
+
+def _reached_by(hit: dict) -> str:
+    """How the line names what reached a moment: the action, as mapped."""
+    return f"`{hit.get('match') or hit.get('tool') or '?'}`"
+
+
+async def run_moment_arm(
+    reached: list[dict], moment: RuleMoment, *, io: MomentIO,
+    limit: int = MOMENT_RULE_LIMIT,
+) -> RuleResult:
+    """Deliver the rules mounted on the moments this act reached.
+
+    `reached` is `moment_actions.resolve`'s answer — each moment with the
+    action that reached it — and every line says both ("at work.deliver,
+    reached by `git push`"), so a moment that fired on the wrong act is
+    visible in the line itself and can be corrected in the session with
+    `unmap_action` (step 2's ruling). Fresh rules come first and carry their
+    trigger; a rule the session was already shown is cited, not repeated in
+    full. Fails open to an empty result.
+    """
+    try:
+        names = [hit["moment"] for hit in reached]
+        if not names:
+            return RuleResult()
+        pairs = await io.lookup(moment.user_id, names, moment.project_id or None)
+        if not pairs:
+            return RuleResult()
+        by_moment = {hit["moment"]: hit for hit in reached}
+        # Stable: catalog order within each half, fresh half first.
+        shown = sorted(pairs, key=lambda pair: pair[0].id in moment.exclude)[:limit]
+
+        lines = []
+        for rule, at in shown:
+            seen = rule.id in moment.exclude
+            lines.append(_rule_hint_line(
+                rule,
+                where=f"at {at}, reached by {_reached_by(by_moment.get(at, {}))}",
+                seen=seen, held=rule.id in moment.held,
+                # A repeat is cited rather than quoted: the session was told
+                # which moment brought it the first time.
+                compact=seen,
+            ))
+        fresh = [(rule, at) for rule, at in shown if rule.id not in moment.exclude]
+        rule_ids = [rule.id for rule, _at in fresh]
+        if rule_ids:
+            try:
+                io.record_rule_surfaced(
+                    user_id=moment.user_id, rule_ids=rule_ids,
+                    source=MOMENT_RULE_SOURCE,
+                    detail={rule.id: at for rule, at in fresh},
+                )
+            except Exception:  # noqa: BLE001 - observation never breaks the observed
+                _telemetry_failed(MOMENT_RULE_SOURCE)
+        return RuleResult(
+            lines=lines, rule_ids=rule_ids,
+            shown_rule_ids=[rule.id for rule, _at in shown],
+            shown=[(None, rule) for rule, _at in shown],
+        )
+    except Exception:  # noqa: BLE001 - a recall aid never breaks its act
+        logger.debug("%s arm failed", MOMENT_RULE_SOURCE, exc_info=True)
+        return RuleResult()

src/scribe/services/retrieval_registry.py

@@ -163,6 +163,15 @@ POINTS: dict[str, Point] = dict([
        "the fixed question asked when a task finishes: how should this report read",
        fixed_query=True),
 
+    # The moment arm (milestone 458). A LOOKUP, not a ranker: a rule mounted on
+    # a moment arrives when that moment happens, with no score, so it writes no
+    # retrieval_logs row and no score-shaped warning can apply. Its rows are in
+    # `rule_usage_events`, each carrying the moment in `detail`.
+    _p("moment_rule", UNBIDDEN,
+       "rules mounted on a moment of work, delivered when that moment happens",
+       expects_traffic=False,
+       quiet_because="speaks only once some rule is mounted on a moment; an "
+                     "install where none is mounted is correctly silent here"),
     # The rulings arm (milestone 444). A LOOKUP, not a ranker: a path falls
     # under a System's patterns or it does not, so nothing here writes to
     # retrieval_logs and no score-shaped warning can apply. Its rows are in

src/scribe/services/rule_scope.py

@@ -0,0 +1,55 @@
+"""Which rules a caller may be shown — a rule's HOME, stated once (milestone 414).
+
+A rule lives in a rulebook topic — GLOBAL, it applies wherever its owner
+works — or on one project, where it applies to that project and nowhere else.
+Every read that hands rules to a session answers the same question about that
+home, so it is answered here and nowhere else:
+
+- `project_id` unset: global rules only. A caller that does not say gets the
+  safe failure, which is surfacing less rather than another project's rules.
+- `project_id=N`: global rules plus project N's own, and N's only when the
+  caller can read that project (access.can_read_project, so a shared project's
+  rules reach its collaborators too).
+- `everywhere=True`: every rule the caller owns, in any home — the explicit
+  whole-rulebook question.
+
+It was written inline in `semantic_search_rules` until the moment lookup
+(milestone 458) needed the same answer. Two copies of a scope clause is how
+one of them starts surfacing another project's rules, so it moved here first.
+"""
+from __future__ import annotations
+
+from sqlalchemy import or_
+
+from scribe.services.access import can_read_project
+
+
+async def rule_home(user_id: int, project_id: int | None = None, *, everywhere: bool = False):
+    """The WHERE clause for the rules this caller may be shown here.
+
+    Needs the statement joined through `joined_to_homes`, which outer-joins
+    the three tables the clause reads. A rule is in a topic XOR on a project
+    (migration 0059), so it matches exactly one arm of whichever clause
+    applies.
+    """
+    from scribe.models.project import Project
+    from scribe.models.rulebook import Rule, Rulebook
+
+    global_rule = Rulebook.owner_user_id == user_id
+    if everywhere:
+        return or_(global_rule, Project.user_id == user_id)
+    if project_id and await can_read_project(user_id, project_id):
+        return or_(global_rule, Rule.project_id == project_id)
+    return global_rule
+
+
+def joined_to_homes(stmt):
+    """Outer-join a statement over `Rule` to the tables `rule_home` reads."""
+    from scribe.models.project import Project
+    from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
+
+    return (
+        stmt.outerjoin(RulebookTopic, Rule.topic_id == RulebookTopic.id)
+        .outerjoin(Rulebook, RulebookTopic.rulebook_id == Rulebook.id)
+        .outerjoin(Project, Rule.project_id == Project.id)
+    )

src/scribe/services/rule_usage.py

@@ -101,6 +101,11 @@ logger = logging.getLogger(__name__)
 # Add a source here only when a ranker picked it.
 RANKED_SOURCES = (
     "write_path_rule", "pre_tool_rule", "prompt_rule",
+    # A rule mounted on a moment (milestone 458). Not a ranker's pick, but
+    # not bulk either: somebody decided this rule applies at this moment, and
+    # that is exactly the claim a pull can confirm or refute. Its
+    # pull-through is the evidence for whether a mount earns its line.
+    "moment_rule",
     # A reserved slot is a ranker's choice twice over — it ran a query AND
     # decided a kind was worth guaranteeing a place. Left out, its line would
     # be counted as bulk delivery and drop out of the denominator, so the one
@@ -154,7 +159,8 @@ def _schedule(rows: list[dict]) -> None:
 
 
 def record_rule_surfaced(
-    *, user_id: int | None, rule_ids: list[int] | set[int], source: str
+    *, user_id: int | None, rule_ids: list[int] | set[int], source: str,
+    detail: dict[int, str] | None = None,
 ) -> None:
     """Fire-and-forget: record that these rules were shown to the agent.
 
@@ -173,6 +179,10 @@ def record_rule_surfaced(
     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.
+
+    `detail` maps a rule id to what to record beside its row — the moment a
+    mounted rule arrived at (milestone 458), so a mount's pull-through can be
+    read per moment rather than only per arm.
     """
     try:
         rows = [
@@ -181,6 +191,7 @@ def record_rule_surfaced(
                 "rule_id": int(rid),
                 "event": SURFACED,
                 "source": source,
+                **({"detail": detail[rid]} if detail and rid in detail else {}),
             }
             for rid in rule_ids
         ]

src/scribe/services/rulebooks.py

@@ -1160,6 +1160,70 @@ async def list_rule_moments(rule_ids: list[int]) -> dict[int, list[str]]:
     return out
 
 
+async def rules_on_moments(
+    user_id: int, moments: list[str], project_id: int | None = None,
+) -> list[tuple[Rule, str]]:
+    """The rules mounted on any of these moments that this caller may be shown.
+
+    The delivery read (milestone 458 step 4), and a LOOKUP: no score, no bar.
+    Scoped by the same home clause the semantic search uses (rule_scope), so
+    a mount never delivers another project's rule. Each rule appears once,
+    under the first of `moments` it is mounted on — the caller passes them in
+    catalog order, so that is the most specific moment the work reached.
+    """
+    from scribe.models.rulebook import rule_moments as rule_moments_t
+    from scribe.services.rule_scope import joined_to_homes, rule_home
+
+    if not moments:
+        return []
+    home = await rule_home(user_id, project_id)
+    async with async_session() as session:
+        rows = (await session.execute(
+            joined_to_homes(
+                select(Rule, rule_moments_t.c.moment)
+                .select_from(rule_moments_t)
+                .join(Rule, Rule.id == rule_moments_t.c.rule_id)
+            )
+            .where(
+                rule_moments_t.c.moment.in_(list(moments)),
+                Rule.deleted_at.is_(None),
+                home,
+            )
+            .order_by(Rule.id)
+        )).all()
+    rank = {m: i for i, m in enumerate(moments)}
+    first: dict[int, tuple[Rule, str]] = {}
+    for rule, moment in rows:
+        held = first.get(rule.id)
+        if held is None or rank[moment] < rank[held[1]]:
+            first[rule.id] = (rule, moment)
+    return sorted(first.values(), key=lambda pair: (rank[pair[1]], pair[0].id))
+
+
+async def mounted_moments(user_id: int) -> set[str]:
+    """Every moment at least one of this caller's live rules is mounted on.
+
+    In ANY home (rule_scope's `everywhere`): this answers "can a call reach
+    anything at all", which is what lets the plugin stay off the wire for a
+    tool whose moments carry nothing. A superset is the safe direction — the
+    delivery read still scopes by project.
+    """
+    from scribe.models.rulebook import rule_moments as rule_moments_t
+    from scribe.services.rule_scope import joined_to_homes, rule_home
+
+    home = await rule_home(user_id, everywhere=True)
+    async with async_session() as session:
+        rows = (await session.execute(
+            joined_to_homes(
+                select(rule_moments_t.c.moment).distinct()
+                .select_from(rule_moments_t)
+                .join(Rule, Rule.id == rule_moments_t.c.rule_id)
+            )
+            .where(Rule.deleted_at.is_(None), home)
+        )).scalars().all()
+    return set(rows)
+
+
 async def list_rule_systems(rule_ids: list[int]) -> dict[int, list[dict]]:
     """The canon tags for a batch of rules, keyed by rule id.