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>

plugin/.claude-plugin/plugin.json

@@ -1,7 +1,7 @@
 {
   "name": "scribe",
   "description": "Scribe for Claude Code: connects the scribe MCP server, adds the hooks that deliver live project state and relevant records at the right moment, ships the shared client-neutral Scribe skills (using-scribe, writing-plans, reporting-back, systematic-debugging, verification, brainstorming, reusing-code, shape-accounting), and syncs your saved Scribe Processes as skills (/scribe:sync).",
-  "version": "2026.10.04.0125",
+  "version": "2026.10.05.1621",
   "author": {
     "name": "Bryan Van Deusen"
   },

plugin/PACKAGING.md

@@ -39,6 +39,7 @@ another one means adding files, not moving or rewriting any.
 | `hooks/scribe_prior_art.sh` | PreToolUse on editor writes: `GET /api/plugin/prior-art`. |
 | `hooks/scribe_after_write.sh` | PostToolUse on shell commands: the same check for code written through the shell. |
 | `hooks/scribe_tool_rules.sh` | PreToolUse on shell commands: `GET /api/plugin/tool-rules`. |
+| `hooks/scribe_moment.sh` | PreToolUse on every tool: the rules mounted on the moments the call reaches, `POST /api/plugin/moment`; skips tools `GET /api/plugin/moment-tools` says reach nothing mounted, and Scribe's own (their responses carry `moment_rules`). |
 | `hooks/scribe_report_check.sh` | Stop: when the turn closed a task, checks the reply for the completion sections and reports to `GET /api/plugin/report-check`; blocks once, with the reason the server returns. |
 | `hooks/scribe_shape_check.sh` | Stop: sends the definitions the turn wrote (the write hooks' `<sid>.written.ids` ledger) to `GET /api/plugin/shape-check`; blocks once, with the reason the server returns, so the agent judges what it built. |
 | `hooks/scribe_sync_processes.sh` + `commands/sync.md` | `GET /api/plugin/processes` → `~/.claude/skills/scribe-proc-*` stubs; `/scribe:sync` on demand. |

plugin/hooks/hooks.json

@@ -42,6 +42,15 @@
             "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_tool_rules.sh\""
           }
         ]
+      },
+      {
+        "matcher": "*",
+        "hooks": [
+          {
+            "type": "command",
+            "command": "bash \"${CLAUDE_PLUGIN_ROOT}/hooks/scribe_moment.sh\""
+          }
+        ]
       }
     ],
     "PostToolUse": [

plugin/hooks/scribe_moment.sh

@@ -0,0 +1,118 @@
+#!/usr/bin/env bash
+# Scribe — PreToolUse moment arm: rules MOUNTED on a moment arrive when the
+# moment happens (milestone 458 step 4).
+#
+# The other PreToolUse arms search: they turn the act into a query and hope a
+# rule's words resemble it. A mount needs no resemblance. The operator said
+# "this rule applies when work is delivered", the install maps `git push` onto
+# work.deliver, and so `git push` brings the rule — whatever the command's
+# words happen to be. The server resolves the call to its moments and looks up
+# what is mounted there; this hook only carries the event and the answer.
+#
+# REGISTERED ON EVERY TOOL, KEPT OFF THE WIRE FOR MOST OF THEM. Which calls can
+# reach a mounted rule depends on this install's mappings and mounts, so the
+# matcher cannot say. Instead the hook keeps the server's answer to "which
+# tools can?" (`/moment-tools`) on disk for a few minutes, and a call to any
+# other tool costs one file read. An install that has mounted nothing gets an
+# empty list and never sends a moment request at all.
+#
+# SCRIBE'S OWN TOOLS ARE SKIPPED. Closing a task or recording a lesson reaches
+# its moment through the tool's own response (`moment_rules`), which works for
+# a client without this plugin too; delivering it here as well would say it
+# twice.
+#
+# SILENT ON OUTAGE, like scribe_tool_rules.sh and for its reason: this fires
+# before many calls, and an outage line before each is noise. A failed list
+# fetch is cached as an empty list, so a down instance costs one timeout per
+# window rather than one per call.
+#
+# Env:
+#   SCRIBE_URL / SCRIBE_TOKEN override for the settings.json dogfooding path.
+
+command -v curl >/dev/null 2>&1 || exit 0
+
+# shellcheck source=plugin/hooks/scribe_defs.sh
+. "$(dirname "${BASH_SOURCE[0]}")/scribe_defs.sh"
+
+event=$(cat 2>/dev/null || true)
+event_flat=$(printf '%s' "$event" | scribe_json_flat)
+tool_name=$(scribe_json_pick "$event_flat" '.tool_name')
+session_id=$(scribe_json_pick "$event_flat" '.session_id')
+event_cwd=$(scribe_json_pick "$event_flat" '.cwd')
+
+[ -n "$tool_name" ] && [ -n "$session_id" ] || exit 0
+
+case "$tool_name" in
+  mcp__*scribe*__*) exit 0 ;;
+esac
+
+scribe_config || exit 0
+
+# The tool's KEY, as the server writes mappings: no MCP server prefix, lowercase.
+tool_key=${tool_name##*__}
+tool_key=$(printf '%s' "$tool_key" | tr '[:upper:]' '[:lower:]')
+
+state_dir="${TMPDIR:-/tmp}/scribe-priorart"
+mkdir -p "$state_dir" 2>/dev/null || true
+safe_sid=$(printf '%s' "$session_id" | tr -c 'A-Za-z0-9._-' '_')
+
+# ── Which tools can reach a mounted rule: cached, first line a timestamp ──
+#
+# Five minutes. A mount or a mapping made in this session reaches the hook
+# within that window; the MCP door delivers Scribe's own moments at once.
+# Not a `.ids` file, so a compaction does not sweep it: it describes the
+# install, not what this context holds.
+tools_file="$state_dir/${safe_sid}.moment.tools"
+now=$(date +%s 2>/dev/null) || now=0
+stamp=""
+[ -f "$tools_file" ] && stamp=$(head -n 1 "$tools_file" 2>/dev/null)
+case "$stamp" in ''|*[!0-9]*) stamp=0 ;; esac
+if [ "$now" -eq 0 ] || [ $((now - stamp)) -gt 300 ]; then
+  listed=$(curl -fsS --max-time 3 \
+    -H "Authorization: Bearer ${token}" \
+    "${url%/}/api/plugin/moment-tools" 2>/dev/null) || listed=""
+  {
+    printf '%s\n' "$now"
+    scribe_json_list "$(printf '%s' "$listed" | scribe_json_flat)" '.tools'
+  } > "$tools_file" 2>/dev/null || true
+fi
+tail -n +2 "$tools_file" 2>/dev/null | grep -Fqx -- "$tool_key" || exit 0
+
+# ── The moment request ────────────────────────────────────────────────────
+
+repo_q=""
+lookup_dir=${event_cwd:-${CLAUDE_PROJECT_DIR:-$PWD}}
+scope=$(scribe_scope_query "$lookup_dir")
+[ -n "$scope" ] && repo_q="&${scope}"
+
+# The shared session ledger, as scribe_tool_rules.sh reads and writes it: a
+# rule this session was already shown is cited, not quoted again.
+rulefile="$state_dir/${safe_sid}.rules.ids"
+ledger_q=""
+rule_seen=$(scribe_rules_live "$rulefile")
+[ -n "$rule_seen" ] && ledger_q="&exclude_rule_ids=${rule_seen}"
+ledger_q="${ledger_q}$(scribe_held_query "$state_dir/${safe_sid}.opened.ids")"
+
+# The event whole, since which field a mapping matches is the mapping's
+# business. A write of a very large file is reduced to its name: the moments a
+# write reaches never depend on its content.
+if [ "${#event}" -gt 65536 ]; then
+  esc=$(printf '%s' "$tool_name" | scribe_json_escape) || exit 0
+  event="{\"tool_name\":\"${esc}\",\"tool_input\":{}}"
+fi
+
+query="${repo_q}${ledger_q}"
+query=${query#&}
+body=$(printf '%s' "$event" | curl -fsS --max-time 3 \
+  -H "Authorization: Bearer ${token}" \
+  -H "Content-Type: application/json" \
+  --data-binary @- \
+  "${url%/}/api/plugin/moment${query:+?$query}" 2>/dev/null) || exit 0
+
+body_flat=$(printf '%s' "$body" | scribe_json_flat)
+scribe_json_list "$body_flat" '.rule_ids' | scribe_rules_append "$rulefile"
+
+context=$(scribe_json_pick "$body_flat" '.context')
+[ -n "$context" ] || exit 0
+scribe_json_out PreToolUse "$context"
+exit 0

scripts/check_plugin.py

@@ -341,6 +341,14 @@ SMOKE_EVENTS: dict[str, str] = {
         {"session_id": "smoke", "cwd": ".", "tool_name": "Bash",
          "tool_input": {"command": "curl -s https://example.invalid/api/v1/runs"}}
     ),
+    # The moment arm (milestone 458). A command the shipped mappings send to
+    # work.deliver, so a configured run reaches the moment-tools fetch and the
+    # moment request; with no instance both fail and it must stay SILENT,
+    # since it fires before every call that can reach a mounted rule.
+    "scribe_moment.sh": json.dumps(
+        {"session_id": "smoke", "cwd": ".", "tool_name": "Bash",
+         "tool_input": {"command": "git push origin dev"}}
+    ),
     "scribe_sync_processes.sh": json.dumps({"source": "startup"}),
     "scribe_session_context.sh": json.dumps({"source": "startup"}),
     # The after-write hook (#2901) diffs the working tree; on CI's clean

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
-    from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
+    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
+        # The home clause is shared with the moment lookup (rule_scope).
-        # one arm of whichever clause applies. Inside the try: the access
+        # Inside the try: the access check reads the database too, and this
-        # check reads the database too, and this function fails open.
+        # function fails open.
-        global_rule = Rulebook.owner_user_id == user_id
+        home = await rule_home(user_id, project_id, everywhere=everywhere)
-        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
 
         async with async_session() as session:
             rows = (await session.execute(
-                select(
+                joined_to_homes(
-                    Rule,
+                    select(
-                    distance.label("distance"),
+                        Rule,
-                    RuleEmbedding.chunk_index,
+                        distance.label("distance"),
-                    RuleEmbedding.chunk_text,
+                        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.
 

tests/conftest.py

@@ -215,6 +215,26 @@ def _no_lesson_rule_links(request):
         yield
 
 
+@pytest.fixture(autouse=True)
+def _no_moment_delivery(request):
+    """Stub the moment lookup the mapped MCP tools attach (milestone 458).
+
+    update_task, the create_* record tools and the milestone tools now hand
+    back the rules mounted on the moment they mark, and finding those is two
+    database reads. Patched BENEATH the attach, at `deliver_for_act`, because
+    the tools bind `attach_moment_rules` by name at import time. Skipped for
+    integration tests; tests/test_moment_delivery.py binds the real one.
+    """
+    if request.node.get_closest_marker("integration"):
+        yield
+        return
+    from scribe.services.retrieval_pipeline import RuleResult
+
+    with patch("scribe.services.moment_delivery.deliver_for_act",
+               AsyncMock(return_value=([], RuleResult()))):
+        yield
+
+
 @pytest.fixture(autouse=True)
 def _no_rule_arm():
     """Stub the write-path hint's standing-RULES arm (milestone 307).

tests/helpers.py

@@ -349,7 +349,8 @@ def design_token_stub(name, value_by_mode, group_name=None, purpose=None,
 
 
 @contextmanager
-def http_sink(reply: bytes = b'{"context":"","note_ids":[]}'):
+def http_sink(reply: bytes = b'{"context":"","note_ids":[]}',
+              by_path: dict[str, bytes] | None = None):
     """A throwaway local HTTP listener for hook end-to-end tests: yields
     ``(port, seen)`` where ``seen`` collects every GET's parsed query string
     (one dict per request, in order). Lets the shell be tested end to end —
@@ -357,6 +358,11 @@ def http_sink(reply: bytes = b'{"context":"","note_ids":[]}'):
 
     Three test modules each carried their own ``_Sink`` handler before #2904
     consolidated them here; pass ``reply`` for the body the hook should see.
+
+    ``by_path`` is for a hook that calls more than one route: each request is
+    answered with the reply for its path (``reply`` when none is listed), and
+    every entry in ``seen`` then also carries ``_path`` and, for a POST,
+    ``_body`` — the raw request body, decoded.
     """
     import http.server
     import threading
@@ -365,12 +371,27 @@ def http_sink(reply: bytes = b'{"context":"","note_ids":[]}'):
     seen: list[dict] = []
 
     class _Sink(http.server.BaseHTTPRequestHandler):
-        def do_GET(self):
+        def _answer(self, body: str | None):
-            seen.append(urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query))
+            parsed = urllib.parse.urlparse(self.path)
+            entry = urllib.parse.parse_qs(parsed.query)
+            out = reply
+            if by_path is not None:
+                entry["_path"] = parsed.path
+                out = by_path.get(parsed.path, reply)
+                if body is not None:
+                    entry["_body"] = body
+            seen.append(entry)
             self.send_response(200)
             self.send_header("Content-Type", "application/json")
             self.end_headers()
-            self.wfile.write(reply)
+            self.wfile.write(out)
+
+        def do_GET(self):
+            self._answer(None)
+
+        def do_POST(self):
+            size = int(self.headers.get("Content-Length") or 0)
+            self._answer(self.rfile.read(size).decode("utf-8", "replace"))
 
         def log_message(self, *a):
             pass

tests/test_integration_rule_moments.py

@@ -6,16 +6,20 @@ claim depends on:
 - only the owner can mount;
 - the rule's detail reads its mounts back;
 - a backup carries them, attached to the RESTORED rule's new id rather than
-  the old number.
+  the old number;
+- the delivery lookup (step 4) answers by home: a project's mounted rule
+  reaches only work bound to that project, and nobody else's mount arrives.
 """
 import pytest
 import pytest_asyncio
 from sqlalchemy import select
 
 from scribe.models import async_session
+from scribe.models.project import Project
 from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
 from scribe.models.user import User
 from scribe.services import backup
+from scribe.services import projects as projects_svc
 from scribe.services import rulebooks as rulebooks_svc
 from tests.helpers import ensure_user
 
@@ -155,3 +159,43 @@ async def test_a_backup_carries_the_mounts_to_the_restored_rule(world):
     assert (await rulebooks_svc.list_rule_moments([restored_rule.id]))[restored_rule.id] == [
         "work.finish", "reply.report",
     ]
+
+
+async def _purge_projects(username: str) -> None:
+    async with async_session() as s:
+        for user in (await s.execute(
+            select(User).where(User.username == username)
+        )).scalars().all():
+            for project in (await s.execute(
+                select(Project).where(Project.user_id == user.id)
+            )).scalars().all():
+                await s.delete(project)
+        await s.commit()
+
+
+async def test_the_delivery_lookup_answers_by_home(world):
+    uid, sid, rule = world["uid"], world["sid"], world["rule"]
+    await _purge_projects(OWNER_USERNAME)
+    project = await projects_svc.create_project(uid, "Moment delivery fixture")
+    local = await rulebooks_svc.create_project_rule(
+        project.id, uid, "Ship notes with the release",
+        "A release carries its notes.", when_to_apply="publishing a release",
+    )
+    await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.finish", "reply.report"])
+    await rulebooks_svc.set_rule_moments(local.id, uid, ["work.deliver"])
+
+    # Unbound: the global rule, under the FIRST of the given moments it is on.
+    got = await rulebooks_svc.rules_on_moments(uid, ["reply.report", "work.finish"])
+    assert [(r.id, m) for r, m in got] == [(rule.id, "reply.report")]
+    assert await rulebooks_svc.rules_on_moments(uid, ["work.deliver"]) == []
+
+    # Bound to the project: its own rule arrives too.
+    got = await rulebooks_svc.rules_on_moments(uid, ["work.deliver"], project.id)
+    assert [(r.id, m) for r, m in got] == [(local.id, "work.deliver")]
+
+    # Somebody else's mount never arrives — not even naming their project.
+    assert await rulebooks_svc.rules_on_moments(sid, ["work.finish", "work.deliver"], project.id) == []
+
+    # The plugin's "can anything arrive" answer spans every home.
+    assert await rulebooks_svc.mounted_moments(uid) >= {"work.finish", "reply.report", "work.deliver"}
+    assert not (await rulebooks_svc.mounted_moments(sid)) & {"work.finish", "work.deliver"}

tests/test_moment_delivery.py

@@ -0,0 +1,373 @@
+"""Delivering mounted rules when their moment happens (milestone 458 step 4).
+
+The database is stubbed: the lookup's scoping is pinned against Postgres in
+tests/test_integration_rule_moments.py. These pin what is decided above it:
+- the line names the moment AND the act that reached it, so a misfire is
+  visible where it lands and can be unmapped in the session;
+- a rule the session was already shown is cited, not repeated, and only the
+  fresh ones are recorded, each with its moment;
+- every door fails open;
+- the plugin is told which tools can reach anything, and nothing more;
+- every Scribe tool the shipped mappings name carries the attach, and the
+  hook and the route agree on every name between them.
+"""
+from __future__ import annotations
+
+import ast
+import inspect
+import json
+import os
+import subprocess
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, MagicMock, patch
+
+from quart import Quart, g
+
+from scribe.services import moment_actions
+from scribe.services import moment_delivery as md
+from scribe.services import retrieval_pipeline as rp
+from scribe.services import rule_usage, rulebooks
+from tests.helpers import fake_rule, http_sink, need_tools
+
+# Bound before conftest's autouse stub replaces the module attribute.
+_REAL_DELIVER = md.deliver_for_act
+
+ROOT = Path(__file__).resolve().parents[1]
+HOOK = ROOT / "plugin" / "hooks" / "scribe_moment.sh"
+
+
+def _rule(rid, title="Done means delivered and checked"):
+    return fake_rule(id=rid, title=title, kind="rule",
+                     when_to_apply="about to say a piece of work is finished")
+
+
+def _moment(**kw):
+    return rp.RuleMoment(user_id=1, query="", project_id=kw.pop("project_id", 2), **kw)
+
+
+PUSH = [{"moment": "work.deliver", "tool": "Bash", "match": "git push", "via": "default"}]
+
+
+# ── the arm ─────────────────────────────────────────────────────────────
+
+
+async def test_a_line_names_the_moment_and_the_act_that_reached_it():
+    lookup = AsyncMock(return_value=[(_rule(5), "work.deliver")])
+    surfaced = MagicMock()
+    result = await rp.run_moment_arm(
+        PUSH, _moment(), io=rp.MomentIO(lookup=lookup, record_rule_surfaced=surfaced),
+    )
+    assert len(result.lines) == 1
+    assert "at work.deliver, reached by `git push`" in result.lines[0]
+    assert "get_rule(5)" in result.lines[0]
+    assert result.rule_ids == [5]
+    lookup.assert_awaited_once_with(1, ["work.deliver"], 2)
+    surfaced.assert_called_once()
+    kw = surfaced.call_args.kwargs
+    assert kw["source"] == rp.MOMENT_RULE_SOURCE
+    assert kw["rule_ids"] == [5]
+    assert kw["detail"] == {5: "work.deliver"}
+
+
+async def test_a_rule_already_shown_is_cited_after_the_fresh_ones_and_not_recorded():
+    lookup = AsyncMock(return_value=[(_rule(5), "work.deliver"), (_rule(9, "Other"), "work.deliver")])
+    surfaced = MagicMock()
+    result = await rp.run_moment_arm(
+        PUSH, _moment(exclude=frozenset({5})),
+        io=rp.MomentIO(lookup=lookup, record_rule_surfaced=surfaced),
+    )
+    assert result.shown_rule_ids == [9, 5]
+    assert result.rule_ids == [9]
+    assert result.lines[1].startswith("Also")
+    assert surfaced.call_args.kwargs["rule_ids"] == [9]
+
+
+async def test_an_unbound_session_asks_for_global_rules_only():
+    lookup = AsyncMock(return_value=[])
+    await rp.run_moment_arm(PUSH, _moment(project_id=0),
+                            io=rp.MomentIO(lookup=lookup, record_rule_surfaced=MagicMock()))
+    assert lookup.await_args.args[2] is None
+
+
+async def test_the_arm_fails_open():
+    broken = AsyncMock(side_effect=RuntimeError("db down"))
+    result = await rp.run_moment_arm(
+        PUSH, _moment(), io=rp.MomentIO(lookup=broken, record_rule_surfaced=MagicMock()),
+    )
+    assert result.lines == [] and result.rule_ids == []
+
+    # A recorder that fails costs the telemetry, never the lines.
+    result = await rp.run_moment_arm(
+        PUSH, _moment(),
+        io=rp.MomentIO(lookup=AsyncMock(return_value=[(_rule(5), "work.deliver")]),
+                       record_rule_surfaced=MagicMock(side_effect=RuntimeError("x"))),
+    )
+    assert len(result.lines) == 1
+
+
+async def test_nothing_reached_asks_nothing():
+    lookup = AsyncMock()
+    result = await rp.run_moment_arm([], _moment(),
+                                     io=rp.MomentIO(lookup=lookup, record_rule_surfaced=MagicMock()))
+    assert result.lines == []
+    lookup.assert_not_awaited()
+
+
+# ── the act → the moments → the rules ───────────────────────────────────
+
+
+def _stub_db(pairs):
+    async def _moments_for(user_id, tool, tool_input):
+        return moment_actions.resolve(tool, tool_input, [])
+
+    return [
+        patch.object(md, "deliver_for_act", _REAL_DELIVER),
+        patch.object(moment_actions, "moments_for", _moments_for),
+        patch.object(rulebooks, "rules_on_moments", AsyncMock(return_value=pairs)),
+        patch.object(rule_usage, "record_rule_surfaced", MagicMock()),
+    ]
+
+
+async def _with(stack, coro_fn):
+    for p in stack:
+        p.start()
+    try:
+        return await coro_fn()
+    finally:
+        for p in reversed(stack):
+            p.stop()
+
+
+async def test_a_task_closed_through_the_tool_carries_its_mounted_rules():
+    data = {"id": 40, "status": "done", "project_id": 2}
+    out = await _with(_stub_db([(_rule(11), "work.finish")]), lambda: md.attach_moment_rules(
+        1, "update_task", {"status": "done", "project_id": 7}, data,
+    ))
+    assert out is data
+    assert out["moment_rules"]["rule_ids"] == [11]
+    assert "at work.finish, reached by `status=done`" in out["moment_rules"]["lines"][0]
+    assert out["moment_rules"]["open_with"] == "get_rule(id)"
+
+
+async def test_the_records_project_wins_over_the_callers_argument():
+    lookup = AsyncMock(return_value=[])
+    stack = _stub_db([])
+    stack[2] = patch.object(rulebooks, "rules_on_moments", lookup)
+    await _with(stack, lambda: md.attach_moment_rules(
+        1, "update_task", {"status": "done", "project_id": 7}, {"project_id": 2},
+    ))
+    assert lookup.await_args.args == (1, ["work.finish"], 2)
+
+
+async def test_an_act_that_reaches_nothing_mounted_adds_nothing():
+    data = {"id": 40}
+    out = await _with(_stub_db([]), lambda: md.attach_moment_rules(
+        1, "update_task", {"status": "todo"}, data,
+    ))
+    assert "moment_rules" not in out
+
+
+async def test_the_attach_fails_open_and_passes_other_payloads_through():
+    data = {"id": 1}
+    with patch.object(md, "deliver_for_act", AsyncMock(side_effect=RuntimeError("x"))):
+        assert await md.attach_moment_rules(1, "create_note", {}, data) == {"id": 1}
+    marker = object()
+    assert await md.attach_moment_rules(1, "start_planning", {}, marker) is marker
+
+
+# ── which tools the plugin needs to ask about ───────────────────────────
+
+
+async def _reachable(mounted, mappings=()):
+    with patch.object(rulebooks, "mounted_moments", AsyncMock(return_value=set(mounted))), \
+         patch.object(moment_actions, "list_mappings", AsyncMock(return_value=list(mappings))):
+        return await md.reachable_tools(1)
+
+
+async def test_an_install_with_nothing_mounted_keeps_the_hook_off_the_wire():
+    listing = AsyncMock()
+    with patch.object(rulebooks, "mounted_moments", AsyncMock(return_value=set())), \
+         patch.object(moment_actions, "list_mappings", listing):
+        assert await md.reachable_tools(1) == []
+    listing.assert_not_awaited()
+
+
+async def test_only_the_tools_whose_moments_carry_a_mount_are_listed():
+    assert await _reachable({"work.deliver"}) == ["bash"]
+    assert await _reachable({"work.finish"}) == ["update_milestone", "update_task"]
+    assert await _reachable({"skill.release"}) == ["skill"]
+
+
+async def test_an_installs_own_mapping_and_removal_both_count():
+    own = SimpleNamespace(tool="mcp__deploy__ship", match="", moment="work.deliver", effect="add")
+    assert "ship" in await _reachable({"work.deliver"}, [own])
+    gone = SimpleNamespace(tool="AskUserQuestion", match="", moment="reply.ask", effect="remove")
+    assert await _reachable({"reply.ask"}, [gone]) == []
+
+
+# ── the plugin routes ───────────────────────────────────────────────────
+
+
+async def test_the_route_reads_the_event_and_the_ledger():
+    from scribe.routes import plugin as routes
+
+    deliver = AsyncMock(return_value=(PUSH, rp.RuleResult(
+        lines=["a line"], rule_ids=[5], shown_rule_ids=[5, 9],
+    )))
+    app = Quart(__name__)
+    event = {"session_id": "s", "tool_name": "Bash", "tool_input": {"command": "git push"}}
+    async with app.test_request_context(
+        "/api/plugin/moment", method="POST", json=event,
+        query_string={"project_id": "3", "exclude_rule_ids": "9", "held_rule_ids": "4"},
+    ):
+        g.user = SimpleNamespace(id=7)
+        with patch.object(routes.moment_delivery_svc, "deliver_for_act", deliver):
+            resp = await routes.moment.__wrapped__()
+        body = await resp.get_json()
+    assert body == {"context": "a line", "rule_ids": [5], "moments": ["work.deliver"]}
+    args, kw = deliver.await_args
+    assert args == (7, "Bash", {"command": "git push"})
+    assert kw == {"project_id": 3, "exclude": frozenset({9}), "held": frozenset({4})}
+
+
+async def test_the_route_answers_an_empty_event_with_nothing():
+    from scribe.routes import plugin as routes
+
+    app = Quart(__name__)
+    async with app.test_request_context("/api/plugin/moment", method="POST", json={}):
+        g.user = SimpleNamespace(id=7)
+        resp = await routes.moment.__wrapped__()
+        assert await resp.get_json() == {"context": "", "rule_ids": [], "moments": []}
+
+
+def test_both_routes_are_on_the_app():
+    from scribe.app import create_app
+
+    rules = {str(r.rule) for r in create_app().url_map.iter_rules()}
+    assert {"/api/plugin/moment", "/api/plugin/moment-tools"} <= rules
+
+
+# ── the hook ────────────────────────────────────────────────────────────
+
+
+def test_the_hook_is_registered_on_every_tool():
+    manifest = json.loads((ROOT / "plugin" / "hooks" / "hooks.json").read_text())
+    entries = {m.get("matcher"): [h["command"] for h in m["hooks"]]
+               for m in manifest["hooks"]["PreToolUse"]}
+    assert any("scribe_moment.sh" in c for c in entries.get("*", []))
+
+
+def test_the_hook_and_the_routes_agree_on_every_name():
+    """Rule 33 across the shell/Python seam: a renamed arg fails silently."""
+    src = HOOK.read_text()
+    assert "/api/plugin/moment-tools" in src and "/api/plugin/moment" in src
+    assert "exclude_rule_ids" in src and "scribe_held_query" in src
+    assert "scribe_rules_live" in src and "scribe_rules_append" in src
+    # The SHARED ledger, not one of its own.
+    assert '"${TMPDIR:-/tmp}/scribe-priorart"' in src and ".rules.ids" in src
+    for field in (".tools", ".rule_ids", ".context"):
+        assert f"'{field}'" in src
+
+    from scribe.routes import plugin as routes
+
+    route = inspect.getsource(routes.moment)
+    for name in ("exclude_rule_ids", "held_rule_ids", "tool_name", "tool_input"):
+        assert name in route
+
+
+def test_the_hook_leaves_scribes_own_tools_to_their_responses():
+    assert "mcp__*scribe*__*) exit 0" in HOOK.read_text()
+
+
+def test_the_hook_never_returns_a_permission_decision():
+    """A mount is a recall aid: it informs the act and never holds it."""
+    code = [ln for ln in HOOK.read_text().splitlines() if not ln.lstrip().startswith("#")]
+    assert not any("permissionDecision" in ln or "scribe_json_deny" in ln for ln in code)
+
+
+def _hook(tmp_path, port, tool="Bash", command="git push origin dev", session="s-moment"):
+    need_tools("bash", "curl", "awk")
+    env = {"PATH": os.environ["PATH"], "SCRIBE_URL": f"http://127.0.0.1:{port}",
+           "SCRIBE_TOKEN": "t", "TMPDIR": str(tmp_path), "HOME": str(tmp_path)}
+    out = subprocess.run(
+        ["bash", str(HOOK)],
+        input=json.dumps({"session_id": session, "cwd": str(tmp_path), "tool_name": tool,
+                          "tool_input": {"command": command}}),
+        capture_output=True, text=True, env=env, timeout=30,
+    )
+    assert out.returncode == 0, out.stderr
+    return out.stdout
+
+
+TOOLS = {"/api/plugin/moment-tools": b'{"tools":["bash"]}'}
+
+
+def test_a_listed_tool_is_asked_about_and_its_rules_injected(tmp_path):
+    replies = dict(TOOLS)
+    replies["/api/plugin/moment"] = json.dumps(
+        {"context": "Standing rule that may apply at work.deliver", "rule_ids": [5],
+         "moments": ["work.deliver"]}).encode()
+    with http_sink(by_path=replies) as (port, seen):
+        out = _hook(tmp_path, port)
+        # The list is read once per window, not once per call.
+        _hook(tmp_path, port)
+    paths = [e["_path"] for e in seen]
+    assert paths == ["/api/plugin/moment-tools", "/api/plugin/moment", "/api/plugin/moment"]
+    sent = json.loads(seen[1]["_body"])
+    assert sent["tool_input"] == {"command": "git push origin dev"}
+    # The first call's fresh rule is on the shared ledger for the second.
+    assert seen[2]["exclude_rule_ids"] == ["5"]
+    ctx = json.loads(out)["hookSpecificOutput"]["additionalContext"]
+    assert "work.deliver" in ctx
+
+
+def test_an_unlisted_tool_never_leaves_the_machine(tmp_path):
+    with http_sink(by_path=TOOLS) as (port, seen):
+        assert _hook(tmp_path, port, tool="Read") == ""
+        assert _hook(tmp_path, port, tool="mcp__plugin_scribe_scribe__update_task") == ""
+    assert [e["_path"] for e in seen] == ["/api/plugin/moment-tools"]
+
+
+def test_an_install_with_nothing_mounted_sends_one_request_per_window(tmp_path):
+    with http_sink(by_path={"/api/plugin/moment-tools": b'{"tools":[]}'}) as (port, seen):
+        for _ in range(3):
+            assert _hook(tmp_path, port) == ""
+    assert len(seen) == 1
+
+
+# ── the MCP door: every shipped Scribe action carries the attach ────────
+
+
+def _attached_names(path: Path) -> dict[str, set[str]]:
+    """{function name: {tool names passed to attach_moment_rules in it}}."""
+    out: dict[str, set[str]] = {}
+    for node in ast.parse(path.read_text()).body:
+        if not isinstance(node, ast.AsyncFunctionDef):
+            continue
+        for call in ast.walk(node):
+            if (isinstance(call, ast.Call) and isinstance(call.func, ast.Name)
+                    and call.func.id == "attach_moment_rules" and len(call.args) > 1
+                    and isinstance(call.args[1], ast.Constant)):
+                out.setdefault(node.name, set()).add(call.args[1].value)
+    return out
+
+
+def test_every_scribe_tool_the_defaults_name_attaches_its_moment_rules():
+    tools_dir = ROOT / "src" / "scribe" / "mcp" / "tools"
+    defined: dict[str, dict[str, set[str]]] = {}
+    scribe_tools: set[str] = set()
+    for path in tools_dir.glob("*.py"):
+        tree = ast.parse(path.read_text())
+        scribe_tools |= {n.name for n in tree.body if isinstance(n, ast.AsyncFunctionDef)}
+        defined[path.name] = _attached_names(path)
+
+    shipped = {a.tool for a in moment_actions.DEFAULT_ACTIONS} & scribe_tools
+    assert {"update_task", "create_rule", "update_milestone"} <= shipped
+    for tool in sorted(shipped):
+        named = set().union(*(per_file.get(tool, set()) for per_file in defined.values()))
+        assert tool in named, (
+            f"{tool} is mapped onto a moment by the shipped defaults but does not "
+            f"attach its mounted rules — a client without the plugin would never "
+            f"receive them"
+        )