feat(moments): rules mount on moments, through every rule door (milestone 458 step 3, #4921)

rule_moments (migration 0117) records which moments a rule arrives at, by
catalog name, cascading with the rule. rule_detail, the one seam every
rule door already returns through, gains moments beside system_ids: None
leaves the mounts alone, a list replaces them. get_rule and both
list_rules doors read them back, batched per page.

All five MCP rule/preference writes and the three REST ones take moments
and validate them before their create or update. An unknown name is
refused with the catalog listed and leaves no half-made rule behind; a
parity test pins that ordering on every door.

Backup v22 carries the mounts as a join table remapped through the rule
map; a real-Postgres round trip checks they land on the restored rule.

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

alembic/versions/0117_rule_moments.py

@@ -0,0 +1,38 @@
+"""rule_moments — the moments of work a rule arrives at (milestone 458 step 3)
+
+Revision ID: 0117
+Revises: 0116
+Create Date: 2026-10-05
+
+A rule mounted on a moment is delivered whenever that moment happens, by
+lookup rather than by score. The moment is a catalog name, validated by the
+service; the catalog is code, so there is no table to key it to. Cascades
+with the rule, like rule_systems beside it.
+"""
+import sqlalchemy as sa
+from alembic import op
+
+revision = "0117"
+down_revision = "0116"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "rule_moments",
+        sa.Column(
+            "rule_id", sa.BigInteger(),
+            sa.ForeignKey("rules.id", ondelete="CASCADE"), primary_key=True,
+        ),
+        sa.Column("moment", sa.Text(), primary_key=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=True),
+    )
+    # The delivery read (step 4) asks "which rules are on this moment?", the
+    # opposite direction from the primary key.
+    op.create_index("ix_rule_moments_moment", "rule_moments", ["moment"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_rule_moments_moment", table_name="rule_moments")
+    op.drop_table("rule_moments")

src/scribe/mcp/tools/rulebooks.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
 from scribe.services import dedup as dedup_svc
+from scribe.services import moments as moments_svc
 from scribe.services import rulebooks as rulebooks_svc
 from scribe.services import trash as trash_svc
 from scribe.services.rule_usage import (
@@ -192,14 +193,16 @@ async def delete_topic(topic_id: int, confirmed: bool = False) -> dict:
 
 # ── Rule CRUD ──────────────────────────────────────────────────────────
 
-def _rule_summary(r) -> dict:
+def _rule_summary(r, moments: list[str] | None = None) -> dict:
     """The list-row shape for a rule: what an agent needs to APPLY it. The
     full record (why, how_to_apply, timestamps) is get_rule's job.
 
     One line, because the shape itself lives in the service — this was one of
     three hand-written copies that had already drifted apart (note 3026).
+    `moments` rides along only when the rule is mounted (rule_brief drops a
+    None), so an unmounted rule's row says nothing rather than "[]".
     """
-    return rulebooks_svc.rule_brief(r)
+    return rulebooks_svc.rule_brief(r, moments=moments or None)
 
 
 async def list_rules(
@@ -224,7 +227,12 @@ async def list_rules(
         topic_id=topic_id or None,
         project_id=project_id or None,
     )
-    return {"rules": [_rule_summary(r) for r in rows], "total": len(rows)}
+    # One batched read for the page, not one per row.
+    mounted = await rulebooks_svc.list_rule_moments([r.id for r in rows])
+    return {
+        "rules": [_rule_summary(r, mounted.get(r.id)) for r in rows],
+        "total": len(rows),
+    }
 
 
 async def get_rule(rule_id: int) -> dict:
@@ -311,7 +319,8 @@ async def create_rule(
     topic_id: int, title: str, statement: str, when_to_apply: str,
     why: str = "", how_to_apply: str = "", order_index: int = 0,
     arose_from_id: int = 0, verify_with: str = "", expires_when: str = "",
-    system_ids: list[int] | None = None, force: bool = False,
+    system_ids: list[int] | None = None,
+    moments: list[str] | None = None, force: bool = False,
 ) -> dict:
     """Create a new rule in a rulebook (a SHARED rule — keep it general).
 
@@ -472,6 +481,12 @@ async def create_rule(
         system_ids: Ids from list_canonical_systems — the global AREAS this
             rule is about. This is what lets a rule reach a project that is
             working in that area, so a CI rule surfaces on a CI change.
+        moments: Moments from list_moments this rule arrives at — whenever
+            one happens, the rule is delivered, whatever the words of the work
+            look like. Mount a rule here when it is about WHEN something is
+            done rather than what it is about: a rule on when work counts as
+            finished belongs at work.finish and reply.report, where nothing
+            said resembles it. Replaces the set; [] clears it.
         arose_from_id: The note or task that CAUSED this rule (an incident, a
             decision). Prefer this over naming the record inside `why`, which
             cannot be followed and does not survive a rewording.
@@ -503,6 +518,7 @@ async def create_rule(
             `overlaps` and `overlap_note` — read the top one and decide.
     """
     uid = current_user_id()
+    moments = moments_svc.require_moments(moments)
     if not force:
         dup = await dedup_svc.find_duplicate_rule(title, topic_id=topic_id)
         if dup is not None:
@@ -518,7 +534,7 @@ async def create_rule(
         why=why, how_to_apply=how_to_apply, order_index=order_index,
         verify_with=verify_with, expires_when=expires_when,
     )
-    data = await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data = await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
     data.update(dedup_svc.overlap_response(overlaps, "rule"))
     return data
 
@@ -527,7 +543,8 @@ async def create_project_rule(
     project_id: int, statement: str, when_to_apply: str, title: str = "",
     why: str = "", how_to_apply: str = "", order_index: int = 0,
     arose_from_id: int = 0, verify_with: str = "", expires_when: str = "",
-    system_ids: list[int] | None = None, force: bool = False,
+    system_ids: list[int] | None = None,
+    moments: list[str] | None = None, force: bool = False,
 ) -> dict:
     """Create a rule scoped to a single project (no rulebook needed).
 
@@ -581,6 +598,12 @@ async def create_project_rule(
             rule is about. Worth setting even on a project rule: it is what
             lets a conditional one surface when the project is working in
             that area.
+        moments: Moments from list_moments this rule arrives at — whenever
+            one happens, the rule is delivered, whatever the words of the work
+            look like. Mount a rule here when it is about WHEN something is
+            done rather than what it is about: a rule on when work counts as
+            finished belongs at work.finish and reply.report, where nothing
+            said resembles it. Replaces the set; [] clears it.
         arose_from_id: The note or task that CAUSED this rule. Reach for it
             harder here than on a rulebook rule — a project rule usually
             comes from one traceable incident in this repo, where a family
@@ -604,6 +627,7 @@ async def create_project_rule(
             `overlap_note` on the reply — see create_rule.
     """
     uid = current_user_id()
+    moments = moments_svc.require_moments(moments)
     derived_title = title.strip() or statement.strip().split(".")[0][:50]
     if not force:
         dup = await dedup_svc.find_duplicate_rule(derived_title, project_id=project_id)
@@ -619,7 +643,7 @@ async def create_project_rule(
         why=why, how_to_apply=how_to_apply, order_index=order_index,
         verify_with=verify_with, expires_when=expires_when,
     )
-    data = await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data = await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
     data.update(dedup_svc.overlap_response(overlaps, "rule"))
     return data
 
@@ -627,7 +651,8 @@ async def create_project_rule(
 async def update_rule(
     rule_id: int, title: str = "", statement: str = "", when_to_apply: str = "",
     why: str = "", how_to_apply: str = "", order_index: int = -1,
-    system_ids: list[int] | None = None, arose_from_id: int = 0,
+    system_ids: list[int] | None = None,
+    moments: list[str] | None = None, arose_from_id: int = 0,
     verify_with: str = "", expires_when: str = "", kind: str = "",
     clear_fields: list[str] | None = None,
 ) -> dict:
@@ -644,6 +669,9 @@ async def update_rule(
     milestone 394, so a rule with no trigger is not a quiet rule — it is one
     no session will ever be shown. `system_ids` REPLACES the rule's areas
     (pass [] to clear), and they decide which PROJECTS a rule binds by area.
+    `moments` REPLACES the moments it arrives at the same way — see
+    create_rule; mounting a rule whose trigger keeps missing on its moment is
+    often the better fix than rewriting the trigger.
 
     RETROFITTING A TRIGGER HAS ITS OWN TRAP, and it is not the one create_rule
     warns about. There the field is empty and the instruction is "write one".
@@ -690,6 +718,7 @@ async def update_rule(
         clear_fields: Names of fields to empty, as above.
     """
     uid = current_user_id()
+    moments = moments_svc.require_moments(moments)
     fields: dict = {}
     if title:
         fields["title"] = title
@@ -716,7 +745,7 @@ async def update_rule(
     )
     if rule is None:
         raise ValueError(f"rule {rule_id} not found")
-    return await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    return await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
 
 
 # ── Preferences ─────────────────────────────────────────────────────────
@@ -737,6 +766,7 @@ async def create_preference(
     topic_id: int, title: str, statement: str, when_to_apply: str,
     arose_from_id: int, why: str = "", how_to_apply: str = "",
     order_index: int = 0, system_ids: list[int] | None = None,
+    moments: list[str] | None = None,
     force: bool = False,
 ) -> dict:
     """Record how the operator wants work done. No approval loop — write it.
@@ -806,6 +836,12 @@ async def create_preference(
             preference is about, which is what lets it reach a session working
             in that area. `update_preference` took this and create did not, so
             a preference could only be filed after the fact (#4249).
+        moments: Moments from list_moments this rule arrives at — whenever
+            one happens, the rule is delivered, whatever the words of the work
+            look like. Mount a rule here when it is about WHEN something is
+            done rather than what it is about: a rule on when work counts as
+            finished belongs at work.finish and reply.report, where nothing
+            said resembles it. Replaces the set; [] clears it.
         force: Bypass the near-duplicate gate. For a genuinely distinct
             preference, not for one that is "mostly" different — a mostly
             different preference is an update. A RULE that already answers
@@ -814,6 +850,7 @@ async def create_preference(
             the weaker copy of it and should go.
     """
     uid = current_user_id()
+    moments = moments_svc.require_moments(moments)
     if not when_to_apply.strip():
         raise ValueError(
             "when_to_apply is required: a preference with no trigger never "
@@ -839,7 +876,7 @@ async def create_preference(
         kind="preference", arose_from_id=arose_from_id,
         why=why, how_to_apply=how_to_apply, order_index=order_index,
     )
-    data = await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    data = await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
     data.update(dedup_svc.overlap_response(overlaps, "preference"))
     return data
 
@@ -848,7 +885,8 @@ async def update_preference(
     rule_id: int, arose_from_id: int, statement: str = "",
     when_to_apply: str = "", title: str = "", why: str = "",
     how_to_apply: str = "", order_index: int = -1,
-    system_ids: list[int] | None = None, clear_fields: list[str] | None = None,
+    system_ids: list[int] | None = None,
+    moments: list[str] | None = None, clear_fields: list[str] | None = None,
 ) -> dict:
     """Bring a preference up to date. Doing this mid-work is expected.
 
@@ -900,9 +938,12 @@ async def update_preference(
     Args:
         rule_id: The preference to update.
         arose_from_id: What taught this change. Required; see above.
+        moments: Replaces the moments this preference arrives at (see
+            create_preference); omit to leave them alone.
         when_to_apply: The moment it applies, in session vocabulary. See above.
     """
     uid = current_user_id()
+    moments = moments_svc.require_moments(moments)
     if not arose_from_id:
         raise ValueError(
             "arose_from_id is required: this edit is the record of how the "
@@ -927,7 +968,7 @@ async def update_preference(
     )
     if rule is None:
         raise ValueError(f"rule {rule_id} not found")
-    return await rulebooks_svc.rule_detail(uid, rule, system_ids)
+    return await rulebooks_svc.rule_detail(uid, rule, system_ids, moments)
 
 
 async def rule_history(rule_id: int, version_id: int = 0) -> dict:

src/scribe/models/__init__.py

@@ -78,7 +78,7 @@ from scribe.models.user_profile import UserProfile  # noqa: E402, F401
 # Imported before rulebook: rule_systems foreign-keys canonical_systems.
 from scribe.models.canonical_system import CanonicalSystem  # noqa: E402, F401
 from scribe.models.rulebook import (  # noqa: E402, F401
-    Rulebook, RulebookTopic, Rule, RuleRelation, rule_systems,
+    Rulebook, RulebookTopic, Rule, RuleRelation, rule_moments, rule_systems,
 )
 # After notes and rules: it foreign-keys both (milestone 440).
 from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink  # noqa: E402, F401

src/scribe/models/rulebook.py

@@ -188,6 +188,21 @@ rule_systems = Table(
 )
 
 
+# Which MOMENTS of work a rule arrives at (milestone 458). A rule mounted on a
+# moment is delivered whenever that moment happens, by lookup and with no
+# score — the route for a rule that is about WHEN something is done rather
+# than what it is about, which semantic retrieval structurally misses. The
+# moment is a catalog NAME (`services/moments.py`), not a key: the catalog is
+# code, so there is nothing to point a foreign key at, and the service refuses
+# an unknown name before it is stored.
+rule_moments = Table(
+    "rule_moments",
+    Base.metadata,
+    Column("rule_id", BigInteger, ForeignKey("rules.id", ondelete="CASCADE"), primary_key=True),
+    Column("moment", Text, primary_key=True),
+    Column("created_at", DateTime(timezone=True), default=lambda: datetime.now(timezone.utc)),
+)
+
 class RuleRelation(Base, CreatedAtMixin):
     """A typed edge between two rules. Each kind exists because its ABSENCE
     forced a workaround somewhere in the operator's rulebook (note 3026).

src/scribe/routes/rulebooks.py

@@ -8,6 +8,7 @@ from __future__ import annotations
 from quart import Blueprint, jsonify, request
 
 from scribe.auth import get_current_user_id, login_required
+import scribe.services.moments as moments_svc
 import scribe.services.rulebooks as rulebooks_svc
 from scribe.services.trash import delete as trash_delete
 from scribe.services.rule_usage import (
@@ -17,6 +18,15 @@ from scribe.services.rule_usage import (
 rulebooks_bp = Blueprint("rulebooks", __name__, url_prefix="/api")
 
 
+def _moments_or_refusal(data: dict):
+    """The request's `moments`, validated BEFORE any write — so an unknown
+    name is a 400 and not a rule created without the mounts it asked for.
+    Absent means "leave them alone" (None), as on the MCP door."""
+    try:
+        return moments_svc.require_moments(data.get("moments")), None
+    except ValueError as exc:
+        return None, (jsonify({"error": str(exc)}), 400)
+
 
 # ── Rulebooks ───────────────────────────────────────────────────────────
 
@@ -154,8 +164,12 @@ async def list_rules():
     # than for snippets: every rule on every install predates this table, so
     # for a while the zero-filled shape IS the common case.
     usage = await usage_for_rules([int(it["id"]) for it in items])
+    # Zero-filled like `usage`, for the same reason: the UI renders "mounted
+    # nowhere" from an empty list rather than treating a missing key as a state.
+    mounted = await rulebooks_svc.list_rule_moments([int(it["id"]) for it in items])
     for it in items:
         it["usage"] = usage.get(int(it["id"]), empty_rule_usage())
+        it["moments"] = mounted.get(int(it["id"]), [])
     return jsonify({"rules": items})
 
 
@@ -167,6 +181,9 @@ async def create_rule(topic_id: int):
     statement = (data.get("statement") or "").strip()
     if not title or not statement:
         return jsonify({"error": "title and statement are required"}), 400
+    moments, refused = _moments_or_refusal(data)
+    if refused:
+        return refused
     try:
         rule = await rulebooks_svc.create_rule(
             topic_id=topic_id,
@@ -189,7 +206,7 @@ async def create_rule(topic_id: int):
     except ValueError as exc:
         return jsonify({"error": str(exc)}), 404
     return jsonify(await rulebooks_svc.rule_detail(
-        get_current_user_id(), rule, data.get("system_ids"),
+        get_current_user_id(), rule, data.get("system_ids"), moments,
     )), 201
 
 
@@ -223,10 +240,15 @@ async def update_rule(rule_id: int):
     # service normalises "" to NULL for every nullable text column. The MCP
     # door needs the explicit list only because "" already means "unchanged"
     # there — two idioms, one outcome.
+    moments, refused = _moments_or_refusal(data)
+    if refused:
+        return refused
     rule = await rulebooks_svc.update_rule(rule_id, uid, **fields)
     if rule is None:
         return jsonify({"error": "rule not found"}), 404
-    return jsonify(await rulebooks_svc.rule_detail(uid, rule, data.get("system_ids")))
+    return jsonify(await rulebooks_svc.rule_detail(
+        uid, rule, data.get("system_ids"), moments,
+    ))
 
 
 @rulebooks_bp.get("/rules/<int:rule_id>/versions")
@@ -401,6 +423,9 @@ async def create_project_rule(project_id: int):
                      "surfaces at the moment it applies."
         }), 400
     title = (data.get("title") or "").strip() or statement.split(".")[0][:50]
+    moments, refused = _moments_or_refusal(data)
+    if refused:
+        return refused
     try:
         rule = await rulebooks_svc.create_project_rule(
             project_id=project_id,
@@ -423,7 +448,7 @@ async def create_project_rule(project_id: int):
     except ValueError as exc:
         return jsonify({"error": str(exc)}), 404
     return jsonify(await rulebooks_svc.rule_detail(
-        get_current_user_id(), rule, data.get("system_ids"),
+        get_current_user_id(), rule, data.get("system_ids"), moments,
     )), 201
 
 

src/scribe/services/backup.py

@@ -17,7 +17,9 @@ from scribe.models.system_usage import SystemUsageEvent
 from scribe.models.moment_mapping import MomentMapping
 from scribe.models.retrieval_tuning import RetrievalTuningEvent
 from scribe.models.canonical_system import CanonicalSystem
-from scribe.models.rulebook import RuleRelation, rule_systems as rule_systems_t
+from scribe.models.rulebook import (
+    RuleRelation, rule_moments as rule_moments_t, rule_systems as rule_systems_t,
+)
 from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink
 from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
 from scribe.models.project import Project
@@ -98,8 +100,12 @@ logger = logging.getLogger(__name__)
 # actions reach which moment, and the shipped defaults it switched off. Each
 # row is a correction an operator made in-session; a restore that dropped them
 # would silently put every misfire back.
+# v22 (2026-10) added rule_moments (milestone 458): the moments each rule
+# arrives at. A mount is a judgment about when a rule applies that nothing in
+# the rule's text records, and losing it would put back exactly the misses the
+# mounts were made to fix.
 # Bump when the serialized schema changes.
-BACKUP_VERSION = 21
+BACKUP_VERSION = 22
 
 # Every table this backup carries, by its REAL name. Paired with _NOT_INCLUDED
 # below, these two lists must together account for the entire schema — which is
@@ -146,6 +152,8 @@ _BACKED_UP = [
     "system_usage_events",
     # v21 (2026-10): an install's action → moment corrections (milestone 458).
     "moment_mappings",
+    # v22 (2026-10): the moments each rule arrives at (milestone 458).
+    "rule_moments",
 ]
 
 # Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -741,6 +749,12 @@ def _topic_rows(rows) -> list[dict]:
     ]
 
 
+def _rule_moment_rows(rows) -> list[dict]:
+    """A rule's mounts. The moment is a catalog NAME, the same on every
+    install, so only the rule id needs remapping on the way back in."""
+    return [{"rule_id": rule_id, "moment": moment} for rule_id, moment in rows]
+
+
 def _rule_system_rows(rows) -> list[dict]:
     """A rule's area tags, carried by canonical SLUG for the same reason the
     Systems are: the catalog is global and its ids are per-install."""
@@ -833,6 +847,9 @@ async def export_full_backup() -> dict:
             select(rule_systems_t.c.rule_id, CanonicalSystem.slug)
             .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
         )).all()
+        rule_moment_rows = (await session.execute(
+            select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
+        )).all()
         rule_relations = (await session.execute(select(RuleRelation))).scalars().all()
         lesson_rule_links = (await session.execute(select(LessonRuleLink))).scalars().all()
         lesson_no_rule = (await session.execute(select(LessonNoRule))).scalars().all()
@@ -899,6 +916,7 @@ async def export_full_backup() -> dict:
         "rules": _rule_rows(rules),
         "canonical_systems": _canonical_system_rows(canonical_systems),
         "rule_systems": _rule_system_rows(rule_system_rows),
+        "rule_moments": _rule_moment_rows(rule_moment_rows),
         "rule_relations": _rule_relation_rows(rule_relations),
         "lesson_rule_links": _lesson_rule_link_rows(lesson_rule_links),
         "lesson_no_rule": _lesson_no_rule_rows(lesson_no_rule),
@@ -1033,6 +1051,10 @@ async def export_user_backup(user_id: int) -> dict:
             .join(CanonicalSystem, CanonicalSystem.id == rule_systems_t.c.canonical_id)
             .where(rule_systems_t.c.rule_id.in_(_rule_ids))
         )).all() if _rule_ids else []
+        rule_moment_rows = (await session.execute(
+            select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
+            .where(rule_moments_t.c.rule_id.in_(_rule_ids))
+        )).all() if _rule_ids else []
         # Scoped through the RULE, not the version's user_id. That column is
         # the ACTOR (milestone 323), so filtering on it would carry the
         # versions this user wrote on someone ELSE's rule and drop the ones
@@ -1114,6 +1136,7 @@ async def export_user_backup(user_id: int) -> dict:
         "rules": _rule_rows(rules),
         "canonical_systems": _canonical_system_rows(canonical_systems),
         "rule_systems": _rule_system_rows(rule_system_rows),
+        "rule_moments": _rule_moment_rows(rule_moment_rows),
         "rule_relations": _rule_relation_rows(rule_relations),
         "lesson_rule_links": _lesson_rule_link_rows(lesson_rule_links),
         "lesson_no_rule": _lesson_no_rule_rows(lesson_no_rule),
@@ -1926,7 +1949,8 @@ async def _restore_v2(data: dict) -> dict:
         "repo_bindings": 0,
         "note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
         "code_shape_uses": 0, "canonical_systems": 0,
-        "rule_systems": 0, "rule_relations": 0, "rule_versions": 0,
+        "rule_systems": 0, "rule_moments": 0,
+        "rule_relations": 0, "rule_versions": 0,
         "retrieval_tuning_events": 0, "lesson_rule_links": 0,
         "lesson_no_rule": 0, "moment_mappings": 0,
     }
@@ -2125,6 +2149,17 @@ async def _restore_v2(data: dict) -> dict:
             ))
             stats["rule_systems"] += 1
 
+        # The same shape for a rule's mounts (v22): a join table with no
+        # model, so no builder — the rule id is the only thing to remap.
+        for rm in data.get("rule_moments", []):
+            mapped_rule = maps.rules.get(rm.get("rule_id", 0))
+            if mapped_rule is None or not rm.get("moment"):
+                continue
+            await session.execute(rule_moments_t.insert().values(
+                rule_id=mapped_rule, moment=rm["moment"],
+            ))
+            stats["rule_moments"] += 1
+
         for rr in data.get("rule_relations", []):
             relation = _build_rule_relation(rr, maps)
             if relation is None:

src/scribe/services/moments.py

@@ -142,6 +142,20 @@ def require_moment(name: str) -> str:
     )
 
 
+def require_moments(names: list[str] | None) -> list[str] | None:
+    """A rule's moments, normalised and de-duplicated — or the first refusal.
+
+    None passes through as None, which every rule door reads as "leave the
+    mounts alone"; a list, [] included, is the whole set after the write.
+    All are checked before any is stored, so a refusal leaves nothing
+    half-mounted.
+    """
+    if names is None:
+        return None
+    if isinstance(names, str):
+        names = [names]
+    return list(dict.fromkeys(require_moment(n) for n in names))
+
 def catalog() -> dict:
     """The catalog as plain data, for the MCP tool and the REST door alike."""
     return {

src/scribe/services/rulebooks.py

@@ -433,26 +433,36 @@ async def co_surfaced_partners(user_id: int, rule_ids: list[int]) -> list[Rule]:
         return out
 
 
-async def rule_detail(user_id: int, rule: Rule, system_ids: list[int] | None = None) -> dict:
-    """The full record, with its areas and edges attached.
+async def rule_detail(
+    user_id: int, rule: Rule, system_ids: list[int] | None = None,
+    moments: list[str] | None = None,
+) -> dict:
+    """The full record, with its areas, moments and edges attached.
 
     ONE seam for both doors and every write path, so create, update and get
     cannot disagree about what a rule looks like coming back — the same
     reasoning as attach_relations for notes (#2859), and the same reasoning
     rule_brief exists for one level down.
 
-    `system_ids=None` means "leave the tags alone"; a list (including [])
-    REPLACES them.
+    `system_ids=None` / `moments=None` mean "leave them alone"; a list
+    (including []) REPLACES them. Doors validate `moments` with
+    `moments.require_moments` BEFORE their create, so an unknown name is
+    refused without leaving a rule behind.
     """
     if system_ids is not None:
         await set_rule_systems(rule.id, user_id, system_ids)
+    if moments is not None:
+        await set_rule_moments(rule.id, user_id, moments)
     data = rule.to_dict()
     systems = (await list_rule_systems([rule.id])).get(rule.id, [])
+    mounted = (await list_rule_moments([rule.id])).get(rule.id, [])
     relations = (await list_rule_relations([rule.id])).get(rule.id, [])
     # Attached only when present (#2483): an empty key reads as a capability
     # the record has and isn't using, which is a different claim.
     if systems:
         data["systems"] = systems
+    if mounted:
+        data["moments"] = mounted
     if relations:
         data["relations"] = relations
     # The concrete situations judged (or proposed) to be instances of this
@@ -1100,6 +1110,56 @@ async def set_rule_systems(
         return sorted(wanted)
 
 
+async def set_rule_moments(
+    rule_id: int, user_id: int, moments: list[str],
+) -> list[str] | None:
+    """Replace which MOMENTS a rule arrives at (milestone 458). None if not owned.
+
+    Set-semantics like set_rule_systems: the list given IS the state after.
+    Every name is checked against the catalog first and the whole write is
+    refused on the first unknown one — a typo stored as a mount would read
+    back as attached and never fire, the silent miss moments exist to end.
+    """
+    from scribe.models.rulebook import rule_moments as rule_moments_t
+    from scribe.services.moments import require_moments
+
+    wanted = require_moments(moments) or []
+    async with async_session() as session:
+        rule = await _fetch_owned_rule(session, rule_id, user_id)
+        if rule is None:
+            return None
+        await session.execute(
+            sql_delete(rule_moments_t).where(rule_moments_t.c.rule_id == rule_id)
+        )
+        for moment in wanted:
+            await session.execute(
+                insert(rule_moments_t).values(rule_id=rule_id, moment=moment)
+            )
+        await session.commit()
+        return wanted
+
+
+async def list_rule_moments(rule_ids: list[int]) -> dict[int, list[str]]:
+    """The moments each of a batch of rules is mounted on, in catalog order."""
+    from scribe.models.rulebook import rule_moments as rule_moments_t
+    from scribe.services.moments import MOMENTS
+
+    if not rule_ids:
+        return {}
+    async with async_session() as session:
+        rows = (await session.execute(
+            select(rule_moments_t.c.rule_id, rule_moments_t.c.moment)
+            .where(rule_moments_t.c.rule_id.in_(rule_ids))
+        )).all()
+    order = {name: i for i, name in enumerate(MOMENTS)}
+    out: dict[int, list[str]] = {}
+    for rule_id, moment in rows:
+        out.setdefault(rule_id, []).append(moment)
+    for names in out.values():
+        names.sort(key=lambda n: (order.get(n, len(order)), n))
+    return out
+
+
 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/helpers.py

@@ -315,7 +315,7 @@ def plain_rule_detail():
     stubbing the seam slightly differently is a test asserting something
     slightly different than it appears to.
     """
-    async def _detail(_uid, rule, _system_ids=None):
+    async def _detail(_uid, rule, _system_ids=None, _moments=None):
         return rule.to_dict()
     return patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail", _detail)
 

tests/test_integration_rule_moments.py

@@ -0,0 +1,157 @@
+"""Real-Postgres tests for rule mounts (milestone 458 step 3).
+
+A mount is the claim "this rule arrives at this moment". These pin what that
+claim depends on:
+- the set given is the set stored, and a refused name stores nothing;
+- 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.
+"""
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
+from scribe.models.user import User
+from scribe.services import backup
+from scribe.services import rulebooks as rulebooks_svc
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+OWNER_USERNAME = "rule_moments_owner"
+STRANGER_USERNAME = "rule_moments_stranger"
+RESTORED_USERNAME = "rule_moments_restored"
+TRIGGER = "about to tell the operator a piece of work is finished"
+
+
+async def _purge_books(username: str) -> None:
+    """At SETUP: rule writes fire a detached embedding refresh, and a teardown
+    delete races it (test_integration_rule_versions records why)."""
+    async with async_session() as s:
+        for user in (await s.execute(
+            select(User).where(User.username == username)
+        )).scalars().all():
+            for book in (await s.execute(
+                select(Rulebook).where(Rulebook.owner_user_id == user.id)
+            )).scalars().all():
+                await s.delete(book)
+        await s.commit()
+
+
+@pytest_asyncio.fixture
+async def world():
+    for name in (OWNER_USERNAME, STRANGER_USERNAME, RESTORED_USERNAME):
+        await _purge_books(name)
+    # The restore below creates a user under a unique username; a previous
+    # run's must be gone first, as in the sibling roundtrip files.
+    async with async_session() as s:
+        for user in (await s.execute(
+            select(User).where(User.username == RESTORED_USERNAME)
+        )).scalars().all():
+            await s.delete(user)
+        await s.commit()
+    async with async_session() as s:
+        owner = await ensure_user(s, OWNER_USERNAME)
+        stranger = await ensure_user(s, STRANGER_USERNAME)
+        await s.commit()
+        uid, sid = owner.id, stranger.id
+    book = await rulebooks_svc.create_rulebook(uid, "Moment fixtures")
+    topic = await rulebooks_svc.create_topic(book.id, uid, "verification")
+    rule = await rulebooks_svc.create_rule(
+        topic.id, uid, "Done means delivered and checked",
+        "Work is done when it is delivered and its automated checks pass.",
+        when_to_apply=TRIGGER,
+    )
+    return {"uid": uid, "sid": sid, "book_id": book.id, "rule": rule}
+
+
+async def test_the_set_given_is_the_set_stored(world):
+    uid, rule = world["uid"], world["rule"]
+    assert await rulebooks_svc.set_rule_moments(
+        rule.id, uid, ["reply.report", " Work.Finish"],
+    ) == ["reply.report", "work.finish"]
+    # Read back in CATALOG order, whatever order they were given in.
+    assert (await rulebooks_svc.list_rule_moments([rule.id]))[rule.id] == [
+        "work.finish", "reply.report",
+    ]
+
+    await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.deliver"])
+    assert (await rulebooks_svc.list_rule_moments([rule.id]))[rule.id] == ["work.deliver"]
+
+    await rulebooks_svc.set_rule_moments(rule.id, uid, [])
+    assert rule.id not in await rulebooks_svc.list_rule_moments([rule.id])
+
+
+async def test_a_refused_name_stores_nothing(world):
+    uid, rule = world["uid"], world["rule"]
+    await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.finish"])
+    with pytest.raises(ValueError, match="unknown moment"):
+        await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.deliver", "work.shipped"])
+    assert (await rulebooks_svc.list_rule_moments([rule.id]))[rule.id] == ["work.finish"]
+
+
+async def test_only_the_owner_can_mount(world):
+    rule = world["rule"]
+    assert await rulebooks_svc.set_rule_moments(rule.id, world["sid"], ["work.finish"]) is None
+    assert rule.id not in await rulebooks_svc.list_rule_moments([rule.id])
+
+
+async def test_the_detail_carries_the_mounts_and_omits_an_empty_set(world):
+    uid, rule = world["uid"], world["rule"]
+    bare = await rulebooks_svc.rule_detail(uid, rule)
+    assert "moments" not in bare
+    mounted = await rulebooks_svc.rule_detail(uid, rule, moments=["work.finish"])
+    assert mounted["moments"] == ["work.finish"]
+    # None leaves them alone: a later read without the argument still has them.
+    assert (await rulebooks_svc.rule_detail(uid, rule))["moments"] == ["work.finish"]
+
+
+async def test_a_backup_carries_the_mounts_to_the_restored_rule(world):
+    uid, rule = world["uid"], world["rule"]
+    await rulebooks_svc.set_rule_moments(rule.id, uid, ["work.finish", "reply.report"])
+
+    async with async_session() as s:
+        payload = {
+            "version": backup.BACKUP_VERSION,
+            "users": backup._user_rows(
+                [(await s.execute(select(User).where(User.id == uid))).scalars().one()]
+            ),
+            "rulebooks": backup._rulebook_rows(
+                [(await s.execute(select(Rulebook).where(
+                    Rulebook.id == world["book_id"]))).scalars().one()]
+            ),
+            "rulebook_topics": backup._topic_rows((await s.execute(
+                select(RulebookTopic).where(RulebookTopic.rulebook_id == world["book_id"])
+            )).scalars().all()),
+            "rules": backup._rule_rows(
+                [(await s.execute(select(Rule).where(Rule.id == rule.id))).scalars().one()]
+            ),
+        }
+    exported = await backup.export_user_backup(uid)
+    payload["rule_moments"] = [
+        row for row in exported["rule_moments"] if row["rule_id"] == rule.id
+    ]
+    assert sorted(r["moment"] for r in payload["rule_moments"]) == [
+        "reply.report", "work.finish",
+    ]
+    payload["users"][0]["username"] = RESTORED_USERNAME
+
+    stats = await backup.restore_full_backup(payload)
+    assert stats["rule_moments"] == 2
+
+    async with async_session() as s:
+        restored_user = (await s.execute(
+            select(User).where(User.username == RESTORED_USERNAME)
+        )).scalars().one()
+        restored_rule = (await s.execute(
+            select(Rule).join(RulebookTopic, RulebookTopic.id == Rule.topic_id)
+            .join(Rulebook, Rulebook.id == RulebookTopic.rulebook_id)
+            .where(Rulebook.owner_user_id == restored_user.id)
+        )).scalars().one()
+    assert restored_rule.id != rule.id
+    assert (await rulebooks_svc.list_rule_moments([restored_rule.id]))[restored_rule.id] == [
+        "work.finish", "reply.report",
+    ]

tests/test_moments.py

@@ -132,3 +132,20 @@ def test_both_doors_read_one_catalog():
     assert routes.moments_svc is moments
     assert tool.actions_svc is moment_actions
     assert routes.moment_actions_svc is moment_actions
+
+
+def test_a_rules_moments_are_normalised_and_deduplicated():
+    assert moments.require_moments([" Work.Finish", "reply.report", "work.finish"]) == [
+        "work.finish", "reply.report",
+    ]
+    assert moments.require_moments("work.finish") == ["work.finish"]
+    assert moments.require_moments([]) == []
+
+
+def test_absent_moments_mean_leave_them_alone():
+    assert moments.require_moments(None) is None
+
+
+def test_one_unknown_moment_refuses_the_whole_set():
+    with pytest.raises(ValueError, match="work.finished"):
+        moments.require_moments(["work.finish", "work.finished"])

tests/test_rule_moment_doors.py

@@ -0,0 +1,132 @@
+"""Every door that writes a rule can mount it on moments (milestone 458 step 3).
+
+THE SHAPE, NOT ONE FUNCTION. test_system_tagging_door_parity records how a
+capability goes missing: whichever door nobody exercised for a kind never
+grows the parameter. So this file asserts the whole table — every rule and
+preference write, on both doors, takes `moments` — and that each one checks
+the names BEFORE its create or update. A refusal after the write would leave
+a rule created without the mounts it asked for, and the caller reading the
+error would reasonably believe nothing happened.
+"""
+from __future__ import annotations
+
+import ast
+import pathlib
+from unittest.mock import AsyncMock, patch
+
+import pytest
+
+from tests.helpers import fake_rule
+from tests.helpers import plain_rule_detail as _plain_detail
+
+ROOT = pathlib.Path(__file__).resolve().parents[1] / "src" / "scribe"
+
+# (tool, the service write it must validate before)
+MCP_DOORS = [
+    ("create_rule", "create_rule"),
+    ("create_project_rule", "create_project_rule"),
+    ("update_rule", "update_rule"),
+    ("create_preference", "create_rule"),
+    ("update_preference", "update_rule"),
+]
+# (route handler, the service write it must validate before)
+REST_DOORS = [
+    ("create_rule", "create_rule"),
+    ("update_rule", "update_rule"),
+    ("create_project_rule", "create_project_rule"),
+]
+
+
+def _fn(path: pathlib.Path, name: str):
+    for node in ast.parse(path.read_text()).body:
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == name:
+            return node
+    raise AssertionError(f"{path.name} has no {name}")
+
+
+def _first_line_calling(fn, attr: str) -> int | None:
+    lines = [
+        n.lineno for n in ast.walk(fn)
+        if isinstance(n, ast.Call) and (
+            (isinstance(n.func, ast.Attribute) and n.func.attr == attr)
+            or (isinstance(n.func, ast.Name) and n.func.id == attr)
+        )
+    ]
+    return min(lines) if lines else None
+
+
+@pytest.mark.parametrize("tool,write", MCP_DOORS)
+def test_every_mcp_rule_door_takes_moments_and_checks_them_first(tool, write):
+    fn = _fn(ROOT / "mcp" / "tools" / "rulebooks.py", tool)
+    params = {a.arg for a in fn.args.args + fn.args.kwonlyargs}
+    assert "moments" in params, f"{tool} cannot mount a rule"
+    check = _first_line_calling(fn, "require_moments")
+    wrote = _first_line_calling(fn, write)
+    assert check is not None and wrote is not None, (tool, check, wrote)
+    assert check < wrote, f"{tool} validates moments after it has already written"
+    detail = [n for n in ast.walk(fn) if isinstance(n, ast.Call)
+              and isinstance(n.func, ast.Attribute) and n.func.attr == "rule_detail"]
+    assert any(any(isinstance(a, ast.Name) and a.id == "moments" for a in c.args)
+               for c in detail), f"{tool} never hands its moments to rule_detail"
+
+
+@pytest.mark.parametrize("handler,write", REST_DOORS)
+def test_every_rest_rule_door_takes_moments_and_checks_them_first(handler, write):
+    fn = _fn(ROOT / "routes" / "rulebooks.py", handler)
+    check = _first_line_calling(fn, "_moments_or_refusal")
+    wrote = _first_line_calling(fn, write)
+    assert check is not None and wrote is not None, (handler, check, wrote)
+    assert check < wrote, f"{handler} validates moments after it has already written"
+
+
+def test_the_guard_can_fail():
+    """A door with no check at all must be reported, not skipped (rule 167)."""
+    fn = ast.parse("async def f():\n    await svc.create_rule()\n").body[0]
+    assert _first_line_calling(fn, "require_moments") is None
+
+
+async def test_an_unknown_moment_is_refused_before_anything_is_written():
+    create = AsyncMock(return_value=fake_rule(id=5))
+    with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.create_rule", create), \
+         patch("scribe.mcp.tools.rulebooks.dedup_svc.find_duplicate_rule", AsyncMock()), \
+         _plain_detail(), patch("scribe.mcp.tools.rulebooks.current_user_id", lambda: 7):
+        from scribe.mcp.tools.rulebooks import create_rule
+        with pytest.raises(ValueError, match="unknown moment"):
+            await create_rule(
+                topic_id=10, title="t", statement="s",
+                when_to_apply="when the moment this fixture stands in for arises",
+                moments=["work.finished"],
+            )
+    create.assert_not_awaited()
+
+
+async def test_the_door_hands_rule_detail_the_normalised_moments():
+    seen = {}
+
+    async def detail(_uid, rule, _system_ids=None, moments=None):
+        seen["moments"] = moments
+        return rule.to_dict()
+
+    with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule",
+               AsyncMock(return_value=fake_rule(id=5))), \
+         patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail", detail), \
+         patch("scribe.mcp.tools.rulebooks.current_user_id", lambda: 7):
+        from scribe.mcp.tools.rulebooks import update_rule
+        await update_rule(rule_id=5, moments=[" Work.Finish", "reply.report", "work.finish"])
+    assert seen["moments"] == ["work.finish", "reply.report"]
+
+
+async def test_omitting_moments_leaves_the_mounts_alone():
+    seen = {}
+
+    async def detail(_uid, rule, _system_ids=None, moments="unset"):
+        seen["moments"] = moments
+        return rule.to_dict()
+
+    with patch("scribe.mcp.tools.rulebooks.rulebooks_svc.update_rule",
+               AsyncMock(return_value=fake_rule(id=5))), \
+         patch("scribe.mcp.tools.rulebooks.rulebooks_svc.rule_detail", detail), \
+         patch("scribe.mcp.tools.rulebooks.current_user_id", lambda: 7):
+        from scribe.mcp.tools.rulebooks import update_rule
+        await update_rule(rule_id=5, title="renamed")
+    assert seen["moments"] is None

tests/test_services_backup.py

@@ -27,7 +27,7 @@ def test_backup_version_is_current():
 
     (Named for the number it asserted until v10, which is exactly the drift a
     name-carrying-a-value invites; it now says what it checks.)"""
-    assert backup.BACKUP_VERSION == 21
+    assert backup.BACKUP_VERSION == 22
 
 
 def _exportable_note(**over):
@@ -452,7 +452,7 @@ def test_the_column_guard_covers_every_table_with_a_row_helper():
     # REAL table names, as _BACKED_UP holds them — not the shorter keys the
     # payload uses for the same sections. Getting this wrong is what the guard
     # caught on its own first run.
-    join_tables = {"rule_systems"}
+    join_tables = {"rule_systems", "rule_moments"}
     covered = set(_column_guard_targets()) | join_tables
     assert set(backup._BACKED_UP) - covered == set()
     # And no stale entries: every declaration must name a real target.