feat(moments): mount the corpus by proposal - a pass and an open-after-moment signal, both stopping at the operator (milestone 458 step 7, #4925)

A rule written before moments existed is mounted on nothing. Step 7 records,
per (rule, moment), whether it belongs there and who said so:

- rule_moment_judgments (migration 0118, backup v23): suggested / confirmed /
  rejected, from a pass, the signal, or an edit. Moment "" is "no moment fits".
- The pass: rules_to_mount lists unjudged rules; propose_rule_moments records
  suggestions that mount nothing; rule_moment_proposals and
  judge_rule_moments put them to the operator. A confirm mounts, a reject is
  kept so the pair is never proposed again. Same service behind REST and a
  "Waiting on you" panel in Settings > Moments.
- Edits are judgments: set_rule_moments, the one mount write path, confirms
  what was added and rejects what was removed in the same transaction.
- The signal: scribe_moment.sh keeps a per-session acts ledger; when a rule
  is opened, scribe_record_opened.sh sends the last three minutes of it to
  /api/plugin/rule-opened. The acts resolve through the install's mappings;
  work.run and work.change are not evidence. Counted per distinct session
  with lesson_rules' evidence model, and once due the open returns one line
  asking the reader to offer the mount.
- scribe_session_end.sh removes the session's scribe-moment files.

Plugin 2026.10.05.2003.

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

tests/test_integration_rule_moment_judgments.py

@@ -0,0 +1,193 @@
+"""Real-Postgres tests for moment proposals and judgments (milestone 458 step 7).
+
+What the step promises, against the real tables:
+- the pass reads only rules nobody has answered for, and any answer — a
+  proposal, a mount, "no moment fits" — takes a rule off that list;
+- a proposal mounts nothing; a confirm mounts, a reject is kept, and a judged
+  pair is never proposed again by either source;
+- an edit that takes a moment off a rule is a rejection the signal respects;
+- the signal counts SESSIONS, not opens, and asks once the bar is crossed;
+- only the owner can propose, judge, or feed the signal.
+"""
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.rule_moment_judgment import RuleMomentJudgment
+from scribe.models.rulebook import Rulebook
+from scribe.models.user import User
+from scribe.services import rule_moment_judgments as judgments_svc
+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 = "moment_judgments_owner"
+STRANGER_USERNAME = "moment_judgments_stranger"
+
+
+async def _purge_books(username: str) -> None:
+    """At SETUP, for the reason test_integration_rule_moments gives."""
+    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):
+        await _purge_books(name)
+    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, "Judgment fixtures")
+    topic = await rulebooks_svc.create_topic(book.id, uid, "verification")
+    done = 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="about to tell the operator a piece of work is finished",
+    )
+    style = await rulebooks_svc.create_rule(
+        topic.id, uid, "Dates are written day first",
+        "Every date shown to a person is written day, month, year.",
+        when_to_apply="formatting a date for display",
+    )
+    return {"uid": uid, "sid": sid, "done": done, "style": style}
+
+
+async def _row(rule_id: int, moment: str) -> RuleMomentJudgment | None:
+    async with async_session() as s:
+        return (await s.execute(
+            select(RuleMomentJudgment).where(
+                RuleMomentJudgment.rule_id == rule_id, RuleMomentJudgment.moment == moment,
+            )
+        )).scalar_one_or_none()
+
+
+async def _unjudged_ids(uid: int) -> set[int]:
+    return {r["id"] for r in (await judgments_svc.unjudged_rules(uid, limit=100))["rules"]}
+
+
+async def test_any_answer_takes_a_rule_off_the_pass(world):
+    uid, done, style = world["uid"], world["done"], world["style"]
+    assert {done.id, style.id} <= await _unjudged_ids(uid)
+
+    out = await judgments_svc.propose(uid, [
+        {"rule_id": done.id, "moments": ["work.finish", "reply.report"], "why": "about finishing"},
+        {"rule_id": style.id, "none": "about how a date looks, whenever one is shown"},
+    ])
+    assert out["proposed"] == 2 and out["no_moment"] == 1 and not out["refused"]
+    assert not ({done.id, style.id} & await _unjudged_ids(uid))
+    # A proposal mounted nothing.
+    assert (await rulebooks_svc.list_rule_moments([done.id])).get(done.id) is None
+
+
+async def test_a_confirm_mounts_and_a_reject_is_never_proposed_again(world):
+    uid, done = world["uid"], world["done"]
+    await judgments_svc.propose(uid, [
+        {"rule_id": done.id, "moments": ["work.finish", "work.plan"], "why": "w"},
+    ])
+    pending = await judgments_svc.pending(uid, rule_id=done.id)
+    assert [p["moment"] for p in pending["rules"][0]["proposals"]] == ["work.finish", "work.plan"]
+
+    out = await judgments_svc.judge(uid, [
+        {"rule_id": done.id, "moment": "work.finish", "verdict": "confirm", "note": "yes"},
+        {"rule_id": done.id, "moment": "work.plan", "verdict": "reject", "note": "too early"},
+    ])
+    assert not out["refused"]
+    assert (await rulebooks_svc.list_rule_moments([done.id]))[done.id] == ["work.finish"]
+    confirmed = await _row(done.id, "work.finish")
+    assert (confirmed.state, confirmed.source, confirmed.note) == ("confirmed", "pass", "yes")
+    assert (await _row(done.id, "work.plan")).state == "rejected"
+    assert (await judgments_svc.pending(uid))["total"] == 0
+
+    again = await judgments_svc.propose(uid, [
+        {"rule_id": done.id, "moments": ["work.plan", "work.finish"], "why": "w"},
+    ])
+    assert again["proposed"] == 0
+    assert {s["moment"] for s in again["skipped"]} == {"work.plan", "work.finish"}
+
+
+async def test_an_edit_that_unmounts_is_a_rejection_the_signal_respects(world):
+    uid, done = world["uid"], world["done"]
+    await rulebooks_svc.set_rule_moments(done.id, uid, ["work.deliver"])
+    await rulebooks_svc.set_rule_moments(done.id, uid, [])
+    row = await _row(done.id, "work.deliver")
+    assert (row.state, row.source) == ("rejected", "edit")
+
+    for session in ("a", "b", "c", "d"):
+        assert await judgments_svc.co_occurred(
+            uid, done.id, ["work.deliver"], situation=session,
+        ) == ""
+    assert (await _row(done.id, "work.deliver")).evidence is None
+
+
+async def test_the_signal_counts_sessions_and_asks_once_the_bar_is_crossed(world):
+    uid, done = world["uid"], world["done"]
+    # Ubiquitous moments are not evidence; a repeat in one session counts once.
+    assert await judgments_svc.co_occurred(uid, done.id, ["work.run", "work.change"],
+                                           situation="s1") == ""
+    assert await _row(done.id, "work.run") is None
+    for situation in ("s1", "s1", "s2"):
+        assert await judgments_svc.co_occurred(uid, done.id, ["work.verify"],
+                                               situation=situation) == ""
+    line = await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation="s3")
+    assert f"#{done.id}" in line and "`work.verify`" in line and "3 distinct sessions" in line
+    assert "judge_rule_moments" in line
+    row = await _row(done.id, "work.verify")
+    assert (row.state, row.source) == ("suggested", "signal")
+    # Asked once; the cooldown keeps the next open quiet.
+    assert await judgments_svc.co_occurred(uid, done.id, ["work.verify"], situation="s4") == ""
+    # And it waits for the operator like any proposal.
+    pending = await judgments_svc.pending(uid, rule_id=done.id)
+    [proposal] = pending["rules"][0]["proposals"]
+    assert proposal["source"] == "signal" and proposal["evidence"]["situations"] == 4
+
+
+async def test_a_mount_overturns_no_moment_fits(world):
+    uid, style = world["uid"], world["style"]
+    await judgments_svc.propose(uid, [{"rule_id": style.id, "none": "about how a date looks"}])
+    assert (await _row(style.id, "")).state == "confirmed"
+    await judgments_svc.judge(uid, [
+        {"rule_id": style.id, "moment": "reply.report", "verdict": "confirm", "note": "dates in reports"},
+    ])
+    assert (await _row(style.id, "")).state == "rejected"
+
+
+async def test_only_the_owner_can_propose_judge_or_feed_the_signal(world):
+    sid, done = world["sid"], world["done"]
+    out = await judgments_svc.propose(sid, [{"rule_id": done.id, "moments": ["work.finish"]}])
+    assert out["proposed"] == 0 and out["refused"]
+    judged = await judgments_svc.judge(sid, [
+        {"rule_id": done.id, "moment": "work.finish", "verdict": "confirm"},
+    ])
+    assert judged["refused"] and not judged["judged"]
+    for situation in ("a", "b", "c"):
+        assert await judgments_svc.co_occurred(sid, done.id, ["work.verify"],
+                                               situation=situation) == ""
+    assert await _row(done.id, "work.verify") is None
+    assert done.id not in await _unjudged_ids(sid)
+    assert (await judgments_svc.pending(sid))["total"] == 0
+
+
+async def test_the_open_resolves_acts_through_the_installs_mappings(world):
+    """`opened_after` reads the acts as delivery does: a push is a deliver."""
+    uid, done = world["uid"], world["done"]
+    acts = [
+        {"tool_name": "Bash", "tool_input": {"command": "git push origin dev"}},
+        {"tool_name": "Bash", "tool_input": {"command": "ls"}},
+        "not an act",
+    ]
+    out = await judgments_svc.opened_after(uid, done.id, acts, session_id="s1")
+    assert "work.deliver" in out["moments"]
+    assert "work.run" not in out["moments"]
+    assert (await _row(done.id, "work.deliver")).source == "signal"

tests/test_integration_rule_moments.py

@@ -16,6 +16,7 @@ from sqlalchemy import select
 
 from scribe.models import async_session
 from scribe.models.project import Project
+from scribe.models.rule_moment_judgment import RuleMomentJudgment
 from scribe.models.rulebook import Rule, Rulebook, RulebookTopic
 from scribe.models.user import User
 from scribe.services import backup
@@ -141,10 +142,20 @@ async def test_a_backup_carries_the_mounts_to_the_restored_rule(world):
     assert sorted(r["moment"] for r in payload["rule_moments"]) == [
         "reply.report", "work.finish",
     ]
+    # The mount was an edit, so it was also recorded as two judgments (step 7)
+    # — and those travel too, onto the restored rule.
+    payload["rule_moment_judgments"] = [
+        row for row in exported["rule_moment_judgments"] if row["rule_id"] == rule.id
+    ]
+    assert sorted((r["moment"], r["state"], r["source"])
+                  for r in payload["rule_moment_judgments"]) == [
+        ("reply.report", "confirmed", "edit"), ("work.finish", "confirmed", "edit"),
+    ]
     payload["users"][0]["username"] = RESTORED_USERNAME
 
     stats = await backup.restore_full_backup(payload)
     assert stats["rule_moments"] == 2
+    assert stats["rule_moment_judgments"] == 2
 
     async with async_session() as s:
         restored_user = (await s.execute(
@@ -159,6 +170,11 @@ 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 with async_session() as s:
+        judged = (await s.execute(
+            select(RuleMomentJudgment.moment).where(RuleMomentJudgment.rule_id == restored_rule.id)
+        )).scalars().all()
+    assert sorted(judged) == ["reply.report", "work.finish"]
 
 
 async def _purge_projects(username: str) -> None:

tests/test_moments.py

@@ -115,9 +115,15 @@ def test_the_tools_are_registered_and_classified():
 
     mcp = FakeMCP()
     tool.register(mcp)
-    assert mcp.names == ["list_moments", "map_action", "unmap_action"]
-    assert "list_moments" in _READ_ONLY_TOOLS
-    assert {"map_action", "unmap_action"} <= _WRITE_TOOLS
+    assert mcp.names == [
+        "list_moments", "map_action", "unmap_action",
+        "rules_to_mount", "propose_rule_moments", "rule_moment_proposals",
+        "judge_rule_moments",
+    ]
+    assert {"list_moments", "rules_to_mount", "rule_moment_proposals"} <= _READ_ONLY_TOOLS
+    # A confirm mounts a rule, so a read key must not reach it.
+    assert {"map_action", "unmap_action",
+            "propose_rule_moments", "judge_rule_moments"} <= _WRITE_TOOLS
 
 
 def test_both_doors_read_one_catalog():

tests/test_rule_moment_judgments.py

@@ -0,0 +1,60 @@
+"""Moment proposals and judgments — what a mock can see (milestone 458 step 7).
+
+The behaviour against real tables is tests/test_integration_rule_moment_judgments.py.
+These pin the pieces that need no database: that the migration's CHECKs and
+the model agree (rule 36), which moments count as evidence, and that the line
+the signal hands a session names the tool that answers it.
+"""
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+from types import SimpleNamespace
+
+from scribe.models.rule_moment_judgment import JUDGMENT_STATES, NO_MOMENT, SOURCES
+from scribe.services import moments as moments_svc
+from scribe.services import rule_moment_judgments as svc
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _migration():
+    path = ROOT / "alembic" / "versions" / "0118_rule_moment_judgments.py"
+    spec = importlib.util.spec_from_file_location("m0118", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_the_migration_checks_and_the_model_agree():
+    m = _migration()
+    assert tuple(m._STATES) == tuple(JUDGMENT_STATES)
+    assert tuple(m._SOURCES) == tuple(SOURCES)
+
+
+def test_verdicts_name_real_states():
+    assert set(svc.VERDICTS.values()) <= set(JUDGMENT_STATES)
+
+
+def test_no_moment_is_not_a_mountable_name():
+    """"" is the "none fits" answer; if it ever became mountable, a pass's
+    "no moment" would read as a mount on nothing."""
+    assert not moments_svc.is_moment(NO_MOMENT)
+
+
+def test_the_ubiquitous_moments_are_not_evidence():
+    assert svc.signal_moments(["work.run", "work.verify", "work.change", "work.verify",
+                               "not.a.moment", "skill.brainstorming"]) == [
+        "work.verify", "skill.brainstorming",
+    ]
+    # Every skipped name is a real moment — a typo here would skip nothing.
+    assert all(moments_svc.is_moment(m) for m in svc.SIGNAL_SKIP)
+
+
+def test_the_proposal_line_names_the_tool_that_answers_it():
+    rule = SimpleNamespace(id=11, title="Definition of done", kind="rule")
+    line = svc._proposal_line(rule, "work.verify", {"situations": ["a", "b", "c"]})
+    assert line.startswith("> Rule #11")
+    assert "`work.verify`" in line and "3 distinct sessions" in line
+    assert '"rule_id": 11' in line and '"moment": "work.verify"' in line
+    assert "judge_rule_moments" in line and '"reject"' in line

tests/test_rule_moment_signal_hooks.py

@@ -0,0 +1,168 @@
+"""The plugin half of the rule↔moment signal (milestone 458 step 7), run for real.
+
+`scribe_moment.sh` writes every call to an acts ledger; `scribe_record_opened.sh`
+sends the last few minutes of it with the rule a session opened; the server
+answers with a proposal line when one is due. These drive both scripts through
+bash with a stand-in `curl` on PATH that records what it was sent, so what is
+pinned is what the scripts actually do — the window, the body's shape, the
+reduction of a large event, the line coming back out — not what their source
+looks like. `scribe_session_end.sh` removing the session's files is pinned the
+same way.
+"""
+from __future__ import annotations
+
+import inspect
+import json
+import os
+import subprocess
+import time
+from pathlib import Path
+
+import pytest
+
+from scribe.routes import plugin as plugin_routes
+from scribe.services import rule_moment_judgments as judgments_svc
+from tests.helpers import need_tools
+
+ROOT = Path(__file__).resolve().parents[1]
+HOOKS = ROOT / "plugin" / "hooks"
+
+FAKE_CURL = """#!/usr/bin/env bash
+data=""
+for a in "$@"; do
+  if [ "$a" = "@-" ]; then data=$(cat); fi
+done
+printf '%s\\n' "$*" >> "$FAKE_CURL_LOG"
+if [ -n "$data" ]; then printf '%s' "$data" > "$FAKE_CURL_BODY"; fi
+if [ -n "${FAKE_CURL_REPLY:-}" ]; then printf '%s' "$FAKE_CURL_REPLY"; else printf '{}'; fi
+"""
+
+
+@pytest.fixture
+def sandbox(tmp_path):
+    need_tools("bash", "awk", "paste", "tail")
+    bin_dir = tmp_path / "bin"
+    bin_dir.mkdir()
+    curl = bin_dir / "curl"
+    curl.write_text(FAKE_CURL)
+    curl.chmod(0o755)
+    env = {
+        "PATH": f"{bin_dir}{os.pathsep}{os.environ['PATH']}",
+        "HOME": str(tmp_path), "TMPDIR": str(tmp_path),
+        "SCRIBE_URL": "http://scribe.invalid", "SCRIBE_TOKEN": "t",
+        "FAKE_CURL_LOG": str(tmp_path / "curl.log"),
+        "FAKE_CURL_BODY": str(tmp_path / "curl.body"),
+        "CLAUDE_PROJECT_DIR": str(tmp_path),
+    }
+    return tmp_path, env
+
+
+def _run(script: str, event: dict, env: dict, **extra) -> subprocess.CompletedProcess:
+    out = subprocess.run(
+        ["bash", str(HOOKS / script)], input=json.dumps(event),
+        capture_output=True, text=True, env={**env, **extra}, timeout=30,
+    )
+    assert out.returncode == 0, out.stderr
+    return out
+
+
+def _acts(tmp: Path, sid: str = "s1") -> list[tuple[int, dict]]:
+    path = tmp / "scribe-moment" / f"{sid}.acts"
+    rows = []
+    for line in path.read_text().splitlines():
+        stamp, _, act = line.partition("\t")
+        rows.append((int(stamp), json.loads(act)))
+    return rows
+
+
+def test_every_call_is_written_to_the_acts_ledger_scribes_own_included(sandbox):
+    tmp, env = sandbox
+    push = {"session_id": "s1", "tool_name": "Bash",
+            "tool_input": {"command": "git push origin dev"}}
+    close = {"session_id": "s1", "tool_name": "mcp__plugin_scribe_scribe__update_task",
+             "tool_input": {"task_id": 4, "status": "done"}}
+    _run("scribe_moment.sh", push, env, FAKE_CURL_REPLY='{"tools":[]}')
+    _run("scribe_moment.sh", close, env)
+    acts = _acts(tmp)
+    assert [a for _, a in acts] == [push, close]
+    assert all(abs(stamp - time.time()) < 60 for stamp, _ in acts)
+
+
+def test_the_open_itself_is_not_an_act(sandbox):
+    tmp, env = sandbox
+    _run("scribe_moment.sh", {"session_id": "s1", "tool_name": "Bash",
+                              "tool_input": {"command": "ls"}}, env,
+         FAKE_CURL_REPLY='{"tools":[]}')
+    _run("scribe_moment.sh", {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
+                              "tool_input": {"rule_id": 11}}, env)
+    assert [a["tool_name"] for _, a in _acts(tmp)] == ["Bash"]
+
+
+def test_a_large_event_is_kept_as_its_tool_name(sandbox):
+    tmp, env = sandbox
+    big = {"session_id": "s1", "tool_name": "Write",
+           "tool_input": {"file_path": "x", "content": "y" * 40000}}
+    _run("scribe_moment.sh", big, env, FAKE_CURL_REPLY='{"tools":[]}')
+    [(_, act)] = _acts(tmp)
+    assert act == {"tool_name": "Write", "tool_input": {}}
+
+
+def test_an_open_sends_the_window_and_prints_the_answer(sandbox):
+    tmp, env = sandbox
+    ledger = tmp / "scribe-moment"
+    ledger.mkdir()
+    now = int(time.time())
+    old = {"tool_name": "Bash", "tool_input": {"command": "make old"}}
+    recent = {"tool_name": "Bash", "tool_input": {"command": "git push"}}
+    (ledger / "s1.acts").write_text(
+        f"{now - 1000}\t{json.dumps(old)}\n{now - 5}\t{json.dumps(recent)}\n"
+    )
+    line = "> Rule #11 has been opened just after `work.deliver` in 3 distinct sessions"
+    out = _run("scribe_record_opened.sh",
+               {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
+                "tool_input": {"rule_id": 11}}, env,
+               FAKE_CURL_REPLY=json.dumps({"moments": ["work.deliver"], "context": line}))
+
+    body = json.loads((tmp / "curl.body").read_text())
+    assert body == {"rule_id": 11, "session_id": "s1", "acts": [recent]}
+    assert "/api/plugin/rule-opened" in (tmp / "curl.log").read_text()
+    printed = json.loads(out.stdout)["hookSpecificOutput"]
+    assert printed["hookEventName"] == "PostToolUse"
+    assert printed["additionalContext"] == line
+    # The open is still recorded in its own ledger, as before.
+    assert (tmp / "scribe-priorart" / "s1.opened.ids").read_text().startswith("11\t")
+
+
+def test_an_open_with_nothing_in_the_window_sends_nothing(sandbox):
+    tmp, env = sandbox
+    ledger = tmp / "scribe-moment"
+    ledger.mkdir()
+    (ledger / "s1.acts").write_text(
+        f"{int(time.time()) - 1000}\t{json.dumps({'tool_name': 'Bash', 'tool_input': {}})}\n"
+    )
+    out = _run("scribe_record_opened.sh",
+               {"session_id": "s1", "tool_name": "mcp__scribe__get_rule",
+                "tool_input": {"rule_id": 11}}, env)
+    assert out.stdout == ""
+    assert not (tmp / "curl.log").exists()
+
+
+def test_session_end_removes_the_sessions_moment_files_and_no_others(sandbox):
+    tmp, env = sandbox
+    ledger = tmp / "scribe-moment"
+    ledger.mkdir()
+    for name in ("s1.acts", "s1.tools", "s2.acts"):
+        (ledger / name).write_text("x")
+    _run("scribe_session_end.sh", {"session_id": "s1", "reason": "clear"}, env)
+    assert sorted(p.name for p in ledger.iterdir()) == ["s2.acts"]
+
+
+def test_the_hook_and_the_route_agree_on_the_body():
+    """Rule 33 across the shell/Python seam: a renamed field fails silently."""
+    src = (HOOKS / "scribe_record_opened.sh").read_text()
+    route = inspect.getsource(plugin_routes.rule_opened)
+    assert "/api/plugin/rule-opened" in src
+    for field in ("rule_id", "session_id", "acts"):
+        assert f'"{field}"' in src and f'"{field}"' in route
+    assert "'.context'" in src
+    assert '"context"' in inspect.getsource(judgments_svc.opened_after)

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 == 22
+    assert backup.BACKUP_VERSION == 23
 
 
 def _exportable_note(**over):
@@ -132,6 +132,7 @@ def _column_guard_targets():
     from scribe.models.canonical_system import CanonicalSystem
     from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
     from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink
+    from scribe.models.rule_moment_judgment import RuleMomentJudgment
     from scribe.models.design_system import DesignSystem, DesignToken
     from scribe.models.milestone import Milestone
     from scribe.models.note import Note
@@ -172,6 +173,7 @@ def _column_guard_targets():
         "rule_relations": (RuleRelation, backup._rule_relation_rows),
         "lesson_rule_links": (LessonRuleLink, backup._lesson_rule_link_rows),
         "lesson_no_rule": (LessonNoRule, backup._lesson_no_rule_rows),
+        "rule_moment_judgments": (RuleMomentJudgment, backup._rule_moment_judgment_rows),
         "note_usage_events": (NoteUsageEvent, backup._usage_event_rows),
         "rule_usage_events": (RuleUsageEvent, backup._rule_usage_event_rows),
         "system_usage_events": (SystemUsageEvent, backup._system_usage_event_rows),
@@ -551,7 +553,9 @@ async def test_export_full_backup_contains_every_declared_section():
                 # v19: the lessons judged to fall under no rule.
                 "lesson_no_rule",
                 # v20: whether an area's rulings were read once shown.
-                "system_usage_events"):
+                "system_usage_events",
+                # v23: proposals and judgments about a rule's moments.
+                "rule_moment_judgments"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []